Files
metrics/lib/streamaggr/streamaggr_test.go
Max Kotliar be3eddf425 lib/prompb: Merge prompbmarshal logic into prompb
The prompb and prompbmarshal share exactly the same models and provide
marshal and unmarshale capabilities for them. This creates duplication
(changes in one model has to be made in another, case with metadata) and
confusion where for example you compare same looking models but golang
says they are not the same (because of the type).

This commit merge prompbmarshal logic into prompb so the rest of the
code is aligned on prompb models.

Moves samplesPool and labelsPool to WriteRequestUnmarshaller.
Make WriteRequest struct clean from unmarshal logic.

The benchmark shows no significant changes:

$benchstat prompbmarshal.bench prompb2.bench
goos: darwin
goarch: arm64
pkg: github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb
cpu: Apple M1 Pro
                                 │ prompbmarshal.bench │           prompb2.bench            │
                                 │       sec/op        │   sec/op     vs base               │
WriteRequestUnmarshalProtobuf-10           189.2µ ± 5%   190.8µ ± 8%       ~ (p=0.579 n=10)
WriteRequestMarshalProtobuf-10             145.3µ ± 7%   143.6µ ± 2%       ~ (p=0.143 n=10)
geomean                                    165.8µ        165.5µ       -0.14%

                                 │ prompbmarshal.bench │            prompb2.bench            │
                                 │         B/s         │     B/s       vs base               │
WriteRequestUnmarshalProtobuf-10          50.42Mi ± 5%   49.99Mi ± 8%       ~ (p=0.593 n=10)
WriteRequestMarshalProtobuf-10            65.64Mi ± 7%   66.39Mi ± 2%       ~ (p=0.143 n=10)
geomean                                   57.53Mi        57.61Mi       +0.14%

                                 │ prompbmarshal.bench │            prompb2.bench             │
                                 │        B/op         │     B/op       vs base               │
WriteRequestUnmarshalProtobuf-10         27.70Ki ±  4%   26.90Ki ±  7%       ~ (p=0.190 n=10)
WriteRequestMarshalProtobuf-10           3.267Ki ± 12%   3.273Ki ± 12%       ~ (p=0.971 n=10)
geomean                                  9.514Ki         9.383Ki        -1.38%

                                 │ prompbmarshal.bench │            prompb2.bench            │
                                 │      allocs/op      │ allocs/op   vs base                 │
WriteRequestUnmarshalProtobuf-10          0.000 ± 0%     0.000 ± 0%       ~ (p=1.000 n=10) ¹
WriteRequestMarshalProtobuf-10            0.000 ± 0%     0.000 ± 0%       ~ (p=1.000 n=10) ¹
geomean                                              ²               +0.00%                ²
¹ all samples are equal
² summaries must be >0 to compute geomean
2025-07-31 01:04:11 +03:00

280 lines
4.8 KiB
Go

package streamaggr
import (
"fmt"
"sort"
"strings"
"testing"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompb"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
)
func TestAggregatorsFailure(t *testing.T) {
f := func(config string) {
t.Helper()
pushFunc := func(_ []prompb.TimeSeries) {
panic(fmt.Errorf("pushFunc shouldn't be called"))
}
a, err := LoadFromData([]byte(config), pushFunc, nil, "some_alias")
if err == nil {
t.Fatalf("expecting non-nil error")
}
if a != nil {
t.Fatalf("expecting nil a")
}
}
// Invalid config
f(`foobar`)
// Unknown option
f(`
- interval: 1m
outputs: [total]
foobar: baz
`)
// missing interval
f(`
- outputs: [total]
`)
// missing outputs
f(`
- interval: 1m
`)
// Bad interval
f(`
- interval: 1foo
outputs: [total]
`)
// Invalid output
f(`
- interval: 1m
outputs: [foobar]
`)
// Negative interval
f(`
- outputs: [total]
interval: -5m
`)
// Too small interval
f(`
- outputs: [total]
interval: 10ms
`)
// bad dedup_interval
f(`
- interval: 1m
dedup_interval: 1foo
outputs: ["quantiles"]
`)
// interval isn't multiple of dedup_interval
f(`
- interval: 1m
dedup_interval: 35s
outputs: ["quantiles"]
`)
// dedup_interval is bigger than dedup_interval
f(`
- interval: 1m
dedup_interval: 1h
outputs: ["quantiles"]
`)
// bad staleness_interval
f(`
- interval: 1m
staleness_interval: 1foo
outputs: ["quantiles"]
`)
// staleness_interval should be > interval
f(`
- interval: 1m
staleness_interval: 30s
outputs: ["quantiles"]
`)
// staleness_interval should be multiple of interval
f(`
- interval: 1m
staleness_interval: 100s
outputs: ["quantiles"]
`)
// keep_metric_names is set for multiple inputs
f(`
- interval: 1m
keep_metric_names: true
outputs: ["total", "increase"]
`)
// keep_metric_names is set for unsupported input
f(`
- interval: 1m
keep_metric_names: true
outputs: ["histogram_bucket"]
`)
// Invalid input_relabel_configs
f(`
- interval: 1m
outputs: [total]
input_relabel_configs:
- foo: bar
`)
f(`
- interval: 1m
outputs: [total]
input_relabel_configs:
- action: replace
`)
// Invalid output_relabel_configs
f(`
- interval: 1m
outputs: [total]
output_relabel_configs:
- foo: bar
`)
f(`
- interval: 1m
outputs: [total]
output_relabel_configs:
- action: replace
`)
// Both by and without are non-empty
f(`
- interval: 1m
outputs: [total]
by: [foo]
without: [bar]
`)
// Invalid quantiles()
f(`
- interval: 1m
outputs: ["quantiles("]
`)
f(`
- interval: 1m
outputs: ["quantiles()"]
`)
f(`
- interval: 1m
outputs: ["quantiles(foo)"]
`)
f(`
- interval: 1m
outputs: ["quantiles(-0.5)"]
`)
f(`
- interval: 1m
outputs: ["quantiles(1.5)"]
`)
f(`
- interval: 1m
outputs: [total, total]
`)
// "quantiles(0.5)", "quantiles(0.9)" should be set as "quantiles(0.5, 0.9)"
f(`
- interval: 1m
outputs: ["quantiles(0.5)", "quantiles(0.9)"]
`)
}
func TestAggregatorsEqual(t *testing.T) {
f := func(a, b string, expectedResult bool) {
t.Helper()
pushFunc := func(_ []prompb.TimeSeries) {}
opts := Options{
EnableWindows: true,
}
aa, err := LoadFromData([]byte(a), pushFunc, &opts, "some_alias")
if err != nil {
t.Fatalf("cannot initialize aggregators: %s", err)
}
ab, err := LoadFromData([]byte(b), pushFunc, &opts, "some_alias")
if err != nil {
t.Fatalf("cannot initialize aggregators: %s", err)
}
result := aa.Equal(ab)
if result != expectedResult {
t.Fatalf("unexpected result; got %v; want %v", result, expectedResult)
}
}
f("", "", true)
f(`
- outputs: [total]
interval: 5m
`, ``, false)
f(`
- outputs: [total]
interval: 5m
`, `
- outputs: [total]
interval: 5m
`, true)
f(`
- outputs: [total]
interval: 3m
`, `
- outputs: [total]
interval: 5m
`, false)
f(`
- outputs: [total]
interval: 5m
flush_on_shutdown: true
`, `
- outputs: [total]
interval: 5m
flush_on_shutdown: false
`, false)
f(`
- outputs: [total]
interval: 5m
ignore_first_intervals: 2
`, `
- outputs: [total]
interval: 5m
ignore_first_intervals: 4`, false)
}
func timeSeriessToString(tss []prompb.TimeSeries) string {
a := make([]string, len(tss))
for i, ts := range tss {
a[i] = timeSeriesToString(ts)
}
sort.Strings(a)
return strings.Join(a, "")
}
func timeSeriesToString(ts prompb.TimeSeries) string {
labelsString := promrelabel.LabelsToString(ts.Labels)
if len(ts.Samples) != 1 {
panic(fmt.Errorf("unexpected number of samples for %s: %d; want 1", labelsString, len(ts.Samples)))
}
return fmt.Sprintf("%s %v\n", labelsString, ts.Samples[0].Value)
}
func appendClonedTimeseries(dst, src []prompb.TimeSeries) []prompb.TimeSeries {
for _, ts := range src {
dst = append(dst, prompb.TimeSeries{
Labels: append(ts.Labels[:0:0], ts.Labels...),
Samples: append(ts.Samples[:0:0], ts.Samples...),
})
}
return dst
}