mirror of
https://github.com/VictoriaMetrics/VictoriaMetrics.git
synced 2024-11-21 14:44:00 +00:00
52c13e9515
- Use a byte slice instead of a map for tracking indexes for matching series. This improves performance, since access by slice index is faster than access by map key. - Re-use the byte slice for tracking indexes for matching series. This removes unnecessary memory allocations and improves stream aggregation performance a bit. - Add an ability to return to the previous behvaiour by specifying -remoteWrite.streamAggr.dropInput command-line flag. In this case all the input samples are dropped when stream aggregation is enabled. - Backport the new stream aggregation behaviour from vmagent to single-node VictoriaMetrics when -streamAggr.config option is set. - Improve docs regarding this change at docs/CHANGELOG.md - Document the new behavior at docs/stream-aggregation.md Updates https://github.com/VictoriaMetrics/VictoriaMetrics/issues/4243 Updates https://github.com/VictoriaMetrics/VictoriaMetrics/pull/4575
78 lines
1.7 KiB
Go
78 lines
1.7 KiB
Go
package streamaggr
|
|
|
|
import (
|
|
"fmt"
|
|
"strings"
|
|
"testing"
|
|
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
|
|
)
|
|
|
|
func BenchmarkAggregatorsPushByJobAvg(b *testing.B) {
|
|
for _, output := range []string{
|
|
"total",
|
|
"increase",
|
|
"count_series",
|
|
"count_samples",
|
|
"sum_samples",
|
|
"last",
|
|
"min",
|
|
"max",
|
|
"avg",
|
|
"stddev",
|
|
"stdvar",
|
|
"histogram_bucket",
|
|
"quantiles(0, 0.5, 1)",
|
|
} {
|
|
b.Run(fmt.Sprintf("output=%s", output), func(b *testing.B) {
|
|
benchmarkAggregatorsPush(b, output)
|
|
})
|
|
}
|
|
}
|
|
|
|
func benchmarkAggregatorsPush(b *testing.B, output string) {
|
|
config := fmt.Sprintf(`
|
|
- match: http_requests_total
|
|
interval: 24h
|
|
without: [job]
|
|
outputs: [%q]
|
|
`, output)
|
|
pushCalls := 0
|
|
pushFunc := func(tss []prompbmarshal.TimeSeries) {
|
|
pushCalls++
|
|
if pushCalls > 1 {
|
|
panic(fmt.Errorf("pushFunc is expected to be called exactly once at MustStop"))
|
|
}
|
|
}
|
|
a, err := NewAggregatorsFromData([]byte(config), pushFunc, 0)
|
|
if err != nil {
|
|
b.Fatalf("unexpected error when initializing aggregators: %s", err)
|
|
}
|
|
defer a.MustStop()
|
|
|
|
b.ReportAllocs()
|
|
b.SetBytes(int64(len(benchSeries)))
|
|
b.RunParallel(func(pb *testing.PB) {
|
|
var matchIdxs []byte
|
|
for pb.Next() {
|
|
matchIdxs = a.Push(benchSeries, matchIdxs)
|
|
}
|
|
})
|
|
}
|
|
|
|
func newBenchSeries(seriesCount, samplesPerSeries int) []prompbmarshal.TimeSeries {
|
|
a := make([]string, seriesCount*samplesPerSeries)
|
|
for i := 0; i < samplesPerSeries; i++ {
|
|
for j := 0; j < seriesCount; j++ {
|
|
s := fmt.Sprintf(`http_requests_total{path="/foo/%d",job="foo",instance="bar"} %d`, j, i*10)
|
|
a = append(a, s)
|
|
}
|
|
}
|
|
metrics := strings.Join(a, "\n")
|
|
return mustParsePromMetrics(metrics)
|
|
}
|
|
|
|
const seriesCount = 10000
|
|
const samplesPerSeries = 10
|
|
|
|
var benchSeries = newBenchSeries(seriesCount, samplesPerSeries)
|