Optimizations on prom-encode and loki-write

- Pre-allocate slices/maps to avoid dynamic reallocations
- In loki write, split labels/lines in 1 pass rather than copying
  everything then deleting labels from lines.

Author: jotak

pkg/pipeline/encode/encode_prom.go

@@ -98,7 +98,8 @@ func (e *EncodeProm) ProcessAggHist(m interface{}, labels map[string]string, val
 	return nil
 }
 
-func (e *EncodeProm) GetChacheEntry(entryLabels map[string]string, m interface{}) interface{} {
+func (e *EncodeProm) GetCacheEntry(entryLabels map[string]string, m interface{}) interface{} {
+	// In prom_encode, the metrics cache just contains cleanup callbacks
 	switch mv := m.(type) {
 	case *prometheus.CounterVec:
 		return func() { mv.Delete(entryLabels) }

pkg/pipeline/encode/metrics_common.go

@@ -42,7 +42,7 @@ type MetricsCommonStruct struct {
 	histos           map[string]mInfoStruct
 	aggHistos        map[string]mInfoStruct
 	mCache           *putils.TimedCache
-	mChacheLenMetric prometheus.Gauge
+	mCacheLenMetric  prometheus.Gauge
 	metricsProcessed prometheus.Counter
 	metricsDropped   prometheus.Counter
 	errorsCounter    *prometheus.CounterVec
@@ -51,7 +51,7 @@ type MetricsCommonStruct struct {
 }
 
 type MetricsCommonInterface interface {
-	GetChacheEntry(entryLabels map[string]string, m interface{}) 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
 	ProcessHist(m interface{}, labels map[string]string, value float64) error
@@ -198,13 +198,14 @@ func (m *MetricsCommonStruct) prepareMetric(mci MetricsCommonInterface, flow con
 	}
 
 	labelSets := extractLabels(flow, flatParts, info)
-	var lkms []labelsKeyAndMap
+	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)
-		cacheEntry := mci.GetChacheEntry(lkm.lMap, mv)
-		ok := m.mCache.UpdateCacheEntry(lkm.key, cacheEntry)
+		ok := m.mCache.UpdateCacheEntry(lkm.key, func() interface{} {
+			return mci.GetCacheEntry(lkm.lMap, mv)
+		})
 		if !ok {
 			m.metricsDropped.Inc()
 			return nil, 0
@@ -231,13 +232,14 @@ func (m *MetricsCommonStruct) prepareAggHisto(mci MetricsCommonInterface, flow c
 	}
 
 	labelSets := extractLabels(flow, flatParts, info)
-	var lkms []labelsKeyAndMap
+	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)
-		cacheEntry := mci.GetChacheEntry(lkm.lMap, mc)
-		ok := m.mCache.UpdateCacheEntry(lkm.key, cacheEntry)
+		ok := m.mCache.UpdateCacheEntry(lkm.key, func() interface{} {
+			return mci.GetCacheEntry(lkm.lMap, mc)
+		})
 		if !ok {
 			m.metricsDropped.Inc()
 			return nil, nil
@@ -273,9 +275,10 @@ type labelsKeyAndMap struct {
 
 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 := map[string]string{}
+	m := make(map[string]string, len(l))
 	for _, kv := range l {
 		key.WriteString(kv.value)
 		key.WriteRune('|')
@@ -302,7 +305,7 @@ func extractLabels(flow config.GenericMap, flatParts []config.GenericMap, info *
 }
 
 func newLabelSet(part config.GenericMap, labels []metrics.MappedLabel) labelSet {
-	var ls labelSet
+	ls := make(labelSet, 0, len(labels))
 	for _, t := range labels {
 		label := label{key: t.Target, value: ""}
 		if v, ok := part[t.Source]; ok {
@@ -337,7 +340,7 @@ func NewMetricsCommonStruct(opMetrics *operational.Metrics, maxCacheEntries int,
 	mChacheLenMetric := opMetrics.NewGauge(&mChacheLen, name)
 	m := &MetricsCommonStruct{
 		mCache:           putils.NewTimedCache(maxCacheEntries, mChacheLenMetric),
-		mChacheLenMetric: mChacheLenMetric,
+		mCacheLenMetric:  mChacheLenMetric,
 		metricsProcessed: opMetrics.NewCounter(&metricsProcessed, name),
 		metricsDropped:   opMetrics.NewCounter(&metricsDropped, name),
 		errorsCounter:    opMetrics.NewCounterVec(&encodePromErrors),

pkg/pipeline/encode/opentelemetry/encode_otlpmetrics.go

@@ -90,7 +90,7 @@ func (e *EncodeOtlpMetrics) ProcessAggHist(m interface{}, labels map[string]stri
 	return nil
 }
 
-func (e *EncodeOtlpMetrics) GetChacheEntry(entryLabels map[string]string, _ interface{}) interface{} {
+func (e *EncodeOtlpMetrics) GetCacheEntry(entryLabels map[string]string, _ interface{}) interface{} {
 	return entryLabels
 }
 

pkg/pipeline/extract/aggregate/aggregate.go

@@ -140,7 +140,9 @@ func (aggregate *Aggregate) UpdateByEntry(entry config.GenericMap, normalizedVal
 	} else {
 		groupState = oldEntry.(*GroupState)
 	}
-	aggregate.cache.UpdateCacheEntry(string(normalizedValues), groupState)
+	aggregate.cache.UpdateCacheEntry(string(normalizedValues), func() interface{} {
+		return groupState
+	})
 
 	// update value
 	operationKey := aggregate.definition.OperationKey

pkg/pipeline/transform/transform_network.go

@@ -130,7 +130,7 @@ func (n *Network) Transform(inputEntry config.GenericMap) (config.GenericMap, bo
 					lbl, ok := n.ipLabelCache.GetCacheEntry(strIP)
 					if !ok {
 						lbl = n.applySubnetLabel(strIP)
-						n.ipLabelCache.UpdateCacheEntry(strIP, lbl)
+						n.ipLabelCache.UpdateCacheEntry(strIP, func() interface{} { return lbl })
 					}
 					if lbl != "" {
 						outputEntry[rule.AddSubnetLabel.Output] = lbl

pkg/pipeline/utils/timed_cache.go

@@ -65,9 +65,9 @@ func (tc *TimedCache) GetCacheEntry(key string) (interface{}, bool) {
 
 var uclog = log.WithField("method", "UpdateCacheEntry")
 
-// If cache entry exists, update it and return it; if it does not exist, create it if there is room.
+// If cache entry exists, update its timestamp; if it does not exist, create it if there is room.
 // If we exceed the size of the cache, then do not allocate new entry
-func (tc *TimedCache) UpdateCacheEntry(key string, entry interface{}) bool {
+func (tc *TimedCache) UpdateCacheEntry(key string, entryProvider func() interface{}) bool {
 	nowInSecs := time.Now()
 	tc.mu.Lock()
 	defer tc.mu.Unlock()
@@ -85,7 +85,7 @@ func (tc *TimedCache) UpdateCacheEntry(key string, entry interface{}) bool {
 		cEntry = &cacheEntry{
 			lastUpdatedTime: nowInSecs,
 			key:             key,
-			SourceEntry:     entry,
+			SourceEntry:     entryProvider(),
 		}
 		uclog.Tracef("adding entry: %#v", cEntry)
 		// place at end of list

pkg/pipeline/write/write_loki.go

@@ -56,6 +56,7 @@ type Loki struct {
 	apiConfig      api.WriteLoki
 	timestampScale float64
 	saneLabels     map[string]model.LabelName
+	ignoreList     map[string]any
 	client         emitter
 	timeNow        func() time.Time
 	exitChan       <-chan struct{}
@@ -107,36 +108,51 @@ func buildLokiConfig(c *api.WriteLoki) (loki.Config, error) {
 }
 
 func (l *Loki) ProcessRecord(in config.GenericMap) error {
-	// copy record before process to avoid alteration on parallel stages
-	out := in.Copy()
-	labels := model.LabelSet{}
+	labels, lines := l.splitLabelsLines(in)
 
-	// Add static labels from config
-	for k, v := range l.apiConfig.StaticLabels {
-		labels[k] = v
-	}
-	l.addLabels(in, labels)
-
-	// Remove labels and configured ignore list from record
-	ignoreList := l.apiConfig.IgnoreList
-	ignoreList = append(ignoreList, l.apiConfig.Labels...)
-	for _, label := range ignoreList {
-		delete(out, label)
-	}
-
-	js, err := jsonEncodingConfig.Marshal(out)
+	js, err := jsonEncodingConfig.Marshal(lines)
 	if err != nil {
 		return err
 	}
 
-	timestamp := l.extractTimestamp(out)
+	timestamp := l.extractTimestamp(lines)
 	err = l.client.Handle(labels, timestamp, string(js))
 	if err == nil {
 		l.metrics.recordsWritten.Inc()
 	}
 	return err
 }
 
+func (l *Loki) splitLabelsLines(in config.GenericMap) (model.LabelSet, config.GenericMap) {
+	// Split the input GenericMap into one map for labels and another for lines / payload
+	nLabels := len(l.apiConfig.StaticLabels) + len(l.saneLabels)
+	labels := make(model.LabelSet, nLabels)
+	lines := make(config.GenericMap, len(in))
+
+	// Add static labels from config
+	for k, v := range l.apiConfig.StaticLabels {
+		labels[k] = v
+	}
+
+	for k, v := range in {
+		if _, ignored := l.ignoreList[k]; ignored {
+			continue
+		}
+		if sanitized, isLabel := l.saneLabels[k]; isLabel {
+			lv := model.LabelValue(utils.ConvertToString(v))
+			if !lv.IsValid() {
+				log.WithFields(logrus.Fields{"key": k, "value": v}).Debug("Invalid label value. Ignoring it")
+				continue
+			}
+			labels[sanitized] = lv
+		} else {
+			lines[k] = v
+		}
+	}
+
+	return labels, lines
+}
+
 func (l *Loki) extractTimestamp(record map[string]interface{}) time.Time {
 	if l.apiConfig.TimestampLabel == "" {
 		return l.timeNow()
@@ -163,27 +179,6 @@ func (l *Loki) extractTimestamp(record map[string]interface{}) time.Time {
 	return time.Unix(tsNanos/int64(time.Second), tsNanos%int64(time.Second))
 }
 
-func (l *Loki) addLabels(record config.GenericMap, labels model.LabelSet) {
-	// Add non-static labels from record
-	for _, label := range l.apiConfig.Labels {
-		val, ok := record[label]
-		if !ok {
-			continue
-		}
-		sanitized, ok := l.saneLabels[label]
-		if !ok {
-			continue
-		}
-		lv := model.LabelValue(utils.ConvertToString(val))
-		if !lv.IsValid() {
-			log.WithFields(logrus.Fields{"key": label, "value": val}).
-				Debug("Invalid label value. Ignoring it")
-			continue
-		}
-		labels[sanitized] = lv
-	}
-}
-
 func getFloat64(timestamp interface{}) (ft float64, ok bool) {
 	switch i := timestamp.(type) {
 	case float64:
@@ -255,11 +250,18 @@ func NewWriteLoki(opMetrics *operational.Metrics, params config.StageParam) (*Lo
 		}
 	}
 
+	// Ignore list to map
+	ignoreList := make(map[string]any, len(lokiConfigIn.IgnoreList))
+	for _, label := range lokiConfigIn.IgnoreList {
+		ignoreList[label] = nil
+	}
+
 	l := &Loki{
 		lokiConfig:     lokiConfig,
 		apiConfig:      lokiConfigIn,
 		timestampScale: float64(timestampScale),
 		saneLabels:     saneLabels,
+		ignoreList:     ignoreList,
 		client:         client,
 		timeNow:        time.Now,
 		exitChan:       pUtils.ExitChannel(),

pkg/pipeline/write/write_loki_test.go

@@ -289,7 +289,7 @@ parameters:
 	loki.client = &fe
 
 	require.NoError(t, loki.ProcessRecord(map[string]interface{}{
-		"ba/z": "isBaz", "fo.o": "isFoo", "ba-r": "isBar", "ignored?": "yes!"}))
+		"ba/z": "isBaz", "fo.o": "isFoo", "ba-r": "isBar", "ignored?": "yes!", "md": "md"}))
 
 	fe.AssertCalled(t, "Handle", model.LabelSet{
 		"ba_r": "isBar",
@@ -336,6 +336,7 @@ func buildFlow(t time.Time) config.GenericMap {
 		"bytes":     rand.Intn(100),
 		"packets":   rand.Intn(10),
 		"latency":   rand.Float64(),
+		"toIgnore":  "--",
 	}
 }
 
@@ -359,9 +360,10 @@ func BenchmarkWriteLoki(b *testing.B) {
 		StaticLabels: model.LabelSet{
 			"app": "flp-benchmark",
 		},
-		Labels:         []string{"srcIP", "dstIP"},
+		Labels:         []string{"srcIP", "dstIP", "toIgnore"},
 		TimestampLabel: "timestamp",
 		TimestampScale: "1ms",
+		IgnoreList:     []string{"toIgnore"},
 	}
 
 	loki, err := NewWriteLoki(operational.NewMetrics(&config.MetricsSettings{}), config.StageParam{Write: &config.Write{Loki: &params}})