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

Author: jotak

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()
@@ -201,9 +200,9 @@ func (m *MetricsCommonStruct) prepareMetric(mci MetricsCommonInterface, flow con
 	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)
+		lkm := ls.toKeysAndMap(info)
 		lkms = append(lkms, lkm)
-		ok := m.mCache.UpdateCacheEntry(lkm.key, func() interface{} {
+		ok := m.mCache.UpdateCacheEntry(lkm.values, func() interface{} {
 			return mci.GetCacheEntry(lkm.lMap, mv)
 		})
 		if !ok {
@@ -235,9 +234,9 @@ func (m *MetricsCommonStruct) prepareAggHisto(mci MetricsCommonInterface, flow c
 	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)
+		lkm := ls.toKeysAndMap(info)
 		lkms = append(lkms, lkm)
-		ok := m.mCache.UpdateCacheEntry(lkm.key, func() interface{} {
+		ok := m.mCache.UpdateCacheEntry(lkm.values, func() interface{} {
 			return mci.GetCacheEntry(lkm.lMap, mc)
 		})
 		if !ok {
@@ -269,22 +268,19 @@ type label struct {
 type labelSet []label
 
 type labelsKeyAndMap struct {
-	key  string
-	lMap map[string]string
+	values []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('|')
+func (l labelSet) toKeysAndMap(info *metrics.Preprocessed) labelsKeyAndMap {
+	values := make([]string, 0, len(l)+1)
+	values = append(values, info.Name)
 	m := make(map[string]string, len(l))
 	for _, kv := range l {
-		key.WriteString(kv.value)
-		key.WriteRune('|')
+		values = append(values, kv.value)
 		m[kv.key] = kv.value
 	}
-	return labelsKeyAndMap{key: key.String(), lMap: m}
+	return labelsKeyAndMap{values: values, lMap: m}
 }
 
 // extractLabels takes the flow and a single metric definition as input.

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

pkg/pipeline/utils/fnv.go

@@ -0,0 +1,44 @@
+// Copyright 2018 The Prometheus Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Borrowed from https://github.com/prometheus/client_golang/blob/7c924b7c82b5a61e1e7e2be165bff781700b5365/prometheus/fnv.go
+
+package utils
+
+// Inline and byte-free variant of hash/fnv's fnv64a.
+
+const (
+	offset64 = 14695981039346656037
+	prime64  = 1099511628211
+)
+
+// hashNew initializies a new fnv64a hash value.
+func hashNew() uint64 {
+	return offset64
+}
+
+// hashAdd adds a string to a fnv64a hash value, returning the updated hash.
+func hashAdd(h uint64, s string) uint64 {
+	for i := 0; i < len(s); i++ {
+		h ^= uint64(s[i])
+		h *= prime64
+	}
+	return h
+}
+
+// hashAddByte adds a byte to a fnv64a hash value, returning the updated hash.
+func hashAddByte(h uint64, b byte) uint64 {
+	h ^= uint64(b)
+	h *= prime64
+	return h
+}