NETOBSERV-2285: opt: use uint64 hash instead of string for cache key (#992)

* Optimization: use uint64 hash instead of string for cache  key

In timed_cache, use uint64 hash (fnv 64a) instead of string builder
Uses the same implementation as in prom go client

* Bump agent, with improved pb decode

(MAC address to string was improved)

* Optimization: reduce allocations in metrics encode

Previously, labels per metric were computed in 2 passes:
1. to extract key/values
2. copying the map in prom-client suitable format and extracting the values to build a key

Step 1 resulted in an ephemere map allocation
Those two steps are now merged, avoiding intermediate allocations

Author: jotak

go.mod

@@ -18,7 +18,7 @@ require (
 	github.com/mitchellh/mapstructure v1.5.0
 	github.com/netobserv/gopipes v0.3.0
 	github.com/netobserv/loki-client-go v0.0.0-20250425113517-526b43e51847
-	github.com/netobserv/netobserv-ebpf-agent v1.9.0-crc0.0.20250528064221-6f34e5b85d2c
+	github.com/netobserv/netobserv-ebpf-agent v1.9.0-crc0.0.20250610144135-d64c5d99f2da
 	github.com/netsampler/goflow2 v1.3.7
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.22.0

go.sum

@@ -258,8 +258,8 @@ github.com/netobserv/gopipes v0.3.0 h1:IYmPnnAVCdSK7VmHmpFhrVBOEm45qpgbZmJz1sSW+
 github.com/netobserv/gopipes v0.3.0/go.mod h1:N7/Gz05EOF0CQQSKWsv3eof22Cj2PB08Pbttw98YFYU=
 github.com/netobserv/loki-client-go v0.0.0-20250425113517-526b43e51847 h1:hjzhVZSSKIOmAzHbGUV4JhVIPkgKs/UtrWDx6JSVKMw=
 github.com/netobserv/loki-client-go v0.0.0-20250425113517-526b43e51847/go.mod h1:Zb/jtD3Lnu88Poo+jnhTASzxYnvncmHOoZaT93xQjJ8=
-github.com/netobserv/netobserv-ebpf-agent v1.9.0-crc0.0.20250528064221-6f34e5b85d2c h1:HhObTHfakq06NAPkljppnT1UT9MV2o5obH2dPD0jxmk=
-github.com/netobserv/netobserv-ebpf-agent v1.9.0-crc0.0.20250528064221-6f34e5b85d2c/go.mod h1:sqvri+RCx8QefV2g+pbJktbNuO3ebB1TE1Zl7M2qqDo=
+github.com/netobserv/netobserv-ebpf-agent v1.9.0-crc0.0.20250610144135-d64c5d99f2da h1:fahbLlD/5BidF+7rcvydhro3Tu0pY81OUGOkxmaY07A=
+github.com/netobserv/netobserv-ebpf-agent v1.9.0-crc0.0.20250610144135-d64c5d99f2da/go.mod h1:IfxvtBeSfhJaCO/7ie3R7mna+j8sAer2vqtbwaBTlzA=
 github.com/netsampler/goflow2 v1.3.7 h1:XZaTy8kkMnGXpJ9hS3KbO1McyrFTpVNhVFEx9rNhMmc=
 github.com/netsampler/goflow2 v1.3.7/go.mod h1:4UZsVGVAs//iMCptUHn3WNScztJeUhZH7kDW2+/vDdQ=
 github.com/nxadm/tail v1.4.4/go.mod h1:kenIhsEOeOJmVchQTgglprH7qJGnHDVpk1VPCcaMI8A=

pkg/pipeline/encode/encode_prom.go

@@ -66,7 +66,7 @@ func (e *EncodeProm) ProcessCounter(m interface{}, labels map[string]string, val
 	return nil
 }
 
-func (e *EncodeProm) ProcessGauge(m interface{}, labels map[string]string, value float64, _ string) error {
+func (e *EncodeProm) ProcessGauge(m interface{}, _ string, labels map[string]string, value float64, _ []string) error {
 	gauge := m.(*prometheus.GaugeVec)
 	mm, err := gauge.GetMetricWith(labels)
 	if err != nil {

pkg/pipeline/encode/encode_prom_test.go

@@ -819,14 +819,6 @@ func buildFlow() config.GenericMap {
 	}
 }
 
-func thousandsFlows() []config.GenericMap {
-	flows := make([]config.GenericMap, 1000)
-	for i := 0; i < 1000; i++ {
-		flows[i] = buildFlow()
-	}
-	return flows
-}
-
 func BenchmarkPromEncode(b *testing.B) {
 	var expiryTimeDuration api.Duration
 	expiryTimeDuration.Duration = time.Duration(60 * time.Second)
@@ -869,9 +861,8 @@ func BenchmarkPromEncode(b *testing.B) {
 	require.NoError(b, err)
 
 	for i := 0; i < b.N; i++ {
-		for _, metric := range thousandsFlows() {
-			prom.Encode(metric)
-		}
+		m := buildFlow()
+		prom.Encode(m)
 	}
 }
 

pkg/pipeline/encode/metrics_common.go

@@ -18,7 +18,6 @@
 package encode
 
 import (
-	"strings"
 	"time"
 
 	"github.com/netobserv/flowlogs-pipeline/pkg/api"
@@ -53,7 +52,7 @@ type MetricsCommonStruct struct {
 type MetricsCommonInterface interface {
 	GetCacheEntry(entryLabels map[string]string, m interface{}) interface{}
 	ProcessCounter(m interface{}, labels map[string]string, value float64) error
-	ProcessGauge(m interface{}, labels map[string]string, value float64, key string) error
+	ProcessGauge(m interface{}, name string, labels map[string]string, value float64, lvs []string) error
 	ProcessHist(m interface{}, labels map[string]string, value float64) error
 	ProcessAggHist(m interface{}, labels map[string]string, value []float64) error
 }
@@ -132,7 +131,7 @@ func (m *MetricsCommonStruct) MetricCommonEncode(mci MetricsCommonInterface, met
 			continue
 		}
 		for _, labels := range labelSets {
-			err := mci.ProcessGauge(mInfo.genericMetric, labels.lMap, value, labels.key)
+			err := mci.ProcessGauge(mInfo.genericMetric, mInfo.info.Name, labels.lMap, value, labels.values)
 			if err != nil {
 				log.Errorf("labels registering error on %s: %v", mInfo.info.Name, err)
 				m.errorsCounter.WithLabelValues("LabelsRegisteringError", mInfo.info.Name, "").Inc()
@@ -198,20 +197,17 @@ func (m *MetricsCommonStruct) prepareMetric(mci MetricsCommonInterface, flow con
 	}
 
 	labelSets := extractLabels(flow, flatParts, info)
-	lkms := make([]labelsKeyAndMap, 0, len(labelSets))
 	for _, ls := range labelSets {
 		// Update entry for expiry mechanism (the entry itself is its own cleanup function)
-		lkm := ls.toKeyAndMap(info)
-		lkms = append(lkms, lkm)
-		ok := m.mCache.UpdateCacheEntry(lkm.key, func() interface{} {
-			return mci.GetCacheEntry(lkm.lMap, mv)
+		ok := m.mCache.UpdateCacheEntry(ls.values, func() interface{} {
+			return mci.GetCacheEntry(ls.lMap, mv)
 		})
 		if !ok {
 			m.metricsDropped.Inc()
 			return nil, 0
 		}
 	}
-	return lkms, floatVal
+	return labelSets, floatVal
 }
 
 func (m *MetricsCommonStruct) prepareAggHisto(mci MetricsCommonInterface, flow config.GenericMap, info *metrics.Preprocessed, mc interface{}) ([]labelsKeyAndMap, []float64) {
@@ -232,20 +228,17 @@ func (m *MetricsCommonStruct) prepareAggHisto(mci MetricsCommonInterface, flow c
 	}
 
 	labelSets := extractLabels(flow, flatParts, info)
-	lkms := make([]labelsKeyAndMap, 0, len(labelSets))
 	for _, ls := range labelSets {
 		// Update entry for expiry mechanism (the entry itself is its own cleanup function)
-		lkm := ls.toKeyAndMap(info)
-		lkms = append(lkms, lkm)
-		ok := m.mCache.UpdateCacheEntry(lkm.key, func() interface{} {
-			return mci.GetCacheEntry(lkm.lMap, mc)
+		ok := m.mCache.UpdateCacheEntry(ls.values, func() interface{} {
+			return mci.GetCacheEntry(ls.lMap, mc)
 		})
 		if !ok {
 			m.metricsDropped.Inc()
 			return nil, nil
 		}
 	}
-	return lkms, values
+	return labelSets, values
 }
 
 func (m *MetricsCommonStruct) extractGenericValue(flow config.GenericMap, info *metrics.Preprocessed) interface{} {
@@ -261,59 +254,44 @@ func (m *MetricsCommonStruct) extractGenericValue(flow config.GenericMap, info *
 	return val
 }
 
-type label struct {
-	key   string
-	value string
-}
-
-type labelSet []label
-
 type labelsKeyAndMap struct {
-	key  string
-	lMap map[string]string
-}
-
-func (l labelSet) toKeyAndMap(info *metrics.Preprocessed) labelsKeyAndMap {
-	key := strings.Builder{}
-	key.Grow(256) // pre-allocate a decent buffer
-	key.WriteString(info.Name)
-	key.WriteRune('|')
-	m := make(map[string]string, len(l))
-	for _, kv := range l {
-		key.WriteString(kv.value)
-		key.WriteRune('|')
-		m[kv.key] = kv.value
-	}
-	return labelsKeyAndMap{key: key.String(), lMap: m}
+	values []string
+	lMap   map[string]string
 }
 
 // extractLabels takes the flow and a single metric definition as input.
 // It returns the flat labels maps (label names and values).
 // Most of the time it will return a single map; it may return several of them when the parsed flow fields are lists (e.g. "interfaces").
-func extractLabels(flow config.GenericMap, flatParts []config.GenericMap, info *metrics.Preprocessed) []labelSet {
-	common := newLabelSet(flow, info.MappedLabels)
+func extractLabels(flow config.GenericMap, flatParts []config.GenericMap, info *metrics.Preprocessed) []labelsKeyAndMap {
+	common := newLabelKeyAndMap(info.Name, flow, info.MappedLabels)
 	if len(flatParts) == 0 {
-		return []labelSet{common}
+		return []labelsKeyAndMap{common}
 	}
-	var all []labelSet
+	all := make([]labelsKeyAndMap, 0, len(flatParts))
 	for _, fp := range flatParts {
-		ls := newLabelSet(fp, info.FlattenedLabels)
-		ls = append(ls, common...)
+		ls := newLabelKeyAndMap(info.Name, fp, info.FlattenedLabels)
+		ls.values = append(ls.values, common.values...)
+		for k, v := range common.lMap {
+			ls.lMap[k] = v
+		}
 		all = append(all, ls)
 	}
 	return all
 }
 
-func newLabelSet(part config.GenericMap, labels []metrics.MappedLabel) labelSet {
-	ls := make(labelSet, 0, len(labels))
+func newLabelKeyAndMap(name string, part config.GenericMap, labels []metrics.MappedLabel) labelsKeyAndMap {
+	values := make([]string, 0, len(labels)+1)
+	values = append(values, name)
+	m := make(map[string]string, len(labels))
 	for _, t := range labels {
-		label := label{key: t.Target, value: ""}
+		value := ""
 		if v, ok := part[t.Source]; ok {
-			label.value = utils.ConvertToString(v)
+			value = utils.ConvertToString(v)
 		}
-		ls = append(ls, label)
+		values = append(values, value)
+		m[t.Target] = value
 	}
-	return ls
+	return labelsKeyAndMap{values: values, lMap: m}
 }
 
 func (m *MetricsCommonStruct) cleanupExpiredEntriesLoop(callback putils.CacheCallback) {

pkg/pipeline/encode/opentelemetry/encode_otlpmetrics.go

@@ -19,6 +19,7 @@ package opentelemetry
 
 import (
 	"context"
+	"strings"
 	"time"
 
 	"github.com/netobserv/flowlogs-pipeline/pkg/api"
@@ -64,10 +65,22 @@ func (e *EncodeOtlpMetrics) ProcessCounter(m interface{}, labels map[string]stri
 	return nil
 }
 
-func (e *EncodeOtlpMetrics) ProcessGauge(m interface{}, labels map[string]string, value float64, key string) error {
+func createKey(name string, keys []string) string {
+	key := strings.Builder{}
+	key.WriteString(name)
+	key.WriteRune('|')
+	for _, k := range keys {
+		key.WriteString(k)
+		key.WriteRune('|')
+	}
+	return key.String()
+}
+
+func (e *EncodeOtlpMetrics) ProcessGauge(m interface{}, name string, labels map[string]string, value float64, lvs []string) error {
 	obs := m.(Float64Gauge)
 	// set attributes using the labels
 	attributes := obtainAttributesFromLabels(labels)
+	key := createKey(name, lvs)
 	obs.Set(key, value, attributes)
 	return nil
 }

pkg/pipeline/extract/aggregate/aggregate.go

@@ -43,7 +43,7 @@ const (
 )
 
 type Labels map[string]string
-type NormalizedValues string
+type NormalizedValues []string
 
 type Aggregate struct {
 	definition *api.AggregateDefinition
@@ -78,7 +78,7 @@ func (aggregate *Aggregate) LabelsFromEntry(entry config.GenericMap) (Labels, bo
 }
 
 func (labels Labels) getNormalizedValues() NormalizedValues {
-	var normalizedAsString string
+	var normalized NormalizedValues
 
 	keys := make([]string, 0, len(labels))
 	for k := range labels {
@@ -87,20 +87,16 @@ func (labels Labels) getNormalizedValues() NormalizedValues {
 	sort.Strings(keys)
 
 	for _, k := range keys {
-		normalizedAsString += labels[k] + ","
+		normalized = append(normalized, labels[k])
 	}
 
-	if len(normalizedAsString) > 0 {
-		normalizedAsString = normalizedAsString[:len(normalizedAsString)-1]
-	}
-
-	return NormalizedValues(normalizedAsString)
+	return normalized
 }
 
 func (aggregate *Aggregate) filterEntry(entry config.GenericMap) (NormalizedValues, Labels, error) {
 	labels, allLabelsFound := aggregate.LabelsFromEntry(entry)
 	if !allLabelsFound {
-		return "", nil, fmt.Errorf("missing keys in entry")
+		return nil, nil, fmt.Errorf("missing keys in entry")
 	}
 
 	normalizedValues := labels.getNormalizedValues()
@@ -128,7 +124,7 @@ func (aggregate *Aggregate) UpdateByEntry(entry config.GenericMap, normalizedVal
 	defer aggregate.mutex.Unlock()
 
 	var groupState *GroupState
-	oldEntry, ok := aggregate.cache.GetCacheEntry(string(normalizedValues))
+	oldEntry, ok := aggregate.cache.GetCacheEntry(normalizedValues)
 	if !ok {
 		groupState = &GroupState{normalizedValues: normalizedValues, labels: labels}
 		initVal := getInitValue(string(aggregate.definition.OperationType))
@@ -140,7 +136,7 @@ func (aggregate *Aggregate) UpdateByEntry(entry config.GenericMap, normalizedVal
 	} else {
 		groupState = oldEntry.(*GroupState)
 	}
-	aggregate.cache.UpdateCacheEntry(string(normalizedValues), func() interface{} {
+	aggregate.cache.UpdateCacheEntry(normalizedValues, func() interface{} {
 		return groupState
 	})
 
@@ -212,20 +208,21 @@ func (aggregate *Aggregate) GetMetrics() []config.GenericMap {
 	var metrics []config.GenericMap
 
 	// iterate over the items in the cache
-	aggregate.cache.Iterate(func(_ string, value interface{}) {
+	aggregate.cache.Iterate(func(_ uint64, value interface{}) {
 		group := value.(*GroupState)
+		nv := strings.Join(group.normalizedValues, ",")
 		newEntry := config.GenericMap{
 			"name":              aggregate.definition.Name,
 			"operation_type":    aggregate.definition.OperationType,
 			"operation_key":     aggregate.definition.OperationKey,
 			"by":                strings.Join(aggregate.definition.GroupByKeys, ","),
-			"aggregate":         string(group.normalizedValues),
+			"aggregate":         nv,
 			"total_value":       group.totalValue,
 			"total_count":       group.totalCount,
 			"recent_raw_values": group.recentRawValues,
 			"recent_op_value":   group.recentOpValue,
 			"recent_count":      group.recentCount,
-			strings.Join(aggregate.definition.GroupByKeys, "_"): string(group.normalizedValues),
+			strings.Join(aggregate.definition.GroupByKeys, "_"): nv,
 		}
 		// add the items in aggregate.definition.GroupByKeys individually to the entry
 		for _, key := range aggregate.definition.GroupByKeys {

pkg/pipeline/extract/aggregate/aggregate_test.go

@@ -62,7 +62,7 @@ func getMockLabels(reverseOrder bool) Labels {
 }
 
 func Test_getNormalizedValues(t *testing.T) {
-	expectedLabels := NormalizedValues("20.0.0.2,10.0.0.1")
+	expectedLabels := NormalizedValues{"20.0.0.2", "10.0.0.1"}
 
 	labels := getMockLabels(false)
 
@@ -103,7 +103,7 @@ func Test_FilterEntry(t *testing.T) {
 	normalizedLabels, labels, err := aggregate.filterEntry(entry)
 	require.Equal(t, err, nil)
 	require.Equal(t, Labels{"srcIP": "10.0.0.1", "dstIP": "20.0.0.2"}, labels)
-	require.Equal(t, NormalizedValues("20.0.0.2,10.0.0.1"), normalizedLabels)
+	require.Equal(t, NormalizedValues{"20.0.0.2", "10.0.0.1"}, normalizedLabels)
 
 	entry = test.GetIngestMockEntry(true)
 
@@ -125,7 +125,7 @@ func Test_Evaluate(t *testing.T) {
 
 	require.Equal(t, nil, err)
 	require.Equal(t, 1, aggregate.cache.GetCacheLen())
-	cacheEntry, found := aggregate.cache.GetCacheEntry(string(normalizedValues))
+	cacheEntry, found := aggregate.cache.GetCacheEntry(normalizedValues)
 	gState := cacheEntry.(*GroupState)
 	require.Equal(t, true, found)
 	require.Equal(t, 2, gState.totalCount)

pkg/pipeline/transform/transform_network.go

@@ -127,10 +127,11 @@ func (n *Network) Transform(inputEntry config.GenericMap) (config.GenericMap, bo
 			}
 			if anyIP, ok := outputEntry[rule.AddSubnetLabel.Input]; ok {
 				if strIP, ok := anyIP.(string); ok {
-					lbl, ok := n.ipLabelCache.GetCacheEntry(strIP)
+					keys := []string{strIP}
+					lbl, ok := n.ipLabelCache.GetCacheEntry(keys)
 					if !ok {
 						lbl = n.applySubnetLabel(strIP)
-						n.ipLabelCache.UpdateCacheEntry(strIP, func() interface{} { return lbl })
+						n.ipLabelCache.UpdateCacheEntry(keys, func() interface{} { return lbl })
 					}
 					if lbl != "" {
 						outputEntry[rule.AddSubnetLabel.Output] = lbl