Kiu/VictoriaMetrics
Kiu/VictoriaMetrics
1
0
Fork 0
mirror of https://github.com/VictoriaMetrics/VictoriaMetrics.git synced 2024-12-15 08:23:34 +01:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

app/vmselect/promql: move common code from aggrFuncOutliersK and newAggrFuncRangeTopK into getRangeTopKTimeseries

Browse Source
This commit is contained in:
Aliaksandr Valialkin 2020-05-19 16:10:52 +03:00
parent 37068064dd
commit 7d46dd452a
1 changed files with 36 additions and 55 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

51
app/vmselect/promql/aggr.go
View File

@ -484,11 +484,6 @@ func newAggrFuncTopK(isReverse bool) aggrFunc {
} }
} }
type tsWithValue struct {
ts *timeseries
value float64
}
func newAggrFuncRangeTopK(f func(values []float64) float64, isReverse bool) aggrFunc { func newAggrFuncRangeTopK(f func(values []float64) float64, isReverse bool) aggrFunc {
return func(afa *aggrFuncArg) ([]*timeseries, error) { return func(afa *aggrFuncArg) ([]*timeseries, error) {
args := afa.args args := afa.args
@ -500,6 +495,17 @@ func newAggrFuncRangeTopK(f func(values []float64) float64, isReverse bool) aggr
return nil, err return nil, err
} }
afe := func(tss []*timeseries) []*timeseries { afe := func(tss []*timeseries) []*timeseries {
return getRangeTopKTimeseries(tss, ks, f, isReverse)
}
return aggrFuncExt(afe, args[1], &afa.ae.Modifier, afa.ae.Limit, true)
}
}
func getRangeTopKTimeseries(tss []*timeseries, ks []float64, f func(values []float64) float64, isReverse bool) []*timeseries {
type tsWithValue struct {
ts *timeseries
value float64
}
maxs := make([]tsWithValue, len(tss)) maxs := make([]tsWithValue, len(tss))
for i, ts := range tss { for i, ts := range tss {
value := f(ts.Values) value := f(ts.Values)
@ -523,9 +529,6 @@ func newAggrFuncRangeTopK(f func(values []float64) float64, isReverse bool) aggr
fillNaNsAtIdx(i, k, tss) fillNaNsAtIdx(i, k, tss)
} }
return removeNaNs(tss) return removeNaNs(tss)
}
return aggrFuncExt(afe, args[1], &afa.ae.Modifier, afa.ae.Limit, true)
}
} }
func fillNaNsAtIdx(idx int, k float64, tss []*timeseries) { func fillNaNsAtIdx(idx int, k float64, tss []*timeseries) {
@ -623,38 +626,16 @@ func aggrFuncOutliersK(afa *aggrFuncArg) ([]*timeseries, error) {
} }
histogram.PutFast(h) histogram.PutFast(h)
// Calculate variation-like value for each tss. // Return topK time series with the highest variance from median.
type variation struct { f := func(values []float64) float64 {
sum2 float64
ts *timeseries
}
variations := make([]variation, len(tss))
for i, ts := range tss {
sum2 := float64(0) sum2 := float64(0)
for n, v := range ts.Values { for n, v := range values {
d := v - medians[n] d := v - medians[n]
sum2 += d * d sum2 += d * d
} }
variations[i] = variation{ return sum2
sum2: sum2,
ts: ts,
} }
} return getRangeTopKTimeseries(tss, ks, f, false)
// Sort variations by sum2.
sort.Slice(variations, func(i, j int) bool {
a, b := variations[i], variations[j]
return lessWithNaNs(a.sum2, b.sum2)
})
// Return only up to k time series with the highest variation.
for i := range variations {
tss[i] = variations[i].ts
}
for i, k := range ks {
fillNaNsAtIdx(i, k, tss)
}
return removeNaNs(tss)
} }
return aggrFuncExt(afe, args[1], &afa.ae.Modifier, afa.ae.Limit, true) return aggrFuncExt(afe, args[1], &afa.ae.Modifier, afa.ae.Limit, true)
} }