VictoriaMetrics/lib/promutils/labelscompressor_test.go

120 lines
2.4 KiB
Go
Raw Permalink Normal View History

lib/streamaggr: huge pile of changes - Reduce memory usage by up to 5x when de-duplicating samples across big number of time series. - Reduce memory usage by up to 5x when aggregating across big number of output time series. - Add lib/promutils.LabelsCompressor, which is going to be used by other VictoriaMetrics components for reducing memory usage for marshaled []prompbmarshal.Label. - Add `dedup_interval` option at aggregation config, which allows setting individual deduplication intervals per each aggregation. - Add `keep_metric_names` option at aggregation config, which allows keeping the original metric names in the output samples. - Add `unique_samples` output, which counts the number of unique sample values. - Add `increase_prometheus` and `total_prometheus` outputs, which ignore the first sample per each newly encountered time series. - Use 64-bit hashes instead of marshaled labels as map keys when calculating `count_series` output. This makes obsolete https://github.com/VictoriaMetrics/VictoriaMetrics/pull/5579 - Expose various metrics, which may help debugging stream aggregation: - vm_streamaggr_dedup_state_size_bytes - the size of data structures responsible for deduplication - vm_streamaggr_dedup_state_items_count - the number of items in the deduplication data structures - vm_streamaggr_labels_compressor_size_bytes - the size of labels compressor data structures - vm_streamaggr_labels_compressor_items_count - the number of entries in the labels compressor - vm_streamaggr_flush_duration_seconds - a histogram, which shows the duration of stream aggregation flushes - vm_streamaggr_dedup_flush_duration_seconds - a histogram, which shows the duration of deduplication flushes - vm_streamaggr_flush_timeouts_total - counter for timed out stream aggregation flushes, which took longer than the configured interval - vm_streamaggr_dedup_flush_timeouts_total - counter for timed out deduplication flushes, which took longer than the configured dedup_interval - Actualize docs/stream-aggregation.md The memory usage reduction increases CPU usage during stream aggregation by up to 30%. This commit is based on https://github.com/VictoriaMetrics/VictoriaMetrics/pull/5850 Updates https://github.com/VictoriaMetrics/VictoriaMetrics/issues/5898
2024-03-02 00:42:26 +00:00
package promutils
import (
"fmt"
"sync"
"testing"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
)
func TestLabelsCompressorSerial(t *testing.T) {
var lc LabelsCompressor
f := func(labels []prompbmarshal.Label) {
t.Helper()
sExpected := labelsToString(labels)
data := lc.Compress(nil, labels)
labelsResult := lc.Decompress(nil, data)
sResult := labelsToString(labelsResult)
if sExpected != sResult {
t.Fatalf("unexpected result; got %s; want %s", sResult, sExpected)
}
if len(labels) > 0 {
if n := lc.SizeBytes(); n == 0 {
t.Fatalf("Unexpected zero SizeBytes()")
}
if n := lc.ItemsCount(); n == 0 {
t.Fatalf("Unexpected zero ItemsCount()")
}
}
}
// empty labels
f(nil)
f([]prompbmarshal.Label{})
// non-empty labels
f([]prompbmarshal.Label{
{
Name: "instance",
Value: "12345.4342.342.3",
},
{
Name: "job",
Value: "kube-pod-12323",
},
})
f([]prompbmarshal.Label{
{
Name: "instance",
Value: "12345.4342.342.3",
},
{
Name: "job",
Value: "kube-pod-12323",
},
{
Name: "pod",
Value: "foo-bar-baz",
},
})
}
func TestLabelsCompressorConcurrent(t *testing.T) {
const concurrency = 5
var lc LabelsCompressor
var wg sync.WaitGroup
for i := 0; i < concurrency; i++ {
wg.Add(1)
go func() {
defer wg.Done()
series := newTestSeries(100, 20)
for i, labels := range series {
sExpected := labelsToString(labels)
data := lc.Compress(nil, labels)
labelsResult := lc.Decompress(nil, data)
sResult := labelsToString(labelsResult)
if sExpected != sResult {
panic(fmt.Errorf("unexpected result on iteration %d; got %s; want %s", i, sResult, sExpected))
}
}
}()
}
wg.Wait()
if n := lc.SizeBytes(); n == 0 {
t.Fatalf("Unexpected zero SizeBytes()")
}
if n := lc.ItemsCount(); n == 0 {
t.Fatalf("Unexpected zero ItemsCount()")
}
}
func labelsToString(labels []prompbmarshal.Label) string {
l := Labels{
Labels: labels,
}
return l.String()
}
func newTestSeries(seriesCount, labelsPerSeries int) [][]prompbmarshal.Label {
series := make([][]prompbmarshal.Label, seriesCount)
for i := 0; i < seriesCount; i++ {
labels := make([]prompbmarshal.Label, labelsPerSeries)
for j := 0; j < labelsPerSeries; j++ {
labels[j] = prompbmarshal.Label{
Name: fmt.Sprintf("label_%d", j),
Value: fmt.Sprintf("value_%d_%d", i, j),
}
}
series[i] = labels
}
return series
}