From c66a691593928ed0d6b5276e54ccddc55bce622d Mon Sep 17 00:00:00 2001 From: Aliaksandr Valialkin Date: Tue, 15 Oct 2019 19:12:27 +0300 Subject: [PATCH] app/vmselect/prometheus: add `-search.maxLookback` command-line flag for overriding dynamic calculations for max lookback interval This flag is similar to `-search.lookback-delta` if set. The max lookback interval is determined dynamically from interval between datapoints for each input time series if the flag isn't set. The interval can be overriden on per-query basis by passing `max_lookback=` query arg to `/api/v1/query` and `/api/v1/query_range`. Updates https://github.com/VictoriaMetrics/VictoriaMetrics/issues/209 --- README.md | 2 +- app/vmselect/prometheus/prometheus.go | 42 ++++++++++++++++++------- app/vmselect/promql/eval.go | 12 +++++-- app/vmselect/promql/rollup.go | 6 ++++ app/vmselect/promql/rollup_test.go | 45 +++++++++++++++++++++++++++ 5 files changed, 92 insertions(+), 15 deletions(-) diff --git a/README.md b/README.md index 09a73bff1..43fa0a3a2 100644 --- a/README.md +++ b/README.md @@ -512,7 +512,7 @@ at `http://:8428/federate?match[]= *maxQueryLen { return fmt.Errorf(`too long query; got %d bytes; mustn't exceed %d bytes`, len(query), *maxQueryLen) @@ -503,10 +512,11 @@ func QueryHandler(w http.ResponseWriter, r *http.Request) error { } ec := promql.EvalConfig{ - Start: start, - End: start, - Step: step, - Deadline: deadline, + Start: start, + End: start, + Step: step, + Deadline: deadline, + LookbackDelta: lookbackDelta, } result, err := promql.Exec(&ec, query, true) if err != nil { @@ -546,6 +556,10 @@ func QueryRangeHandler(w http.ResponseWriter, r *http.Request) error { } deadline := getDeadline(r) mayCache := !getBool(r, "nocache") + lookbackDelta, err := getMaxLookback(r) + if err != nil { + return err + } // Validate input args. if len(query) > *maxQueryLen { @@ -562,11 +576,12 @@ func QueryRangeHandler(w http.ResponseWriter, r *http.Request) error { } ec := promql.EvalConfig{ - Start: start, - End: end, - Step: step, - Deadline: deadline, - MayCache: mayCache, + Start: start, + End: end, + Step: step, + Deadline: deadline, + MayCache: mayCache, + LookbackDelta: lookbackDelta, } result, err := promql.Exec(&ec, query, false) if err != nil { @@ -726,6 +741,11 @@ func getDuration(r *http.Request, argKey string, defaultValue int64) (int64, err const maxDurationMsecs = 100 * 365 * 24 * 3600 * 1000 +func getMaxLookback(r *http.Request) (int64, error) { + d := int64(*maxLookback / time.Millisecond) + return getDuration(r, "max_lookback", d) +} + func getDeadline(r *http.Request) netstorage.Deadline { d, err := getDuration(r, "timeout", 0) if err != nil { diff --git a/app/vmselect/promql/eval.go b/app/vmselect/promql/eval.go index 3b8df6531..c4094e511 100644 --- a/app/vmselect/promql/eval.go +++ b/app/vmselect/promql/eval.go @@ -70,6 +70,9 @@ type EvalConfig struct { MayCache bool + // LookbackDelta is analog to `-query.lookback-delta` from Prometheus. + LookbackDelta int64 + timestamps []int64 timestampsOnce sync.Once } @@ -82,6 +85,7 @@ func newEvalConfig(src *EvalConfig) *EvalConfig { ec.Step = src.Step ec.Deadline = src.Deadline ec.MayCache = src.MayCache + ec.LookbackDelta = src.LookbackDelta // do not copy src.timestamps - they must be generated again. return &ec @@ -465,7 +469,7 @@ func evalRollupFuncWithSubquery(ec *EvalConfig, name string, rf rollupFunc, re * } sharedTimestamps := getTimestamps(ec.Start, ec.End, ec.Step) - preFunc, rcs := getRollupConfigs(name, rf, ec.Start, ec.End, ec.Step, window, sharedTimestamps) + preFunc, rcs := getRollupConfigs(name, rf, ec.Start, ec.End, ec.Step, window, ec.LookbackDelta, sharedTimestamps) tss := make([]*timeseries, 0, len(tssSQ)*len(rcs)) var tssLock sync.Mutex removeMetricGroup := !rollupFuncsKeepMetricGroup[name] @@ -586,7 +590,7 @@ func evalRollupFuncWithMetricExpr(ec *EvalConfig, name string, rf rollupFunc, me return tss, nil } sharedTimestamps := getTimestamps(start, ec.End, ec.Step) - preFunc, rcs := getRollupConfigs(name, rf, start, ec.End, ec.Step, window, sharedTimestamps) + preFunc, rcs := getRollupConfigs(name, rf, start, ec.End, ec.Step, window, ec.LookbackDelta, sharedTimestamps) // Verify timeseries fit available memory after the rollup. // Take into account points from tssCached. @@ -689,7 +693,8 @@ func doRollupForTimeseries(rc *rollupConfig, tsDst *timeseries, mnSrc *storage.M tsDst.denyReuse = true } -func getRollupConfigs(name string, rf rollupFunc, start, end, step, window int64, sharedTimestamps []int64) (func(values []float64, timestamps []int64), []*rollupConfig) { +func getRollupConfigs(name string, rf rollupFunc, start, end, step, window int64, lookbackDelta int64, sharedTimestamps []int64) ( + func(values []float64, timestamps []int64), []*rollupConfig) { preFunc := func(values []float64, timestamps []int64) {} if rollupFuncsRemoveCounterResets[name] { preFunc = func(values []float64, timestamps []int64) { @@ -705,6 +710,7 @@ func getRollupConfigs(name string, rf rollupFunc, start, end, step, window int64 Step: step, Window: window, MayAdjustWindow: rollupFuncsMayAdjustWindow[name], + LookbackDelta: lookbackDelta, Timestamps: sharedTimestamps, } } diff --git a/app/vmselect/promql/rollup.go b/app/vmselect/promql/rollup.go index 5e0571668..0754ed4a8 100644 --- a/app/vmselect/promql/rollup.go +++ b/app/vmselect/promql/rollup.go @@ -149,6 +149,9 @@ type rollupConfig struct { MayAdjustWindow bool Timestamps []int64 + + // LoookbackDelta is the analog to `-query.lookback-delta` from Prometheus world. + LookbackDelta int64 } var ( @@ -186,6 +189,9 @@ func (rc *rollupConfig) Do(dstValues []float64, values []float64, timestamps []i dstValues = decimal.ExtendFloat64sCapacity(dstValues, len(rc.Timestamps)) maxPrevInterval := getMaxPrevInterval(timestamps) + if rc.LookbackDelta > 0 && maxPrevInterval > rc.LookbackDelta { + maxPrevInterval = rc.LookbackDelta + } window := rc.Window if window <= 0 { window = rc.Step diff --git a/app/vmselect/promql/rollup_test.go b/app/vmselect/promql/rollup_test.go index 8b99a635d..d16cec875 100644 --- a/app/vmselect/promql/rollup_test.go +++ b/app/vmselect/promql/rollup_test.go @@ -488,6 +488,51 @@ func TestRollupWindowPartialPoints(t *testing.T) { }) } +func TestRollupFuncsLookbackDelta(t *testing.T) { + t.Run("1", func(t *testing.T) { + rc := rollupConfig{ + Func: rollupFirst, + Start: 80, + End: 140, + Step: 10, + LookbackDelta: 1, + } + rc.Timestamps = getTimestamps(rc.Start, rc.End, rc.Step) + values := rc.Do(nil, testValues, testTimestamps) + valuesExpected := []float64{99, 12, 44, nan, 32, 34, nan} + timestampsExpected := []int64{80, 90, 100, 110, 120, 130, 140} + testRowsEqual(t, values, rc.Timestamps, valuesExpected, timestampsExpected) + }) + t.Run("7", func(t *testing.T) { + rc := rollupConfig{ + Func: rollupFirst, + Start: 80, + End: 140, + Step: 10, + LookbackDelta: 7, + } + rc.Timestamps = getTimestamps(rc.Start, rc.End, rc.Step) + values := rc.Do(nil, testValues, testTimestamps) + valuesExpected := []float64{99, 12, 44, 44, 32, 34, nan} + timestampsExpected := []int64{80, 90, 100, 110, 120, 130, 140} + testRowsEqual(t, values, rc.Timestamps, valuesExpected, timestampsExpected) + }) + t.Run("0", func(t *testing.T) { + rc := rollupConfig{ + Func: rollupFirst, + Start: 80, + End: 140, + Step: 10, + LookbackDelta: 0, + } + rc.Timestamps = getTimestamps(rc.Start, rc.End, rc.Step) + values := rc.Do(nil, testValues, testTimestamps) + valuesExpected := []float64{34, 12, 12, 44, 44, 34, nan} + timestampsExpected := []int64{80, 90, 100, 110, 120, 130, 140} + testRowsEqual(t, values, rc.Timestamps, valuesExpected, timestampsExpected) + }) +} + func TestRollupFuncsNoWindow(t *testing.T) { t.Run("first", func(t *testing.T) { rc := rollupConfig{