Repository: netobserv/flowlogs-pipeline
Pull request #319 (2022-10-03 10:23:15)

NETOBSERV-578: removing arrays from pipeline interfaces
This PR removes the batching logic in the Kafka ingester and processes flows 1 by 1 on each pipeline stage.

While the CPU usage doesn't seem to noticeably decrease:

<img width="1525" alt="image" src="https://user-images.githubusercontent.com/939550/193555066-675227cb-bd2a-48e2-bcc8-5d4e55aba011.png">

It greatly improves memory consumption:

<img width="1535" alt="image" src="https://user-images.githubusercontent.com/939550/193554945-4e63a7c5-d10b-436d-ae46-20404e875bda.png">

(as well as simplifies some parts of the code)

github-actions[bot]: New image: ["quay.io/netobserv/flowlogs-pipeline:1e0449c"]. It will expire after two weeks.
mariomac: It seems this PR is not possible unless we reimplement the connection tracking stage.

@ronensc do you have an estimation about how difficult would be to reimplement the function:
```go
func (ct *conntrackImpl) Extract(flowLogs []config.GenericMap) []config.GenericMap
```
To:
```go
func (ct *conntrackImpl) Extract(flowLogs config.GenericMap) config.GenericMap
```
Currently it seems it's not able to detect endConnections
KalmanMeth: Not only the connection tracking, but also the aggregation and timebased-topk would need changes. In each of these we save some state and do some computation after batches of flow logs. If now the flow logs come one a time, we have to check that our implementation only runs every so often and not after each call to Extract().
mariomac: Ok. I'll close this PR at the moment. I noticed similar improvements in memory just by setting the PullQueueCapacity to 50-100.
mariomac: @jotak @ronensc @KalmanMeth @eranra reopening this PR as I found a compromise solution. I added an intermediate batcher for these parts of the API that really need to work in batches, like the connection tracking and the timebased-topk.

This would improve overall performance and memory but specially in those FLP configurations that do not use extractors. At the same time, this gives us more room for a later reimplementation of extractors to avoid batching.

Patch:
diff --git a/go.sum b/go.sum
index 1ac91993..4e6fc745 100644
--- a/go.sum
+++ b/go.sum
@@ -699,8 +699,6 @@ github.com/nats-io/nkeys v0.1.3/go.mod h1:xpnFELMwJABBLVhffcfd1MZx6VsNRFpEugbxzi
 github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
 github.com/netobserv/gopipes v0.1.1 h1:f8zJsvnMgRFRa2B+1siwRtW0Y4dqeBROmkcI/HgT1gE=
 github.com/netobserv/gopipes v0.1.1/go.mod h1:eGoHZW1ON8Dx/zmDXUhsbVNqatPjtpdO0UZBmGZGmVI=
-github.com/netobserv/loki-client-go v0.0.0-20211018150932-cb17208397a9 h1:c2swm3EamzgjBq9idNbEs5bNz20FJo/HK6uxyigXekQ=
-github.com/netobserv/loki-client-go v0.0.0-20211018150932-cb17208397a9/go.mod h1:LHXpc5tjKvsfZn0pwLKrvlgEhZcCaw3Di9mUEZGAI4E=
 github.com/netobserv/loki-client-go v0.0.0-20220927092034-f37122a54500 h1:RmnoJe/ci5q+QdM7upFdxiU+D8F3L3qTd5wXCwwHefw=
 github.com/netobserv/loki-client-go v0.0.0-20220927092034-f37122a54500/go.mod h1:LHXpc5tjKvsfZn0pwLKrvlgEhZcCaw3Di9mUEZGAI4E=
 github.com/netobserv/netobserv-ebpf-agent v0.1.1-0.20220608092850-3fd4695b7cc2 h1:K7SjoqEfzpMfIjHV85Lg8UDMvZu8rPfrsgKRoo7W30o=
diff --git a/pkg/api/transform_network.go b/pkg/api/transform_network.go
index 0c97d826..ce914806 100644
--- a/pkg/api/transform_network.go
+++ b/pkg/api/transform_network.go
@@ -36,6 +36,15 @@ func (tn *TransformNetwork) GetServiceFiles() (string, string) {
 	return p, s
 }
 
+const (
+	OpAddRegexIf    = "add_regex_if"
+	OpAddIf         = "add_if"
+	OpAddSubnet     = "add_subnet"
+	OpAddLocation   = "add_location"
+	OpAddService    = "add_service"
+	OpAddKubernetes = "add_kubernetes"
+)
+
 type TransformNetworkOperationEnum struct {
 	AddRegExIf    string `yaml:"add_regex_if" json:"add_regex_if" doc:"add output field if input field satisfies regex pattern from parameters field"`
 	AddIf         string `yaml:"add_if" json:"add_if" doc:"add output field if input field satisfies criteria from parameters field"`
diff --git a/pkg/config/config.go b/pkg/config/config.go
index 0d0757b5..6400cb1d 100644
--- a/pkg/config/config.go
+++ b/pkg/config/config.go
@@ -20,6 +20,7 @@ package config
 import (
 	"bytes"
 	"encoding/json"
+	"time"
 
 	"github.com/netobserv/flowlogs-pipeline/pkg/api"
 	"github.com/sirupsen/logrus"
@@ -38,6 +39,7 @@ type ConfigFileStruct struct {
 	Pipeline        []Stage         `yaml:"pipeline,omitempty" json:"pipeline,omitempty"`
 	Parameters      []StageParam    `yaml:"parameters,omitempty" json:"parameters,omitempty"`
 	MetricsSettings MetricsSettings `yaml:"metricsSettings,omitempty" json:"metricsSettings,omitempty"`
+	PerfSettings    PerfSettings    `yaml:"perfSettings,omitempty" json:"perfSettings,omitempty"`
 }
 
 type Health struct {
@@ -60,6 +62,12 @@ type MetricsSettings struct {
 	NoPanic bool   `yaml:"noPanic,omitempty" json:"noPanic,omitempty"`
 }
 
+// PerfSettings allows setting some internal configuration parameters
+type PerfSettings struct {
+	BatcherMaxLen  int           `yaml:"batcherMaxLen,omitempty" json:"batcherMaxLen,omitempty"`
+	BatcherTimeout time.Duration `yaml:"batcherMaxTimeout,omitempty" json:"batcherMaxTimeout,omitempty"`
+}
+
 type Stage struct {
 	Name    string `yaml:"name" json:"name"`
 	Follows string `yaml:"follows,omitempty" json:"follows,omitempty"`
diff --git a/pkg/pipeline/decode/decode.go b/pkg/pipeline/decode/decode.go
index 6a5aa0e6..e771af10 100644
--- a/pkg/pipeline/decode/decode.go
+++ b/pkg/pipeline/decode/decode.go
@@ -25,7 +25,7 @@ import (
 )
 
 type Decoder interface {
-	Decode(in [][]byte) []config.GenericMap
+	Decode(in []byte) (config.GenericMap, error)
 }
 
 func GetDecoder(params api.Decoder) (Decoder, error) {
diff --git a/pkg/pipeline/decode/decode_json.go b/pkg/pipeline/decode/decode_json.go
index f6f89edd..4bffdcaf 100644
--- a/pkg/pipeline/decode/decode_json.go
+++ b/pkg/pipeline/decode/decode_json.go
@@ -29,32 +29,26 @@ type DecodeJson struct {
 }
 
 // Decode decodes input strings to a list of flow entries
-func (c *DecodeJson) Decode(in [][]byte) []config.GenericMap {
-	out := make([]config.GenericMap, 0)
-	for _, line := range in {
-		if log.IsLevelEnabled(log.DebugLevel) {
-			log.Debugf("decodeJson: line = %v", string(line))
-		}
-		var decodedLine map[string]interface{}
-		err := json.Unmarshal(line, &decodedLine)
-		if err != nil {
-			log.Errorf("decodeJson Decode: error unmarshalling a line: %v", err)
-			log.Errorf("line = %s", line)
+func (c *DecodeJson) Decode(line []byte) (config.GenericMap, error) {
+
+	if log.IsLevelEnabled(log.DebugLevel) {
+		log.Debugf("decodeJson: line = %v", string(line))
+	}
+	var decodedLine map[string]interface{}
+	if err := json.Unmarshal(line, &decodedLine); err != nil {
+		return nil, err
+	}
+	decodedLine2 := make(config.GenericMap, len(decodedLine))
+	// flows directly ingested by flp-transformer won't have this field, so we need to add it
+	// here. If the received line already contains the field, it will be overridden later
+	decodedLine2["TimeReceived"] = time.Now().Unix()
+	for k, v := range decodedLine {
+		if v == nil {
 			continue
 		}
-		decodedLine2 := make(config.GenericMap, len(decodedLine))
-		// flows directly ingested by flp-transformer won't have this field, so we need to add it
-		// here. If the received line already contains the field, it will be overridden later
-		decodedLine2["TimeReceived"] = time.Now().Unix()
-		for k, v := range decodedLine {
-			if v == nil {
-				continue
-			}
-			decodedLine2[k] = v
-		}
-		out = append(out, decodedLine2)
+		decodedLine2[k] = v
 	}
-	return out
+	return decodedLine2, nil
 }
 
 // NewDecodeJson create a new decode
diff --git a/pkg/pipeline/decode/decode_protobuf.go b/pkg/pipeline/decode/decode_protobuf.go
index a7808b9f..7a2660a1 100644
--- a/pkg/pipeline/decode/decode_protobuf.go
+++ b/pkg/pipeline/decode/decode_protobuf.go
@@ -7,12 +7,9 @@ import (
 
 	"github.com/netobserv/flowlogs-pipeline/pkg/config"
 	"github.com/netobserv/netobserv-ebpf-agent/pkg/pbflow"
-	"github.com/sirupsen/logrus"
 	"google.golang.org/protobuf/proto"
 )
 
-var pflog = logrus.WithField("component", "Protobuf")
-
 // Protobuf decodes protobuf flow records definitions, as forwarded by
 // ingest.NetObservAgent, into a Generic Map that follows the same naming conventions
 // as the IPFIX flows from ingest.IngestCollector
@@ -25,30 +22,15 @@ func NewProtobuf() (*Protobuf, error) {
 
 // Decode decodes the protobuf raw flows and returns a list of GenericMaps representing all
 // the flows there
-func (p *Protobuf) Decode(rawFlows [][]byte) []config.GenericMap {
-	flows := make([]config.GenericMap, 0, len(rawFlows))
-	for _, pbRaw := range rawFlows {
-		record := pbflow.Record{}
-		if err := proto.Unmarshal(pbRaw, &record); err != nil {
-			pflog.WithError(err).Debug("can't unmarshall received protobuf flow. Ignoring")
-			continue
-		}
-		flows = append(flows, pbFlowToMap(&record))
+func (p *Protobuf) Decode(rawFlow []byte) (config.GenericMap, error) {
+	record := pbflow.Record{}
+	if err := proto.Unmarshal(rawFlow, &record); err != nil {
+		return nil, fmt.Errorf("unmarshaling ProtoBuf record: %w", err)
 	}
-	return flows
-}
-
-// PBRecordsAsMaps transform all the flows in a pbflow.Records entry into a slice
-// of GenericMaps
-func PBRecordsAsMaps(flow *pbflow.Records) []config.GenericMap {
-	out := make([]config.GenericMap, 0, len(flow.Entries))
-	for _, entry := range flow.Entries {
-		out = append(out, pbFlowToMap(entry))
-	}
-	return out
+	return PBFlowToMap(&record), nil
 }
 
-func pbFlowToMap(flow *pbflow.Record) config.GenericMap {
+func PBFlowToMap(flow *pbflow.Record) config.GenericMap {
 	if flow == nil {
 		return config.GenericMap{}
 	}
diff --git a/pkg/pipeline/encode/encode.go b/pkg/pipeline/encode/encode.go
index 62cb9080..a7c6eefa 100644
--- a/pkg/pipeline/encode/encode.go
+++ b/pkg/pipeline/encode/encode.go
@@ -23,22 +23,20 @@ import (
 )
 
 type encodeNone struct {
-	prevRecords []config.GenericMap
+	prevRecord config.GenericMap
 }
 
 type Encoder interface {
-	Encode(in []config.GenericMap)
+	Encode(in config.GenericMap)
 }
 
 // Encode encodes a flow before being stored
-func (t *encodeNone) Encode(in []config.GenericMap) {
-	t.prevRecords = in
+func (t *encodeNone) Encode(in config.GenericMap) {
+	t.prevRecord = in
 }
 
 // NewEncodeNone create a new encode
 func NewEncodeNone() (Encoder, error) {
 	log.Debugf("entering NewEncodeNone")
-	return &encodeNone{
-		prevRecords: make([]config.GenericMap, 0),
-	}, nil
+	return &encodeNone{}, nil
 }
diff --git a/pkg/pipeline/encode/encode_kafka.go b/pkg/pipeline/encode/encode_kafka.go
index 1f1b7a6a..15cf9e92 100644
--- a/pkg/pipeline/encode/encode_kafka.go
+++ b/pkg/pipeline/encode/encode_kafka.go
@@ -47,23 +47,17 @@ type encodeKafka struct {
 }
 
 // Encode writes entries to kafka topic
-func (r *encodeKafka) Encode(in []config.GenericMap) {
-	log.Debugf("entering encodeKafka Encode, #items = %d", len(in))
-	var msgs []kafkago.Message
-	msgs = make([]kafkago.Message, 0)
-	for _, entry := range in {
-		var entryByteArray []byte
-		entryByteArray, _ = json.Marshal(entry)
-		msg := kafkago.Message{
-			Value: entryByteArray,
-		}
-		msgs = append(msgs, msg)
+func (r *encodeKafka) Encode(entry config.GenericMap) {
+	var entryByteArray []byte
+	entryByteArray, _ = json.Marshal(entry)
+	msg := kafkago.Message{
+		Value: entryByteArray,
 	}
-	err := r.kafkaWriter.WriteMessages(context.Background(), msgs...)
+	err := r.kafkaWriter.WriteMessages(context.Background(), msg)
 	if err != nil {
 		log.Errorf("encodeKafka error: %v", err)
 	} else {
-		r.recordsWritten.Add(float64(len(msgs)))
+		r.recordsWritten.Inc()
 	}
 }
 
diff --git a/pkg/pipeline/encode/encode_prom.go b/pkg/pipeline/encode/encode_prom.go
index 09727879..8e278cf0 100644
--- a/pkg/pipeline/encode/encode_prom.go
+++ b/pkg/pipeline/encode/encode_prom.go
@@ -82,15 +82,7 @@ var (
 )
 
 // Encode encodes a metric before being stored
-func (e *EncodeProm) Encode(metrics []config.GenericMap) {
-	log.Debugf("entering EncodeProm Encode")
-	for _, metric := range metrics {
-		e.EncodeMetric(metric)
-	}
-	log.Debugf("cache = %v", e.mCache)
-}
-
-func (e *EncodeProm) EncodeMetric(metricRecord config.GenericMap) {
+func (e *EncodeProm) Encode(metricRecord config.GenericMap) {
 	log.Debugf("entering EncodeMetric. metricRecord = %v", metricRecord)
 
 	// Process counters
diff --git a/pkg/pipeline/extract/aggregate/aggregate.go b/pkg/pipeline/extract/aggregate/aggregate.go
index 70d9c22b..0428694a 100644
--- a/pkg/pipeline/extract/aggregate/aggregate.go
+++ b/pkg/pipeline/extract/aggregate/aggregate.go
@@ -60,7 +60,7 @@ type GroupState struct {
 	totalCount       int
 }
 
-func (aggregate Aggregate) LabelsFromEntry(entry config.GenericMap) (Labels, bool) {
+func (aggregate *Aggregate) LabelsFromEntry(entry config.GenericMap) (Labels, bool) {
 	allLabelsFound := true
 	labels := Labels{}
 
@@ -69,7 +69,7 @@ func (aggregate Aggregate) LabelsFromEntry(entry config.GenericMap) (Labels, boo
 		if !ok {
 			allLabelsFound = false
 		}
-		labels[key] = fmt.Sprintf("%v", value)
+		labels[key] = fmt.Sprint(value)
 	}
 
 	return labels, allLabelsFound
@@ -95,7 +95,7 @@ func (labels Labels) getNormalizedValues() NormalizedValues {
 	return NormalizedValues(normalizedAsString)
 }
 
-func (aggregate Aggregate) FilterEntry(entry config.GenericMap) (error, NormalizedValues, Labels) {
+func (aggregate *Aggregate) FilterEntry(entry config.GenericMap) (error, NormalizedValues, Labels) {
 	labels, allLabelsFound := aggregate.LabelsFromEntry(entry)
 	if !allLabelsFound {
 		return fmt.Errorf("missing keys in entry"), "", nil
@@ -120,7 +120,7 @@ func getInitValue(operation string) float64 {
 	}
 }
 
-func (aggregate Aggregate) UpdateByEntry(entry config.GenericMap, normalizedValues NormalizedValues, labels Labels) error {
+func (aggregate *Aggregate) UpdateByEntry(entry config.GenericMap, normalizedValues NormalizedValues, labels Labels) error {
 
 	aggregate.mutex.Lock()
 	defer aggregate.mutex.Unlock()
@@ -180,7 +180,7 @@ func (aggregate Aggregate) UpdateByEntry(entry config.GenericMap, normalizedValu
 	return nil
 }
 
-func (aggregate Aggregate) Evaluate(entries []config.GenericMap) error {
+func (aggregate *Aggregate) Evaluate(entries []config.GenericMap) error {
 	for _, entry := range entries {
 		// filter entries matching labels with aggregates
 		err, normalizedValues, labels := aggregate.FilterEntry(entry)
@@ -199,7 +199,7 @@ func (aggregate Aggregate) Evaluate(entries []config.GenericMap) error {
 	return nil
 }
 
-func (aggregate Aggregate) GetMetrics() []config.GenericMap {
+func (aggregate *Aggregate) GetMetrics() []config.GenericMap {
 	aggregate.mutex.Lock()
 	defer aggregate.mutex.Unlock()
 
diff --git a/pkg/pipeline/ingest/ingest.go b/pkg/pipeline/ingest/ingest.go
index 7d5b3843..de804f1a 100644
--- a/pkg/pipeline/ingest/ingest.go
+++ b/pkg/pipeline/ingest/ingest.go
@@ -20,7 +20,7 @@ package ingest
 import "github.com/netobserv/flowlogs-pipeline/pkg/config"
 
 type Ingester interface {
-	Ingest(out chan<- []config.GenericMap)
+	Ingest(out chan<- config.GenericMap)
 }
 type IngesterNone struct {
 }
diff --git a/pkg/pipeline/ingest/ingest_collector.go b/pkg/pipeline/ingest/ingest_collector.go
index d1bfb9a2..e4cd06fe 100644
--- a/pkg/pipeline/ingest/ingest_collector.go
+++ b/pkg/pipeline/ingest/ingest_collector.go
@@ -22,7 +22,6 @@ import (
 	"encoding/binary"
 	"fmt"
 	"net"
-	"time"
 
 	ms "github.com/mitchellh/mapstructure"
 	"github.com/netobserv/flowlogs-pipeline/pkg/api"
@@ -39,20 +38,16 @@ import (
 )
 
 const (
-	channelSize           = 1000
-	defaultBatchFlushTime = time.Second
-	defaultBatchMaxLength = 500
+	channelSize = 1000
 )
 
 type ingestCollector struct {
-	hostname       string
-	port           int
-	portLegacy     int
-	in             chan map[string]interface{}
-	batchFlushTime time.Duration
-	batchMaxLength int
-	exitChan       <-chan struct{}
-	metrics        *metrics
+	hostname   string
+	port       int
+	portLegacy int
+	in         chan map[string]interface{}
+	exitChan   <-chan struct{}
+	metrics    *metrics
 }
 
 // TransportWrapper is an implementation of the goflow2 transport interface
@@ -102,7 +97,7 @@ func (w *TransportWrapper) Send(_, data []byte) error {
 }
 
 // Ingest ingests entries from a network collector using goflow2 library (https://github.com/netsampler/goflow2)
-func (c *ingestCollector) Ingest(out chan<- []config.GenericMap) {
+func (c *ingestCollector) Ingest(out chan<- config.GenericMap) {
 	ctx := context.Background()
 	c.metrics.createOutQueueLen(out)
 
@@ -148,40 +143,14 @@ func (c *ingestCollector) initCollectorListener(ctx context.Context) {
 	}
 }
 
-func (c *ingestCollector) processLogLines(out chan<- []config.GenericMap) {
-	var records []config.GenericMap
-	// Maximum batch time for each batch
-	flushRecords := time.NewTicker(c.batchFlushTime)
-	defer flushRecords.Stop()
+func (c *ingestCollector) processLogLines(out chan<- config.GenericMap) {
 	for {
 		select {
 		case <-c.exitChan:
 			log.Debugf("exiting ingestCollector because of signal")
 			return
 		case record := <-c.in:
-			records = append(records, record)
-			if len(records) >= c.batchMaxLength {
-				log.Debugf("ingestCollector sending %d entries, %d entries waiting", len(records), len(c.in))
-				c.metrics.batchSize.Observe(float64(len(records)))
-				log.Debugf("ingestCollector records = %v", records)
-				out <- records
-				records = []config.GenericMap{}
-			}
-		case <-flushRecords.C:
-			// Process batch of records (if not empty)
-			if len(records) > 0 {
-				if len(c.in) > 0 {
-					for len(records) < c.batchMaxLength && len(c.in) > 0 {
-						record := <-c.in
-						records = append(records, record)
-					}
-				}
-				log.Debugf("ingestCollector sending %d entries, %d entries waiting", len(records), len(c.in))
-				c.metrics.batchSize.Observe(float64(len(records)))
-				log.Debugf("ingestCollector records = %v", records)
-				out <- records
-				records = []config.GenericMap{}
-			}
+			out <- record
 		}
 	}
 }
@@ -203,22 +172,15 @@ func NewIngestCollector(opMetrics *operational.Metrics, params config.StageParam
 	log.Infof("port = %d", jsonIngestCollector.Port)
 	log.Infof("portLegacy = %d", jsonIngestCollector.PortLegacy)
 
-	bml := defaultBatchMaxLength
-	if jsonIngestCollector.BatchMaxLen != 0 {
-		bml = jsonIngestCollector.BatchMaxLen
-	}
-
 	in := make(chan map[string]interface{}, channelSize)
 	metrics := newMetrics(opMetrics, params.Name, params.Ingest.Type, func() int { return len(in) })
 
 	return &ingestCollector{
-		hostname:       jsonIngestCollector.HostName,
-		port:           jsonIngestCollector.Port,
-		portLegacy:     jsonIngestCollector.PortLegacy,
-		exitChan:       pUtils.ExitChannel(),
-		batchFlushTime: defaultBatchFlushTime,
-		batchMaxLength: bml,
-		in:             in,
-		metrics:        metrics,
+		hostname:   jsonIngestCollector.HostName,
+		port:       jsonIngestCollector.Port,
+		portLegacy: jsonIngestCollector.PortLegacy,
+		exitChan:   pUtils.ExitChannel(),
+		in:         in,
+		metrics:    metrics,
 	}, nil
 }
diff --git a/pkg/pipeline/ingest/ingest_fake.go b/pkg/pipeline/ingest/ingest_fake.go
index d2e550ca..a9472d59 100644
--- a/pkg/pipeline/ingest/ingest_fake.go
+++ b/pkg/pipeline/ingest/ingest_fake.go
@@ -19,6 +19,7 @@ package ingest
 
 import (
 	"fmt"
+	"sync/atomic"
 
 	"github.com/netobserv/flowlogs-pipeline/pkg/config"
 	"github.com/netobserv/flowlogs-pipeline/pkg/pipeline/utils"
@@ -26,13 +27,14 @@ import (
 )
 
 type IngestFake struct {
+	Count    int64
 	params   config.Ingest
-	In       chan []config.GenericMap
+	In       chan config.GenericMap
 	exitChan <-chan struct{}
 }
 
 // Ingest reads records from an input channel and writes them as-is to the output channel
-func (inf *IngestFake) Ingest(out chan<- []config.GenericMap) {
+func (inf *IngestFake) Ingest(out chan<- config.GenericMap) {
 	for {
 		select {
 		case <-inf.exitChan:
@@ -40,6 +42,7 @@ func (inf *IngestFake) Ingest(out chan<- []config.GenericMap) {
 			return
 		case records := <-inf.In:
 			out <- records
+			atomic.AddInt64(&inf.Count, 1)
 		}
 	}
 }
@@ -53,7 +56,7 @@ func NewIngestFake(params config.StageParam) (Ingester, error) {
 
 	return &IngestFake{
 		params:   *params.Ingest,
-		In:       make(chan []config.GenericMap),
+		In:       make(chan config.GenericMap),
 		exitChan: utils.ExitChannel(),
 	}, nil
 }
diff --git a/pkg/pipeline/ingest/ingest_file.go b/pkg/pipeline/ingest/ingest_file.go
index 19f21054..2c42da8b 100644
--- a/pkg/pipeline/ingest/ingest_file.go
+++ b/pkg/pipeline/ingest/ingest_file.go
@@ -43,7 +43,7 @@ const (
 )
 
 // Ingest ingests entries from a file and resends the same data every delaySeconds seconds
-func (ingestF *IngestFile) Ingest(out chan<- []config.GenericMap) {
+func (ingestF *IngestFile) Ingest(out chan<- config.GenericMap) {
 	var filename string
 	if ingestF.params.File != nil {
 		filename = ingestF.params.File.Filename
@@ -63,16 +63,11 @@ func (ingestF *IngestFile) Ingest(out chan<- []config.GenericMap) {
 		log.Debugf("%s", text)
 		lines = append(lines, []byte(text))
 	}
-	decoded := ingestF.decoder.Decode(lines)
-	log.Debugf("IngestFile decoded = %v", decoded)
 
 	log.Debugf("Ingesting %d log lines from %s", len(lines), filename)
 	switch ingestF.params.Type {
 	case "file":
-		ingestF.PrevRecords = decoded
-		ingestF.TotalRecords = len(lines)
-		log.Debugf("ingestFile sending %d lines", len(lines))
-		out <- decoded
+		ingestF.sendAllLines(lines, out)
 	case "file_loop":
 		// loop forever
 		ticker := time.NewTicker(time.Duration(delaySeconds) * time.Second)
@@ -82,10 +77,7 @@ func (ingestF *IngestFile) Ingest(out chan<- []config.GenericMap) {
 				log.Debugf("exiting ingestFile because of signal")
 				return
 			case <-ticker.C:
-				ingestF.PrevRecords = decoded
-				ingestF.TotalRecords += len(lines)
-				log.Debugf("ingestFile sending %d lines", len(lines))
-				out <- decoded
+				ingestF.sendAllLines(lines, out)
 			}
 		}
 	case "file_chunks":
@@ -93,16 +85,29 @@ func (ingestF *IngestFile) Ingest(out chan<- []config.GenericMap) {
 		ingestF.TotalRecords = len(lines)
 		for len(lines) > 0 {
 			if len(lines) > chunkLines {
-				out <- decoded[:chunkLines]
+				ingestF.sendAllLines(lines[:chunkLines], out)
 				lines = lines[chunkLines:]
 			} else {
-				out <- decoded
+				ingestF.sendAllLines(lines, out)
 				lines = nil
 			}
 		}
 	}
 }
 
+func (ingestF *IngestFile) sendAllLines(lines [][]byte, out chan<- config.GenericMap) {
+	log.Debugf("ingestFile sending %d lines", len(lines))
+	ingestF.TotalRecords = len(lines)
+	for _, line := range lines {
+		decoded, err := ingestF.decoder.Decode(line)
+		if err != nil {
+			log.WithError(err).Warnf("ignoring line")
+			continue
+		}
+		out <- decoded
+	}
+}
+
 // NewIngestFile create a new ingester
 func NewIngestFile(params config.StageParam) (Ingester, error) {
 	log.Debugf("entering NewIngestFile")
diff --git a/pkg/pipeline/ingest/ingest_grpc.go b/pkg/pipeline/ingest/ingest_grpc.go
index 4c39c886..aac8f4b6 100644
--- a/pkg/pipeline/ingest/ingest_grpc.go
+++ b/pkg/pipeline/ingest/ingest_grpc.go
@@ -43,7 +43,7 @@ func NewGRPCProtobuf(opMetrics *operational.Metrics, params config.StageParam) (
 	flowPackets := make(chan *pbflow.Records, bufLen)
 	metrics := newMetrics(opMetrics, params.Name, params.Ingest.Type, func() int { return len(flowPackets) })
 	collector, err := grpc.StartCollector(netObserv.Port, flowPackets,
-		grpc.WithGRPCServerOptions(grpc2.UnaryInterceptor(instrumentGRPC(netObserv.Port, metrics))))
+		grpc.WithGRPCServerOptions(grpc2.UnaryInterceptor(instrumentGRPC(metrics))))
 	if err != nil {
 		return nil, err
 	}
@@ -54,7 +54,7 @@ func NewGRPCProtobuf(opMetrics *operational.Metrics, params config.StageParam) (
 	}, nil
 }
 
-func (no *GRPCProtobuf) Ingest(out chan<- []config.GenericMap) {
+func (no *GRPCProtobuf) Ingest(out chan<- config.GenericMap) {
 	no.metrics.createOutQueueLen(out)
 	go func() {
 		<-utils.ExitChannel()
@@ -62,9 +62,10 @@ func (no *GRPCProtobuf) Ingest(out chan<- []config.GenericMap) {
 		no.collector.Close()
 	}()
 	for fp := range no.flowPackets {
-		records := decode.PBRecordsAsMaps(fp)
-		log.Debugf("GRPCProtobuf records = %v", records)
-		out <- records
+		log.Debugf("GRPCProtobuf records len = %v", len(fp.Entries))
+		for _, entry := range fp.Entries {
+			out <- decode.PBFlowToMap(entry)
+		}
 	}
 }
 
@@ -74,7 +75,7 @@ func (no *GRPCProtobuf) Close() error {
 	return err
 }
 
-func instrumentGRPC(port int, m *metrics) grpc2.UnaryServerInterceptor {
+func instrumentGRPC(m *metrics) grpc2.UnaryServerInterceptor {
 	return func(
 		ctx context.Context,
 		req interface{},
diff --git a/pkg/pipeline/ingest/ingest_kafka.go b/pkg/pipeline/ingest/ingest_kafka.go
index db2731b0..e67fe317 100644
--- a/pkg/pipeline/ingest/ingest_kafka.go
+++ b/pkg/pipeline/ingest/ingest_kafka.go
@@ -26,7 +26,6 @@ import (
 	"github.com/netobserv/flowlogs-pipeline/pkg/operational"
 	"github.com/netobserv/flowlogs-pipeline/pkg/pipeline/decode"
 	"github.com/netobserv/flowlogs-pipeline/pkg/pipeline/utils"
-
 	kafkago "github.com/segmentio/kafka-go"
 	log "github.com/sirupsen/logrus"
 	"golang.org/x/net/context"
@@ -53,15 +52,15 @@ const defaultKafkaBatchMaxLength = 500
 const defaultKafkaCommitInterval = 500
 
 // Ingest ingests entries from kafka topic
-func (ingestK *ingestKafka) Ingest(out chan<- []config.GenericMap) {
+func (k *ingestKafka) Ingest(out chan<- config.GenericMap) {
 	log.Debugf("entering ingestKafka.Ingest")
-	ingestK.metrics.createOutQueueLen(out)
+	k.metrics.createOutQueueLen(out)
 
 	// initialize background listener
-	ingestK.kafkaListener()
+	k.kafkaListener()
 
 	// forever process log lines received by collector
-	ingestK.processLogLines(out)
+	k.processLogLines(out)
 }
 
 // background thread to read kafka messages; place received items into ingestKafka input channel
@@ -107,53 +106,28 @@ func (k *ingestKafka) processRecordDelay(record config.GenericMap) {
 	k.metrics.latency.Observe(delay)
 }
 
-func (k *ingestKafka) processBatch(out chan<- []config.GenericMap, records [][]byte, timer *operational.Timer) {
-	log.Debugf("ingestKafka sending %d records, %d entries waiting", len(records), len(k.in))
-
+func (k *ingestKafka) processRecord(record []byte, out chan<- config.GenericMap) {
 	// Decode batch
-	decoded := k.decoder.Decode(records)
-
-	for _, record := range decoded {
-		k.processRecordDelay(record)
+	decoded, err := k.decoder.Decode(record)
+	if err != nil {
+		log.WithError(err).Warnf("ignoring flow")
+		return
 	}
-
-	// Stage duration
-	timer.ObserveMilliseconds()
+	k.processRecordDelay(decoded)
 
 	// Send batch
 	out <- decoded
 }
 
 // read items from ingestKafka input channel, pool them, and send down the pipeline
-func (k *ingestKafka) processLogLines(out chan<- []config.GenericMap) {
-	var records [][]byte
-	timer := k.metrics.stageDurationTimer()
-	duration := time.Duration(k.batchReadTimeout) * time.Millisecond
-	flushRecords := time.NewTicker(duration)
+func (k *ingestKafka) processLogLines(out chan<- config.GenericMap) {
 	for {
 		select {
 		case <-k.exitChan:
 			log.Debugf("exiting ingestKafka because of signal")
 			return
 		case record := <-k.in:
-			timer.StartOnce()
-			records = append(records, record)
-			if len(records) >= k.batchMaxLength {
-				k.processBatch(out, records, timer)
-				records = [][]byte{}
-			}
-		case <-flushRecords.C: // Maximum batch time for each batch
-			// Process batch of records (if not empty)
-			if len(records) > 0 {
-				if len(k.in) > 0 {
-					for len(records) < k.batchMaxLength && len(k.in) > 0 {
-						record := <-k.in
-						records = append(records, record)
-					}
-				}
-				k.processBatch(out, records, timer)
-				records = [][]byte{}
-			}
+			k.processRecord(record, out)
 		}
 	}
 }
diff --git a/pkg/pipeline/ingest/metrics.go b/pkg/pipeline/ingest/metrics.go
index fea5b4f6..c49f1a42 100644
--- a/pkg/pipeline/ingest/metrics.go
+++ b/pkg/pipeline/ingest/metrics.go
@@ -58,7 +58,7 @@ func newMetrics(opMetrics *operational.Metrics, stage, stageType string, inGauge
 	}
 }
 
-func (m *metrics) createOutQueueLen(out chan<- []config.GenericMap) {
+func (m *metrics) createOutQueueLen(out chan<- config.GenericMap) {
 	m.CreateOutQueueSizeGauge(m.stage, func() int { return len(out) })
 }
 
diff --git a/pkg/pipeline/pipeline.go b/pkg/pipeline/pipeline.go
index 228b30e1..b9fa1d74 100644
--- a/pkg/pipeline/pipeline.go
+++ b/pkg/pipeline/pipeline.go
@@ -55,10 +55,7 @@ func NewPipeline(cfg *config.ConfigFileStruct) (*Pipeline, error) {
 	configParams := cfg.Parameters
 	log.Debugf("configParams = %v ", configParams)
 
-	// Get global metrics settings
-	om := operational.NewMetrics(&cfg.MetricsSettings)
-
-	build := newBuilder(configParams, stages, om)
+	build := newBuilder(cfg)
 	if err := build.readStages(); err != nil {
 		return nil, err
 	}
diff --git a/pkg/pipeline/pipeline_builder.go b/pkg/pipeline/pipeline_builder.go
index 6e876f05..3c2490b2 100644
--- a/pkg/pipeline/pipeline_builder.go
+++ b/pkg/pipeline/pipeline_builder.go
@@ -14,12 +14,18 @@ import (
 	"github.com/netobserv/flowlogs-pipeline/pkg/pipeline/extract/conntrack"
 	"github.com/netobserv/flowlogs-pipeline/pkg/pipeline/ingest"
 	"github.com/netobserv/flowlogs-pipeline/pkg/pipeline/transform"
+	"github.com/netobserv/flowlogs-pipeline/pkg/pipeline/utils"
 	"github.com/netobserv/flowlogs-pipeline/pkg/pipeline/write"
 	"github.com/netobserv/gopipes/pkg/node"
 	"github.com/prometheus/client_golang/prometheus"
 	log "github.com/sirupsen/logrus"
 )
 
+const (
+	defaultExtractBatching        = 1000
+	defaultExtractBatchingTimeout = 5 * time.Second
+)
+
 // Error wraps any error caused by a wrong formation of the pipeline
 type Error struct {
 	StageName string
@@ -45,6 +51,8 @@ type builder struct {
 	terminalNodes    []*node.Terminal
 	opMetrics        *operational.Metrics
 	stageDuration    *prometheus.HistogramVec
+	batchMaxLen      int
+	batchTimeout     time.Duration
 }
 
 type pipelineEntry struct {
@@ -57,16 +65,29 @@ type pipelineEntry struct {
 	Writer      write.Writer
 }
 
-func newBuilder(params []config.StageParam, stages []config.Stage, opMetrics *operational.Metrics) *builder {
+func newBuilder(cfg *config.ConfigFileStruct) *builder {
+	// Get global metrics settings
+	opMetrics := operational.NewMetrics(&cfg.MetricsSettings)
 	stageDuration := opMetrics.GetOrCreateStageDurationHisto()
 
+	bl := cfg.PerfSettings.BatcherMaxLen
+	if bl == 0 {
+		bl = defaultExtractBatching
+	}
+	bt := cfg.PerfSettings.BatcherTimeout
+	if bt == 0 {
+		bt = defaultExtractBatchingTimeout
+	}
+
 	return &builder{
 		pipelineEntryMap: map[string]*pipelineEntry{},
 		createdStages:    map[string]interface{}{},
-		configStages:     stages,
-		configParams:     params,
+		configStages:     cfg.Pipeline,
+		configParams:     cfg.Parameters,
 		opMetrics:        opMetrics,
 		stageDuration:    stageDuration,
+		batchMaxLen:      bl,
+		batchTimeout:     bt,
 	}
 }
 
@@ -238,7 +259,7 @@ func (b *builder) getStageNode(pe *pipelineEntry, stageID string) (interface{},
 		b.startNodes = append(b.startNodes, init)
 		stage = init
 	case StageWrite:
-		term := node.AsTerminal(func(in <-chan []config.GenericMap) {
+		term := node.AsTerminal(func(in <-chan config.GenericMap) {
 			b.opMetrics.CreateInQueueSizeGauge(stageID, func() int { return len(in) })
 			for i := range in {
 				b.runMeasured(stageID, func() {
@@ -249,7 +270,7 @@ func (b *builder) getStageNode(pe *pipelineEntry, stageID string) (interface{},
 		b.terminalNodes = append(b.terminalNodes, term)
 		stage = term
 	case StageEncode:
-		encode := node.AsTerminal(func(in <-chan []config.GenericMap) {
+		encode := node.AsTerminal(func(in <-chan config.GenericMap) {
 			b.opMetrics.CreateInQueueSizeGauge(stageID, func() int { return len(in) })
 			for i := range in {
 				b.runMeasured(stageID, func() {
@@ -260,24 +281,31 @@ func (b *builder) getStageNode(pe *pipelineEntry, stageID string) (interface{},
 		b.terminalNodes = append(b.terminalNodes, encode)
 		stage = encode
 	case StageTransform:
-		stage = node.AsMiddle(func(in <-chan []config.GenericMap, out chan<- []config.GenericMap) {
+		stage = node.AsMiddle(func(in <-chan config.GenericMap, out chan<- config.GenericMap) {
 			b.opMetrics.CreateInQueueSizeGauge(stageID, func() int { return len(in) })
 			b.opMetrics.CreateOutQueueSizeGauge(stageID, func() int { return len(out) })
 			for i := range in {
 				b.runMeasured(stageID, func() {
-					out <- pe.Transformer.Transform(i)
+					if transformed, ok := pe.Transformer.Transform(i); ok {
+						out <- transformed
+					}
 				})
 			}
 		})
 	case StageExtract:
-		stage = node.AsMiddle(func(in <-chan []config.GenericMap, out chan<- []config.GenericMap) {
+		stage = node.AsMiddle(func(in <-chan config.GenericMap, out chan<- config.GenericMap) {
 			b.opMetrics.CreateInQueueSizeGauge(stageID, func() int { return len(in) })
 			b.opMetrics.CreateOutQueueSizeGauge(stageID, func() int { return len(out) })
-			for i := range in {
-				b.runMeasured(stageID, func() {
-					out <- pe.Extractor.Extract(i)
-				})
-			}
+			// TODO: replace batcher by rewriting the different extractor implementations
+			// to keep the status while processing flows one by one
+			utils.Batcher(utils.ExitChannel(), b.batchMaxLen, b.batchTimeout, in,
+				func(maps []config.GenericMap) {
+					outs := pe.Extractor.Extract(maps)
+					for _, o := range outs {
+						out <- o
+					}
+				},
+			)
 		})
 	default:
 		return nil, &Error{
diff --git a/pkg/pipeline/transform/transform.go b/pkg/pipeline/transform/transform.go
index 74ec4b31..598fb83b 100644
--- a/pkg/pipeline/transform/transform.go
+++ b/pkg/pipeline/transform/transform.go
@@ -24,15 +24,15 @@ import (
 )
 
 type Transformer interface {
-	Transform(in []config.GenericMap) []config.GenericMap
+	Transform(in config.GenericMap) (config.GenericMap, bool)
 }
 
 type transformNone struct {
 }
 
 // Transform transforms a flow before being stored
-func (t *transformNone) Transform(f []config.GenericMap) []config.GenericMap {
-	return f
+func (t *transformNone) Transform(f config.GenericMap) (config.GenericMap, bool) {
+	return f, true
 }
 
 // NewTransformNone create a new transform
diff --git a/pkg/pipeline/transform/transform_filter.go b/pkg/pipeline/transform/transform_filter.go
index 5365a08b..9cca6cb1 100644
--- a/pkg/pipeline/transform/transform_filter.go
+++ b/pkg/pipeline/transform/transform_filter.go
@@ -28,48 +28,39 @@ type Filter struct {
 }
 
 // Transform transforms a flow
-func (f *Filter) Transform(input []config.GenericMap) []config.GenericMap {
-	log.Debugf("f = %v", f)
-	output := make([]config.GenericMap, 0)
-	for _, entry := range input {
-		// copy input entry before transform to avoid alteration on parallel stages
-		outputEntry := entry.Copy()
-		addToOutput := true
-		for _, rule := range f.Rules {
-			log.Debugf("rule = %v", rule)
-			switch rule.Type {
-			case api.TransformFilterOperationName("RemoveField"):
-				delete(outputEntry, rule.Input)
-			case api.TransformFilterOperationName("RemoveEntryIfExists"):
-				if _, ok := entry[rule.Input]; ok {
-					addToOutput = false
-				}
-			case api.TransformFilterOperationName("RemoveEntryIfDoesntExist"):
-				if _, ok := entry[rule.Input]; !ok {
-					addToOutput = false
-				}
-			case api.TransformFilterOperationName("RemoveEntryIfEqual"):
-				if val, ok := entry[rule.Input]; ok {
-					if val == rule.Value {
-						addToOutput = false
-					}
+func (f *Filter) Transform(entry config.GenericMap) (config.GenericMap, bool) {
+	log.Tracef("f = %v", f)
+	outputEntry := entry.Copy()
+	for _, rule := range f.Rules {
+		log.Debugf("rule = %v", rule)
+		switch rule.Type {
+		case api.TransformFilterOperationName("RemoveField"):
+			delete(outputEntry, rule.Input)
+		case api.TransformFilterOperationName("RemoveEntryIfExists"):
+			if _, ok := entry[rule.Input]; ok {
+				return nil, false
+			}
+		case api.TransformFilterOperationName("RemoveEntryIfDoesntExist"):
+			if _, ok := entry[rule.Input]; !ok {
+				return nil, false
+			}
+		case api.TransformFilterOperationName("RemoveEntryIfEqual"):
+			if val, ok := entry[rule.Input]; ok {
+				if val == rule.Value {
+					return nil, false
 				}
-			case api.TransformFilterOperationName("RemoveEntryIfNotEqual"):
-				if val, ok := entry[rule.Input]; ok {
-					if val != rule.Value {
-						addToOutput = false
-					}
+			}
+		case api.TransformFilterOperationName("RemoveEntryIfNotEqual"):
+			if val, ok := entry[rule.Input]; ok {
+				if val != rule.Value {
+					return nil, false
 				}
-			default:
-				log.Panicf("unknown type %s for transform.Filter rule: %v", rule.Type, rule)
 			}
-		}
-		if addToOutput {
-			output = append(output, outputEntry)
-			log.Debugf("Transform.GenericMap = %v", outputEntry)
+		default:
+			log.Panicf("unknown type %s for transform.Filter rule: %v", rule.Type, rule)
 		}
 	}
-	return output
+	return outputEntry, true
 }
 
 // NewTransformFilter create a new filter transform
diff --git a/pkg/pipeline/transform/transform_generic.go b/pkg/pipeline/transform/transform_generic.go
index 820e2c09..9b9a84bc 100644
--- a/pkg/pipeline/transform/transform_generic.go
+++ b/pkg/pipeline/transform/transform_generic.go
@@ -29,27 +29,23 @@ type Generic struct {
 }
 
 // Transform transforms a flow to a new set of keys
-func (g *Generic) Transform(input []config.GenericMap) []config.GenericMap {
-	log.Debugf("entering Generic Transform g = %v", g)
-	output := make([]config.GenericMap, 0)
-	for _, entry := range input {
-		log.Debugf("entry.GenericMap = %v", entry)
-		outputEntry := make(config.GenericMap)
-		if g.policy != "replace_keys" {
-			// copy old map to new map
-			for key, value := range entry {
-				outputEntry[key] = value
-			}
+func (g *Generic) Transform(entry config.GenericMap) (config.GenericMap, bool) {
+	log.Tracef("entering Generic Transform g = %v", g)
+	log.Tracef("entry.GenericMap = %v", entry)
+	outputEntry := make(config.GenericMap)
+	if g.policy != "replace_keys" {
+		// copy old map to new map
+		for key, value := range entry {
+			outputEntry[key] = value
 		}
-		log.Debugf("outputEntry = %v", outputEntry)
-		for _, transformRule := range g.rules {
-			log.Debugf("transformRule = %v", transformRule)
-			outputEntry[transformRule.Output] = entry[transformRule.Input]
-		}
-		log.Debugf("Transform.GenericMap = %v", outputEntry)
-		output = append(output, outputEntry)
 	}
-	return output
+	log.Debugf("outputEntry = %v", outputEntry)
+	for _, transformRule := range g.rules {
+		log.Debugf("transformRule = %v", transformRule)
+		outputEntry[transformRule.Output] = entry[transformRule.Input]
+	}
+	log.Debugf("Transform.GenericMap = %v", outputEntry)
+	return outputEntry, true
 }
 
 // NewTransformGeneric create a new transform
diff --git a/pkg/pipeline/transform/transform_network.go b/pkg/pipeline/transform/transform_network.go
index 591fcb14..22c2d76f 100644
--- a/pkg/pipeline/transform/transform_network.go
+++ b/pkg/pipeline/transform/transform_network.go
@@ -38,23 +38,14 @@ type Network struct {
 	svcNames *netdb.ServiceNames
 }
 
-func (n *Network) Transform(input []config.GenericMap) []config.GenericMap {
-	outputEntries := make([]config.GenericMap, 0)
-	for _, entry := range input {
-		outputEntry := n.TransformEntry(entry)
-		outputEntries = append(outputEntries, outputEntry)
-	}
-	return outputEntries
-}
-
-func (n *Network) TransformEntry(inputEntry config.GenericMap) config.GenericMap {
+func (n *Network) Transform(inputEntry config.GenericMap) (config.GenericMap, bool) {
 	// copy input entry before transform to avoid alteration on parallel stages
 	outputEntry := inputEntry.Copy()
 
 	// TODO: for efficiency and maintainability, maybe each case in the switch below should be an individual implementation of Transformer
 	for _, rule := range n.Rules {
 		switch rule.Type {
-		case api.TransformNetworkOperationName("AddRegExIf"):
+		case api.OpAddRegexIf:
 			matched, err := regexp.MatchString(rule.Parameters, fmt.Sprintf("%s", outputEntry[rule.Input]))
 			if err != nil {
 				continue
@@ -63,7 +54,7 @@ func (n *Network) TransformEntry(inputEntry config.GenericMap) config.GenericMap
 				outputEntry[rule.Output] = outputEntry[rule.Input]
 				outputEntry[rule.Output+"_Matched"] = true
 			}
-		case api.TransformNetworkOperationName("AddIf"):
+		case api.OpAddIf:
 			expressionString := fmt.Sprintf("val %s", rule.Parameters)
 			expression, err := govaluate.NewEvaluableExpression(expressionString)
 			if err != nil {
@@ -79,14 +70,14 @@ func (n *Network) TransformEntry(inputEntry config.GenericMap) config.GenericMap
 				}
 				outputEntry[rule.Output+"_Evaluate"] = true
 			}
-		case api.TransformNetworkOperationName("AddSubnet"):
+		case api.OpAddSubnet:
 			_, ipv4Net, err := net.ParseCIDR(fmt.Sprintf("%v%s", outputEntry[rule.Input], rule.Parameters))
 			if err != nil {
 				log.Errorf("Can't find subnet for IP %v and prefix length %s - err %v", outputEntry[rule.Input], rule.Parameters, err)
 				continue
 			}
 			outputEntry[rule.Output] = ipv4Net.String()
-		case api.TransformNetworkOperationName("AddLocation"):
+		case api.OpAddLocation:
 			var locationInfo *location.Info
 			err, locationInfo := location.GetLocation(fmt.Sprintf("%s", outputEntry[rule.Input]))
 			if err != nil {
@@ -99,7 +90,7 @@ func (n *Network) TransformEntry(inputEntry config.GenericMap) config.GenericMap
 			outputEntry[rule.Output+"_CityName"] = locationInfo.CityName
 			outputEntry[rule.Output+"_Latitude"] = locationInfo.Latitude
 			outputEntry[rule.Output+"_Longitude"] = locationInfo.Longitude
-		case api.TransformNetworkOperationName("AddService"):
+		case api.OpAddService:
 			protocol := fmt.Sprintf("%v", outputEntry[rule.Parameters])
 			portNumber, err := strconv.Atoi(fmt.Sprintf("%v", outputEntry[rule.Input]))
 			if err != nil {
@@ -122,7 +113,7 @@ func (n *Network) TransformEntry(inputEntry config.GenericMap) config.GenericMap
 				}
 			}
 			outputEntry[rule.Output] = serviceName
-		case api.TransformNetworkOperationName("AddKubernetes"):
+		case api.OpAddKubernetes:
 			kubeInfo, err := kubernetes.Data.GetInfo(fmt.Sprintf("%s", outputEntry[rule.Input]))
 			if err != nil {
 				log.Debugf("Can't find kubernetes info for IP %v err %v", outputEntry[rule.Input], err)
@@ -149,7 +140,7 @@ func (n *Network) TransformEntry(inputEntry config.GenericMap) config.GenericMap
 		}
 	}
 
-	return outputEntry
+	return outputEntry, true
 }
 
 // NewTransformNetwork create a new transform
diff --git a/pkg/pipeline/utils/batcher.go b/pkg/pipeline/utils/batcher.go
new file mode 100644
index 00000000..1b01816e
--- /dev/null
+++ b/pkg/pipeline/utils/batcher.go
@@ -0,0 +1,44 @@
+package utils
+
+import (
+	"time"
+
+	"github.com/netobserv/flowlogs-pipeline/pkg/config"
+	"github.com/sirupsen/logrus"
+)
+
+func Batcher(
+	closeCh <-chan struct{},
+	maxBatchLength int,
+	batchTimeout time.Duration,
+	inCh <-chan config.GenericMap,
+	action func([]config.GenericMap),
+) {
+	log := logrus.WithField("component", "utils.Batcher")
+	invokeTicker := time.NewTicker(batchTimeout)
+	var entries []config.GenericMap
+	log.Debug("starting")
+	for {
+		select {
+		case <-closeCh:
+			log.Debug("exiting due to closeCh")
+			return
+		case <-invokeTicker.C:
+			if len(entries) == 0 {
+				continue
+			}
+			log.Debugf("ticker signal: invoking action with %d entries", len(entries))
+			es := entries
+			entries = nil
+			action(es)
+		case gm := <-inCh:
+			entries = append(entries, gm)
+			if len(entries) >= maxBatchLength {
+				log.Debugf("batch complete: invoking action with %d entries", len(entries))
+				es := entries
+				entries = nil
+				action(es)
+			}
+		}
+	}
+}
diff --git a/pkg/pipeline/write/write.go b/pkg/pipeline/write/write.go
index 5b21e16b..46c2db27 100644
--- a/pkg/pipeline/write/write.go
+++ b/pkg/pipeline/write/write.go
@@ -18,21 +18,37 @@
 package write
 
 import (
+	"sync"
+
 	"github.com/netobserv/flowlogs-pipeline/pkg/config"
 	log "github.com/sirupsen/logrus"
 )
 
 type Writer interface {
-	Write(in []config.GenericMap)
+	Write(in config.GenericMap)
 }
 type WriteNone struct {
-	PrevRecords []config.GenericMap
+	// synchronized access to avoid race conditions
+	mt          sync.Mutex
+	prevRecords []config.GenericMap
 }
 
 // Write writes entries
-func (t *WriteNone) Write(in []config.GenericMap) {
+func (t *WriteNone) Write(in config.GenericMap) {
 	log.Debugf("entering Write none, in = %v", in)
-	t.PrevRecords = in
+	t.mt.Lock()
+	t.prevRecords = append(t.prevRecords, in)
+	t.mt.Unlock()
+}
+
+func (t *WriteNone) PrevRecords() []config.GenericMap {
+	t.mt.Lock()
+	defer t.mt.Unlock()
+	var copies []config.GenericMap
+	for _, rec := range t.prevRecords {
+		copies = append(copies, rec.Copy())
+	}
+	return copies
 }
 
 // NewWriteNone create a new write
diff --git a/pkg/pipeline/write/write_fake.go b/pkg/pipeline/write/write_fake.go
index 088e7705..3c408035 100644
--- a/pkg/pipeline/write/write_fake.go
+++ b/pkg/pipeline/write/write_fake.go
@@ -18,40 +18,39 @@
 package write
 
 import (
+	"sync"
+
 	"github.com/netobserv/flowlogs-pipeline/pkg/config"
 	log "github.com/sirupsen/logrus"
 )
 
 type WriteFake struct {
-	AllRecords []config.GenericMap
-	wait       chan struct{}
+	// access is locked and copied to avoid race condition errors during tests
+	mt         sync.Mutex
+	allRecords []config.GenericMap
 }
 
 // Write stores in memory all records.
-func (w *WriteFake) Write(in []config.GenericMap) {
-	log.Debugf("entering writeFake Write")
-	log.Debugf("writeFake: number of entries = %d", len(in))
-	for _, r := range in {
-		w.AllRecords = append(w.AllRecords, r.Copy())
-	}
-	close(w.wait)
+func (w *WriteFake) Write(in config.GenericMap) {
+	log.Trace("entering writeFake Write")
+	w.mt.Lock()
+	w.allRecords = append(w.allRecords, in.Copy())
+	w.mt.Unlock()
 }
 
-// Wait waits until Write() is done processing all the records.
-func (w *WriteFake) Wait() {
-	<-w.wait
-}
-
-// ResetWait resets the wait channel to allow waiters to block on Wait() for the next Write().
-// It should be invoked after Write() is done and all waiters are released.
-func (w *WriteFake) ResetWait() {
-	w.wait = make(chan struct{})
+func (w *WriteFake) AllRecords() []config.GenericMap {
+	w.mt.Lock()
+	defer w.mt.Unlock()
+	var copies []config.GenericMap
+	for _, r := range w.allRecords {
+		copies = append(copies, r.Copy())
+	}
+	return copies
 }
 
 // NewWriteFake creates a new write.
-func NewWriteFake(params config.StageParam) (Writer, error) {
+func NewWriteFake(_ config.StageParam) (Writer, error) {
 	log.Debugf("entering NewWriteFake")
 	w := &WriteFake{}
-	w.ResetWait()
 	return w, nil
 }
diff --git a/pkg/pipeline/write/write_loki.go b/pkg/pipeline/write/write_loki.go
index 8a7633ff..6c3d8db1 100644
--- a/pkg/pipeline/write/write_loki.go
+++ b/pkg/pipeline/write/write_loki.go
@@ -204,11 +204,9 @@ func getFloat64(timestamp interface{}) (ft float64, ok bool) {
 }
 
 // Write writes a flow before being stored
-func (l *Loki) Write(entries []config.GenericMap) {
+func (l *Loki) Write(entry config.GenericMap) {
 	log.Debugf("entering Loki Write")
-	for _, entry := range entries {
-		l.in <- entry
-	}
+	l.in <- entry
 }
 
 func (l *Loki) processRecords() {
diff --git a/pkg/pipeline/write/write_stdout.go b/pkg/pipeline/write/write_stdout.go
index 70442b50..f37086e6 100644
--- a/pkg/pipeline/write/write_stdout.go
+++ b/pkg/pipeline/write/write_stdout.go
@@ -34,32 +34,25 @@ type writeStdout struct {
 }
 
 // Write writes a flow before being stored
-func (t *writeStdout) Write(in []config.GenericMap) {
-	log.Debugf("entering writeStdout Write")
-	log.Debugf("writeStdout: number of entries = %d", len(in))
+func (t *writeStdout) Write(v config.GenericMap) {
+	log.Tracef("entering writeStdout Write")
 	if t.format == "json" {
-		for _, v := range in {
-			txt, _ := json.Marshal(v)
-			fmt.Println(string(txt))
-		}
+		txt, _ := json.Marshal(v)
+		fmt.Println(string(txt))
 	} else if t.format == "fields" {
-		for _, v := range in {
-			var order sort.StringSlice
-			for fieldName := range v {
-				order = append(order, fieldName)
-			}
-			order.Sort()
-			w := tabwriter.NewWriter(os.Stdout, 0, 0, 1, ' ', 0)
-			fmt.Fprintf(w, "\n\nFlow record at %s:\n", time.Now().Format(time.StampMilli))
-			for _, field := range order {
-				fmt.Fprintf(w, "%v\t=\t%v\n", field, v[field])
-			}
-			w.Flush()
+		var order sort.StringSlice
+		for fieldName := range v {
+			order = append(order, fieldName)
 		}
-	} else {
-		for _, v := range in {
-			fmt.Printf("%s: %v\n", time.Now().Format(time.StampMilli), v)
+		order.Sort()
+		w := tabwriter.NewWriter(os.Stdout, 0, 0, 1, ' ', 0)
+		fmt.Fprintf(w, "\n\nFlow record at %s:\n", time.Now().Format(time.StampMilli))
+		for _, field := range order {
+			fmt.Fprintf(w, "%v\t=\t%v\n", field, v[field])
 		}
+		w.Flush()
+	} else {
+		fmt.Printf("%s: %v\n", time.Now().Format(time.StampMilli), v)
 	}
 }
 
diff --git a/vendor/github.com/prometheus/client_golang/prometheus/promauto/auto.go b/vendor/github.com/prometheus/client_golang/prometheus/promauto/auto.go
deleted file mode 100644
index f8d50d1f..00000000
--- a/vendor/github.com/prometheus/client_golang/prometheus/promauto/auto.go
+++ /dev/null
@@ -1,376 +0,0 @@
-// 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.
-
-// Package promauto provides alternative constructors for the fundamental
-// Prometheus metric types and their …Vec and …Func variants. The difference to
-// their counterparts in the prometheus package is that the promauto
-// constructors return Collectors that are already registered with a
-// registry. There are two sets of constructors. The constructors in the first
-// set are top-level functions, while the constructors in the other set are
-// methods of the Factory type. The top-level function return Collectors
-// registered with the global registry (prometheus.DefaultRegisterer), while the
-// methods return Collectors registered with the registry the Factory was
-// constructed with. All constructors panic if the registration fails.
-//
-// The following example is a complete program to create a histogram of normally
-// distributed random numbers from the math/rand package:
-//
-//      package main
-//
-//      import (
-//              "math/rand"
-//              "net/http"
-//
-//              "github.com/prometheus/client_golang/prometheus"
-//              "github.com/prometheus/client_golang/prometheus/promauto"
-//              "github.com/prometheus/client_golang/prometheus/promhttp"
-//      )
-//
-//      var histogram = promauto.NewHistogram(prometheus.HistogramOpts{
-//              Name:    "random_numbers",
-//              Help:    "A histogram of normally distributed random numbers.",
-//              Buckets: prometheus.LinearBuckets(-3, .1, 61),
-//      })
-//
-//      func Random() {
-//              for {
-//                      histogram.Observe(rand.NormFloat64())
-//              }
-//      }
-//
-//      func main() {
-//              go Random()
-//              http.Handle("/metrics", promhttp.Handler())
-//              http.ListenAndServe(":1971", nil)
-//      }
-//
-// Prometheus's version of a minimal hello-world program:
-//
-//      package main
-//
-//      import (
-//      	"fmt"
-//      	"net/http"
-//
-//      	"github.com/prometheus/client_golang/prometheus"
-//      	"github.com/prometheus/client_golang/prometheus/promauto"
-//      	"github.com/prometheus/client_golang/prometheus/promhttp"
-//      )
-//
-//      func main() {
-//      	http.Handle("/", promhttp.InstrumentHandlerCounter(
-//      		promauto.NewCounterVec(
-//      			prometheus.CounterOpts{
-//      				Name: "hello_requests_total",
-//      				Help: "Total number of hello-world requests by HTTP code.",
-//      			},
-//      			[]string{"code"},
-//      		),
-//      		http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-//      			fmt.Fprint(w, "Hello, world!")
-//      		}),
-//      	))
-//      	http.Handle("/metrics", promhttp.Handler())
-//      	http.ListenAndServe(":1971", nil)
-//      }
-//
-// A Factory is created with the With(prometheus.Registerer) function, which
-// enables two usage pattern. With(prometheus.Registerer) can be called once per
-// line:
-//
-//        var (
-//        	reg           = prometheus.NewRegistry()
-//        	randomNumbers = promauto.With(reg).NewHistogram(prometheus.HistogramOpts{
-//        		Name:    "random_numbers",
-//        		Help:    "A histogram of normally distributed random numbers.",
-//        		Buckets: prometheus.LinearBuckets(-3, .1, 61),
-//        	})
-//        	requestCount = promauto.With(reg).NewCounterVec(
-//        		prometheus.CounterOpts{
-//        			Name: "http_requests_total",
-//        			Help: "Total number of HTTP requests by status code and method.",
-//        		},
-//        		[]string{"code", "method"},
-//        	)
-//        )
-//
-// Or it can be used to create a Factory once to be used multiple times:
-//
-//        var (
-//        	reg           = prometheus.NewRegistry()
-//        	factory       = promauto.With(reg)
-//        	randomNumbers = factory.NewHistogram(prometheus.HistogramOpts{
-//        		Name:    "random_numbers",
-//        		Help:    "A histogram of normally distributed random numbers.",
-//        		Buckets: prometheus.LinearBuckets(-3, .1, 61),
-//        	})
-//        	requestCount = factory.NewCounterVec(
-//        		prometheus.CounterOpts{
-//        			Name: "http_requests_total",
-//        			Help: "Total number of HTTP requests by status code and method.",
-//        		},
-//        		[]string{"code", "method"},
-//        	)
-//        )
-//
-// This appears very handy. So why are these constructors locked away in a
-// separate package?
-//
-// The main problem is that registration may fail, e.g. if a metric inconsistent
-// with or equal to the newly to be registered one is already registered.
-// Therefore, the Register method in the prometheus.Registerer interface returns
-// an error, and the same is the case for the top-level prometheus.Register
-// function that registers with the global registry. The prometheus package also
-// provides MustRegister versions for both. They panic if the registration
-// fails, and they clearly call this out by using the Must…  idiom. Panicking is
-// problematic in this case because it doesn't just happen on input provided by
-// the caller that is invalid on its own. Things are a bit more subtle here:
-// Metric creation and registration tend to be spread widely over the
-// codebase. It can easily happen that an incompatible metric is added to an
-// unrelated part of the code, and suddenly code that used to work perfectly
-// fine starts to panic (provided that the registration of the newly added
-// metric happens before the registration of the previously existing
-// metric). This may come as an even bigger surprise with the global registry,
-// where simply importing another package can trigger a panic (if the newly
-// imported package registers metrics in its init function). At least, in the
-// prometheus package, creation of metrics and other collectors is separate from
-// registration. You first create the metric, and then you decide explicitly if
-// you want to register it with a local or the global registry, and if you want
-// to handle the error or risk a panic. With the constructors in the promauto
-// package, registration is automatic, and if it fails, it will always
-// panic. Furthermore, the constructors will often be called in the var section
-// of a file, which means that panicking will happen as a side effect of merely
-// importing a package.
-//
-// A separate package allows conservative users to entirely ignore it. And
-// whoever wants to use it, will do so explicitly, with an opportunity to read
-// this warning.
-//
-// Enjoy promauto responsibly!
-package promauto
-
-import "github.com/prometheus/client_golang/prometheus"
-
-// NewCounter works like the function of the same name in the prometheus package
-// but it automatically registers the Counter with the
-// prometheus.DefaultRegisterer. If the registration fails, NewCounter panics.
-func NewCounter(opts prometheus.CounterOpts) prometheus.Counter {
-	return With(prometheus.DefaultRegisterer).NewCounter(opts)
-}
-
-// NewCounterVec works like the function of the same name in the prometheus
-// package but it automatically registers the CounterVec with the
-// prometheus.DefaultRegisterer. If the registration fails, NewCounterVec
-// panics.
-func NewCounterVec(opts prometheus.CounterOpts, labelNames []string) *prometheus.CounterVec {
-	return With(prometheus.DefaultRegisterer).NewCounterVec(opts, labelNames)
-}
-
-// NewCounterFunc works like the function of the same name in the prometheus
-// package but it automatically registers the CounterFunc with the
-// prometheus.DefaultRegisterer. If the registration fails, NewCounterFunc
-// panics.
-func NewCounterFunc(opts prometheus.CounterOpts, function func() float64) prometheus.CounterFunc {
-	return With(prometheus.DefaultRegisterer).NewCounterFunc(opts, function)
-}
-
-// NewGauge works like the function of the same name in the prometheus package
-// but it automatically registers the Gauge with the
-// prometheus.DefaultRegisterer. If the registration fails, NewGauge panics.
-func NewGauge(opts prometheus.GaugeOpts) prometheus.Gauge {
-	return With(prometheus.DefaultRegisterer).NewGauge(opts)
-}
-
-// NewGaugeVec works like the function of the same name in the prometheus
-// package but it automatically registers the GaugeVec with the
-// prometheus.DefaultRegisterer. If the registration fails, NewGaugeVec panics.
-func NewGaugeVec(opts prometheus.GaugeOpts, labelNames []string) *prometheus.GaugeVec {
-	return With(prometheus.DefaultRegisterer).NewGaugeVec(opts, labelNames)
-}
-
-// NewGaugeFunc works like the function of the same name in the prometheus
-// package but it automatically registers the GaugeFunc with the
-// prometheus.DefaultRegisterer. If the registration fails, NewGaugeFunc panics.
-func NewGaugeFunc(opts prometheus.GaugeOpts, function func() float64) prometheus.GaugeFunc {
-	return With(prometheus.DefaultRegisterer).NewGaugeFunc(opts, function)
-}
-
-// NewSummary works like the function of the same name in the prometheus package
-// but it automatically registers the Summary with the
-// prometheus.DefaultRegisterer. If the registration fails, NewSummary panics.
-func NewSummary(opts prometheus.SummaryOpts) prometheus.Summary {
-	return With(prometheus.DefaultRegisterer).NewSummary(opts)
-}
-
-// NewSummaryVec works like the function of the same name in the prometheus
-// package but it automatically registers the SummaryVec with the
-// prometheus.DefaultRegisterer. If the registration fails, NewSummaryVec
-// panics.
-func NewSummaryVec(opts prometheus.SummaryOpts, labelNames []string) *prometheus.SummaryVec {
-	return With(prometheus.DefaultRegisterer).NewSummaryVec(opts, labelNames)
-}
-
-// NewHistogram works like the function of the same name in the prometheus
-// package but it automatically registers the Histogram with the
-// prometheus.DefaultRegisterer. If the registration fails, NewHistogram panics.
-func NewHistogram(opts prometheus.HistogramOpts) prometheus.Histogram {
-	return With(prometheus.DefaultRegisterer).NewHistogram(opts)
-}
-
-// NewHistogramVec works like the function of the same name in the prometheus
-// package but it automatically registers the HistogramVec with the
-// prometheus.DefaultRegisterer. If the registration fails, NewHistogramVec
-// panics.
-func NewHistogramVec(opts prometheus.HistogramOpts, labelNames []string) *prometheus.HistogramVec {
-	return With(prometheus.DefaultRegisterer).NewHistogramVec(opts, labelNames)
-}
-
-// NewUntypedFunc works like the function of the same name in the prometheus
-// package but it automatically registers the UntypedFunc with the
-// prometheus.DefaultRegisterer. If the registration fails, NewUntypedFunc
-// panics.
-func NewUntypedFunc(opts prometheus.UntypedOpts, function func() float64) prometheus.UntypedFunc {
-	return With(prometheus.DefaultRegisterer).NewUntypedFunc(opts, function)
-}
-
-// Factory provides factory methods to create Collectors that are automatically
-// registered with a Registerer. Create a Factory with the With function,
-// providing a Registerer to auto-register created Collectors with. The zero
-// value of a Factory creates Collectors that are not registered with any
-// Registerer. All methods of the Factory panic if the registration fails.
-type Factory struct {
-	r prometheus.Registerer
-}
-
-// With creates a Factory using the provided Registerer for registration of the
-// created Collectors. If the provided Registerer is nil, the returned Factory
-// creates Collectors that are not registered with any Registerer.
-func With(r prometheus.Registerer) Factory { return Factory{r} }
-
-// NewCounter works like the function of the same name in the prometheus package
-// but it automatically registers the Counter with the Factory's Registerer.
-func (f Factory) NewCounter(opts prometheus.CounterOpts) prometheus.Counter {
-	c := prometheus.NewCounter(opts)
-	if f.r != nil {
-		f.r.MustRegister(c)
-	}
-	return c
-}
-
-// NewCounterVec works like the function of the same name in the prometheus
-// package but it automatically registers the CounterVec with the Factory's
-// Registerer.
-func (f Factory) NewCounterVec(opts prometheus.CounterOpts, labelNames []string) *prometheus.CounterVec {
-	c := prometheus.NewCounterVec(opts, labelNames)
-	if f.r != nil {
-		f.r.MustRegister(c)
-	}
-	return c
-}
-
-// NewCounterFunc works like the function of the same name in the prometheus
-// package but it automatically registers the CounterFunc with the Factory's
-// Registerer.
-func (f Factory) NewCounterFunc(opts prometheus.CounterOpts, function func() float64) prometheus.CounterFunc {
-	c := prometheus.NewCounterFunc(opts, function)
-	if f.r != nil {
-		f.r.MustRegister(c)
-	}
-	return c
-}
-
-// NewGauge works like the function of the same name in the prometheus package
-// but it automatically registers the Gauge with the Factory's Registerer.
-func (f Factory) NewGauge(opts prometheus.GaugeOpts) prometheus.Gauge {
-	g := prometheus.NewGauge(opts)
-	if f.r != nil {
-		f.r.MustRegister(g)
-	}
-	return g
-}
-
-// NewGaugeVec works like the function of the same name in the prometheus
-// package but it automatically registers the GaugeVec with the Factory's
-// Registerer.
-func (f Factory) NewGaugeVec(opts prometheus.GaugeOpts, labelNames []string) *prometheus.GaugeVec {
-	g := prometheus.NewGaugeVec(opts, labelNames)
-	if f.r != nil {
-		f.r.MustRegister(g)
-	}
-	return g
-}
-
-// NewGaugeFunc works like the function of the same name in the prometheus
-// package but it automatically registers the GaugeFunc with the Factory's
-// Registerer.
-func (f Factory) NewGaugeFunc(opts prometheus.GaugeOpts, function func() float64) prometheus.GaugeFunc {
-	g := prometheus.NewGaugeFunc(opts, function)
-	if f.r != nil {
-		f.r.MustRegister(g)
-	}
-	return g
-}
-
-// NewSummary works like the function of the same name in the prometheus package
-// but it automatically registers the Summary with the Factory's Registerer.
-func (f Factory) NewSummary(opts prometheus.SummaryOpts) prometheus.Summary {
-	s := prometheus.NewSummary(opts)
-	if f.r != nil {
-		f.r.MustRegister(s)
-	}
-	return s
-}
-
-// NewSummaryVec works like the function of the same name in the prometheus
-// package but it automatically registers the SummaryVec with the Factory's
-// Registerer.
-func (f Factory) NewSummaryVec(opts prometheus.SummaryOpts, labelNames []string) *prometheus.SummaryVec {
-	s := prometheus.NewSummaryVec(opts, labelNames)
-	if f.r != nil {
-		f.r.MustRegister(s)
-	}
-	return s
-}
-
-// NewHistogram works like the function of the same name in the prometheus
-// package but it automatically registers the Histogram with the Factory's
-// Registerer.
-func (f Factory) NewHistogram(opts prometheus.HistogramOpts) prometheus.Histogram {
-	h := prometheus.NewHistogram(opts)
-	if f.r != nil {
-		f.r.MustRegister(h)
-	}
-	return h
-}
-
-// NewHistogramVec works like the function of the same name in the prometheus
-// package but it automatically registers the HistogramVec with the Factory's
-// Registerer.
-func (f Factory) NewHistogramVec(opts prometheus.HistogramOpts, labelNames []string) *prometheus.HistogramVec {
-	h := prometheus.NewHistogramVec(opts, labelNames)
-	if f.r != nil {
-		f.r.MustRegister(h)
-	}
-	return h
-}
-
-// NewUntypedFunc works like the function of the same name in the prometheus
-// package but it automatically registers the UntypedFunc with the Factory's
-// Registerer.
-func (f Factory) NewUntypedFunc(opts prometheus.UntypedOpts, function func() float64) prometheus.UntypedFunc {
-	u := prometheus.NewUntypedFunc(opts, function)
-	if f.r != nil {
-		f.r.MustRegister(u)
-	}
-	return u
-}


Test patch:
diff --git a/pkg/pipeline/aggregate_prom_test.go b/pkg/pipeline/aggregate_prom_test.go
index e6bb4530..43b99476 100644
--- a/pkg/pipeline/aggregate_prom_test.go
+++ b/pkg/pipeline/aggregate_prom_test.go
@@ -178,7 +178,9 @@ parameters:
 			// we use ElementsMatch() rather than Equals()
 			require.ElementsMatch(t, tt.expectedAggs, actualAggs)
 
-			promEncode.Encode(actualAggs)
+			for _, aa := range actualAggs {
+				promEncode.Encode(aa)
+			}
 			exposed := test.ReadExposedMetrics(t)
 
 			for _, expected := range tt.expectedEncode {
diff --git a/pkg/pipeline/conntrack_integ_test.go b/pkg/pipeline/conntrack_integ_test.go
index 9146149d..449e59a0 100644
--- a/pkg/pipeline/conntrack_integ_test.go
+++ b/pkg/pipeline/conntrack_integ_test.go
@@ -20,9 +20,11 @@ package pipeline
 import (
 	"bufio"
 	"os"
+	"sync/atomic"
 	"testing"
 	"time"
 
+	test2 "github.com/mariomac/guara/pkg/test"
 	"github.com/netobserv/flowlogs-pipeline/pkg/config"
 	"github.com/netobserv/flowlogs-pipeline/pkg/pipeline/decode"
 	"github.com/netobserv/flowlogs-pipeline/pkg/pipeline/ingest"
@@ -95,24 +97,31 @@ func TestConnTrack(t *testing.T) {
 	var err error
 	v, cfg := test.InitConfig(t, testConfigConntrack)
 	require.NotNil(t, v)
+	cfg.PerfSettings.BatcherMaxLen = 200_000
+	cfg.PerfSettings.BatcherTimeout = 2 * time.Second
 
 	mainPipeline, err = NewPipeline(cfg)
 	require.NoError(t, err)
 
 	go mainPipeline.Run()
 
-	in := mainPipeline.pipelineStages[0].Ingester.(*ingest.IngestFake).In
+	ingest := mainPipeline.pipelineStages[0].Ingester.(*ingest.IngestFake)
+	in := ingest.In
 	writer := mainPipeline.pipelineStages[2].Writer.(*write.WriteFake)
 
-	ingestFile(t, in, "../../hack/examples/ocp-ipfix-flowlogs.json")
-	writer.Wait()
-	writer.ResetWait()
+	sentLines := ingestFile(t, in, "../../hack/examples/ocp-ipfix-flowlogs.json")
+
+	// wait for all the lines to be ingested
+	test2.Eventually(t, 15*time.Second, func(t require.TestingT) {
+		require.EqualValues(t, atomic.LoadInt64(&ingest.Count), sentLines,
+			"sent: %d. got: %d", sentLines, ingest.Count)
+	}, test2.Interval(10*time.Millisecond))
 
 	// Wait a moment to make the connections expired
 	time.Sleep(2 * time.Second)
-	// Send an empty list to the pipeline to allow the connection tracking output end connection records
-	in <- []config.GenericMap{}
-	writer.Wait()
+
+	// Send something to the pipeline to allow the connection tracking output end connection records
+	in <- config.GenericMap{"DstAddr": "1.2.3.4"}
 
 	// Verify that the output records contain an expected end connection record.
 	expected := config.GenericMap{
@@ -131,10 +140,14 @@ func TestConnTrack(t *testing.T) {
 		"_RecordType":   "endConnection",
 		"numFlowLogs":   5.0,
 	}
-	require.Containsf(t, writer.AllRecords, expected, "The output records don't include the expected record %v", expected)
+	// Wait for the record to be eventually forwarded to the writer
+	test2.Eventually(t, 15*time.Second, func(t require.TestingT) {
+		require.Containsf(t, writer.AllRecords(), expected,
+			"The output records don't include the expected record %v", expected)
+	})
 }
 
-func ingestFile(t *testing.T, in chan<- []config.GenericMap, filepath string) {
+func ingestFile(t *testing.T, in chan<- config.GenericMap, filepath string) int {
 	t.Helper()
 	file, err := os.Open(filepath)
 	require.NoError(t, err)
@@ -147,8 +160,14 @@ func ingestFile(t *testing.T, in chan<- []config.GenericMap, filepath string) {
 		text := scanner.Text()
 		lines = append(lines, []byte(text))
 	}
+	submittedLines := 0
 	decoder, err := decode.NewDecodeJson()
 	require.NoError(t, err)
-	decoded := decoder.Decode(lines)
-	in <- decoded
+	for _, line := range lines {
+		line, err := decoder.Decode(line)
+		require.NoError(t, err)
+		in <- line
+		submittedLines++
+	}
+	return submittedLines
 }
diff --git a/pkg/pipeline/decode/decode_json_test.go b/pkg/pipeline/decode/decode_json_test.go
index bc9fe07e..ec2c93de 100644
--- a/pkg/pipeline/decode/decode_json_test.go
+++ b/pkg/pipeline/decode/decode_json_test.go
@@ -20,7 +20,6 @@ package decode
 import (
 	"testing"
 
-	"github.com/netobserv/flowlogs-pipeline/pkg/config"
 	"github.com/stretchr/testify/require"
 )
 
@@ -33,27 +32,21 @@ func initNewDecodeJson(t *testing.T) Decoder {
 func TestDecodeJson(t *testing.T) {
 	newDecode := initNewDecodeJson(t)
 	decodeJson := newDecode.(*DecodeJson)
-	inputString1 := "{\"varInt\": 12, \"varString\":\"testString\", \"varBool\":false}"
-	inputString2 := "{\"varInt\": 14, \"varString\":\"testString2\", \"varBool\":true, \"TimeReceived\":12345}"
-	inputString3 := "{}"
-	inputStringErr := "{\"varInt\": 14, \"varString\",\"testString2\", \"varBool\":true}"
-	var in [][]byte
-	var out []config.GenericMap
-	out = decodeJson.Decode(in)
-	require.Equal(t, 0, len(out))
-	in = append(in, []byte(inputString1))
-	in = append(in, []byte(inputString2))
-	in = append(in, []byte(inputString3))
-	in = append(in, []byte(inputStringErr))
-	out = decodeJson.Decode(in)
-	require.Equal(t, len(out), 3)
-	require.Equal(t, float64(12), out[0]["varInt"])
-	require.Equal(t, "testString", out[0]["varString"])
-	require.Equal(t, false, out[0]["varBool"])
+
+	out, err := decodeJson.Decode([]byte(
+		"{\"varInt\": 12, \"varString\":\"testString\", \"varBool\":false}"))
+	require.NoError(t, err)
+	require.Equal(t, float64(12), out["varInt"])
+	require.Equal(t, "testString", out["varString"])
+	require.Equal(t, false, out["varBool"])
 	// TimeReceived is added if it does not exist
-	require.NotZero(t, out[0]["TimeReceived"])
+	require.NotZero(t, out["TimeReceived"])
+
+	out, err = decodeJson.Decode([]byte(
+		"{\"varInt\": 14, \"varString\":\"testString2\", \"varBool\":true, \"TimeReceived\":12345}"))
+	require.NoError(t, err)
 	// TimeReceived is kept if it already existed
-	require.EqualValues(t, 12345, out[1]["TimeReceived"])
+	require.EqualValues(t, 12345, out["TimeReceived"])
 
 	// TODO: Check for more complicated json structures
 }
@@ -61,13 +54,7 @@ func TestDecodeJson(t *testing.T) {
 func TestDecodeJsonTimestamps(t *testing.T) {
 	newDecode := initNewDecodeJson(t)
 	decodeJson := newDecode.(*DecodeJson)
-	inputString1 := "{\"unixTime\": 1645104030 }"
-	var in [][]byte
-	var out []config.GenericMap
-	out = decodeJson.Decode(in)
-	require.Equal(t, 0, len(out))
-	in = append(in, []byte(inputString1))
-	out = decodeJson.Decode(in)
-	require.Equal(t, len(out), 1)
-	require.Equal(t, float64(1645104030), out[0]["unixTime"])
+	out, err := decodeJson.Decode([]byte("{\"unixTime\": 1645104030 }"))
+	require.NoError(t, err)
+	require.Equal(t, float64(1645104030), out["unixTime"])
 }
diff --git a/pkg/pipeline/decode/decode_protobuf_test.go b/pkg/pipeline/decode/decode_protobuf_test.go
index f79d74e2..9cbe8920 100644
--- a/pkg/pipeline/decode/decode_protobuf_test.go
+++ b/pkg/pipeline/decode/decode_protobuf_test.go
@@ -45,10 +45,10 @@ func TestDecodeProtobuf(t *testing.T) {
 	rawPB, err := proto.Marshal(&flow)
 	require.NoError(t, err)
 
-	out := decoder.Decode([][]byte{rawPB})
-	require.Len(t, out, 1)
-	assert.NotZero(t, out[0]["TimeReceived"])
-	delete(out[0], "TimeReceived")
+	out, err := decoder.Decode(rawPB)
+	require.NoError(t, err)
+	assert.NotZero(t, out["TimeReceived"])
+	delete(out, "TimeReceived")
 	assert.Equal(t, config.GenericMap{
 		"FlowDirection":   1,
 		"Bytes":           uint64(456),
@@ -64,10 +64,10 @@ func TestDecodeProtobuf(t *testing.T) {
 		"TimeFlowStartMs": someTime.UnixMilli(),
 		"TimeFlowEndMs":   someTime.UnixMilli(),
 		"Interface":       "eth0",
-	}, out[0])
+	}, out)
 }
 
-func TestPBRecordsAsMaps(t *testing.T) {
+func TestPBFlowToMap(t *testing.T) {
 	someTime := time.Now()
 	flow := &pbflow.Record{
 		Interface:     "eth0",
@@ -96,10 +96,9 @@ func TestPBRecordsAsMaps(t *testing.T) {
 		},
 	}
 
-	out := PBRecordsAsMaps(&pbflow.Records{Entries: []*pbflow.Record{flow}})
-	require.Len(t, out, 1)
-	assert.NotZero(t, out[0]["TimeReceived"])
-	delete(out[0], "TimeReceived")
+	out := PBFlowToMap(flow)
+	assert.NotZero(t, out["TimeReceived"])
+	delete(out, "TimeReceived")
 	assert.Equal(t, config.GenericMap{
 		"FlowDirection":   1,
 		"Bytes":           uint64(456),
@@ -115,6 +114,6 @@ func TestPBRecordsAsMaps(t *testing.T) {
 		"TimeFlowStartMs": someTime.UnixMilli(),
 		"TimeFlowEndMs":   someTime.UnixMilli(),
 		"Interface":       "eth0",
-	}, out[0])
+	}, out)
 
 }
diff --git a/pkg/pipeline/encode/encode_kafka_test.go b/pkg/pipeline/encode/encode_kafka_test.go
index abec8eb6..721cfbc6 100644
--- a/pkg/pipeline/encode/encode_kafka_test.go
+++ b/pkg/pipeline/encode/encode_kafka_test.go
@@ -48,10 +48,10 @@ type fakeKafkaWriter struct {
 	mock.Mock
 }
 
-var receivedData []interface{}
+var receivedData []kafkago.Message
 
-func (f *fakeKafkaWriter) WriteMessages(ctx context.Context, msg ...kafkago.Message) error {
-	receivedData = append(receivedData, msg)
+func (f *fakeKafkaWriter) WriteMessages(_ context.Context, msg ...kafkago.Message) error {
+	receivedData = append(receivedData, msg...)
 	return nil
 }
 
@@ -73,24 +73,17 @@ func Test_EncodeKafka(t *testing.T) {
 	fw := fakeKafkaWriter{}
 	encodeKafka.kafkaWriter = &fw
 
-	receivedData = make([]interface{}, 0)
 	entry1 := test.GetExtractMockEntry()
 	entry2 := test.GetIngestMockEntry(false)
-	in := []config.GenericMap{entry1, entry2}
-	newEncode.Encode(in)
-	var expectedOutputString1 []byte
-	var expectedOutputString2 []byte
-	expectedOutputString1, _ = json.Marshal(entry1)
-	expectedOutputString2, _ = json.Marshal(entry2)
+	newEncode.Encode(entry1)
+	newEncode.Encode(entry2)
+	expectedOutputString1, _ := json.Marshal(entry1)
+	expectedOutputString2, _ := json.Marshal(entry2)
 	expectedOutput := []kafkago.Message{
-		{
-			Value: expectedOutputString1,
-		},
-		{
-			Value: expectedOutputString2,
-		},
+		{Value: expectedOutputString1},
+		{Value: expectedOutputString2},
 	}
-	require.Equal(t, expectedOutput, receivedData[0])
+	require.Equal(t, expectedOutput, receivedData)
 }
 
 func Test_TLSConfigEmpty(t *testing.T) {
diff --git a/pkg/pipeline/encode/encode_prom_test.go b/pkg/pipeline/encode/encode_prom_test.go
index 66698659..e6816c1a 100644
--- a/pkg/pipeline/encode/encode_prom_test.go
+++ b/pkg/pipeline/encode/encode_prom_test.go
@@ -117,8 +117,7 @@ func Test_NewEncodeProm(t *testing.T) {
 	expectedList = []string{"srcAddr", "dstAddr", "dstPort"}
 	require.Equal(t, encodeProm.counters[0].info.Labels, expectedList)
 	entry := test.GetExtractMockEntry()
-	input := []config.GenericMap{entry}
-	encodeProm.Encode(input)
+	encodeProm.Encode(entry)
 
 	// verify entries are in cache; one for the gauge and one for the counter
 	entriesMapLen := encodeProm.mCache.GetCacheLen()
@@ -199,8 +198,9 @@ func Test_CustomMetric(t *testing.T) {
 	encodeProm, cleanup, err := initPromWithServer(&params)
 	require.NoError(t, err)
 	defer cleanup()
-
-	encodeProm.Encode(metrics)
+	for _, metric := range metrics {
+		encodeProm.Encode(metric)
+	}
 	time.Sleep(100 * time.Millisecond)
 
 	exposed := test.ReadExposedMetrics(t)
@@ -257,8 +257,9 @@ func Test_MetricTTL(t *testing.T) {
 	require.NoError(t, err)
 	defer cleanup()
 
-	encodeProm.Encode(metrics)
-
+	for _, metric := range metrics {
+		encodeProm.Encode(metric)
+	}
 	exposed := test.ReadExposedMetrics(t)
 
 	require.Contains(t, exposed, `test_bytes_total{dstIP="10.0.0.1",srcIP="20.0.0.2"}`)
@@ -322,6 +323,8 @@ func BenchmarkPromEncode(b *testing.B) {
 	defer cleanup()
 
 	for i := 0; i < b.N; i++ {
-		prom.Encode(hundredFlows())
+		for _, metric := range hundredFlows() {
+			prom.Encode(metric)
+		}
 	}
 }
diff --git a/pkg/pipeline/encode/encode_test.go b/pkg/pipeline/encode/encode_test.go
index 6cd05dab..d85c1cdf 100644
--- a/pkg/pipeline/encode/encode_test.go
+++ b/pkg/pipeline/encode/encode_test.go
@@ -33,23 +33,11 @@ func initNewEncodeNone(t *testing.T) Encoder {
 func TestEncodeNone(t *testing.T) {
 	newEncode := initNewEncodeNone(t)
 	encodeNone := newEncode.(*encodeNone)
-	map1 := config.GenericMap{
+	in := config.GenericMap{
 		"varInt":    12,
 		"varString": "string1",
 		"varbool":   true,
 	}
-	map2 := config.GenericMap{
-		"varInt":    14,
-		"varString": "string2",
-		"varbool":   false,
-	}
-	map3 := config.GenericMap{}
-	var in []config.GenericMap
-	encodeNone.Encode(in)
-	require.Equal(t, 0, len(encodeNone.prevRecords))
-	in = append(in, map1)
-	in = append(in, map2)
-	in = append(in, map3)
 	encodeNone.Encode(in)
-	require.Equal(t, len(in), len(encodeNone.prevRecords))
+	require.Equal(t, in, encodeNone.prevRecord)
 }
diff --git a/pkg/pipeline/ingest/ingest_collector_test.go b/pkg/pipeline/ingest/ingest_collector_test.go
index 7b1e9436..ac8dc2dd 100644
--- a/pkg/pipeline/ingest/ingest_collector_test.go
+++ b/pkg/pipeline/ingest/ingest_collector_test.go
@@ -40,7 +40,7 @@ func TestIngest(t *testing.T) {
 	})
 	ic, err := NewIngestCollector(operational.NewMetrics(&config.MetricsSettings{}), stage.GetStageParams()[0])
 	require.NoError(t, err)
-	forwarded := make(chan []config.GenericMap)
+	forwarded := make(chan config.GenericMap)
 
 	// GIVEN an IPFIX collector Ingester
 	go ic.Ingest(forwarded)
@@ -48,9 +48,8 @@ func TestIngest(t *testing.T) {
 	client, err := test.NewIPFIXClient(collectorPort)
 	require.NoError(t, err)
 
-	received := waitForFlows(t, client, forwarded)
-	require.NotEmpty(t, received)
-	flow := received[0]
+	flow := waitForFlow(t, client, forwarded)
+	require.NotEmpty(t, flow)
 	assert.EqualValues(t, 12345678, flow["TimeFlowStart"])
 	assert.EqualValues(t, 12345678, flow["TimeFlowEnd"])
 	assert.Equal(t, "1.2.3.4", flow["SrcAddr"])
@@ -58,7 +57,7 @@ func TestIngest(t *testing.T) {
 
 // The IPFIX client might send information before the Ingester is actually listening,
 // so we might need to repeat the submission until the ingest starts forwarding logs
-func waitForFlows(t *testing.T, client *test.IPFIXClient, forwarded chan []config.GenericMap) []config.GenericMap {
+func waitForFlow(t *testing.T, client *test.IPFIXClient, forwarded chan config.GenericMap) config.GenericMap {
 	var start = time.Now()
 	for {
 		if client.SendTemplate() == nil &&
diff --git a/pkg/pipeline/ingest/ingest_kafka_test.go b/pkg/pipeline/ingest/ingest_kafka_test.go
index 365dcec9..7adcddf7 100644
--- a/pkg/pipeline/ingest/ingest_kafka_test.go
+++ b/pkg/pipeline/ingest/ingest_kafka_test.go
@@ -108,16 +108,14 @@ func Test_NewIngestKafka2(t *testing.T) {
 	require.Equal(t, time.Duration(1000)*time.Millisecond, ingestKafka.kafkaReader.Config().CommitInterval)
 }
 
-func removeTimestamp(receivedEntries []config.GenericMap) {
-	for _, entry := range receivedEntries {
-		delete(entry, "TimeReceived")
-	}
+func removeTimestamp(receivedEntry config.GenericMap) {
+	delete(receivedEntry, "TimeReceived")
 }
 
 func Test_IngestKafka(t *testing.T) {
 	newIngest := initNewIngestKafka(t, testConfig1)
 	ingestKafka := newIngest.(*ingestKafka)
-	ingestOutput := make(chan []config.GenericMap)
+	ingestOutput := make(chan config.GenericMap)
 
 	// run Ingest in a separate thread
 	go func() {
@@ -137,16 +135,22 @@ func Test_IngestKafka(t *testing.T) {
 	inChan <- []byte(record3)
 
 	// wait for the data to have been processed
-	receivedEntries := <-ingestOutput
-
+	receivedEntry, err := test.WaitFromChannel(ingestOutput, timeout)
+	require.NoError(t, err)
 	// we remove timestamp for test stability
 	// Timereceived field is tested in the decodeJson tests
-	removeTimestamp(receivedEntries)
+	removeTimestamp(receivedEntry)
+	require.Equal(t, test.DeserializeJSONToMap(t, record1), receivedEntry)
+
+	receivedEntry, err = test.WaitFromChannel(ingestOutput, timeout)
+	require.NoError(t, err)
+	removeTimestamp(receivedEntry)
+	require.Equal(t, test.DeserializeJSONToMap(t, record2), receivedEntry)
 
-	require.Equal(t, 3, len(receivedEntries))
-	require.Equal(t, test.DeserializeJSONToMap(t, record1), receivedEntries[0])
-	require.Equal(t, test.DeserializeJSONToMap(t, record2), receivedEntries[1])
-	require.Equal(t, test.DeserializeJSONToMap(t, record3), receivedEntries[2])
+	receivedEntry, err = test.WaitFromChannel(ingestOutput, timeout)
+	require.NoError(t, err)
+	removeTimestamp(receivedEntry)
+	require.Equal(t, test.DeserializeJSONToMap(t, record3), receivedEntry)
 }
 
 type fakeKafkaReader struct {
@@ -177,7 +181,7 @@ func (f *fakeKafkaReader) Config() kafkago.ReaderConfig {
 }
 
 func Test_KafkaListener(t *testing.T) {
-	ingestOutput := make(chan []config.GenericMap)
+	ingestOutput := make(chan config.GenericMap)
 	newIngest := initNewIngestKafka(t, testConfig1)
 	ingestKafka := newIngest.(*ingestKafka)
 
@@ -191,69 +195,14 @@ func Test_KafkaListener(t *testing.T) {
 	}()
 
 	// wait for the data to have been processed
-	receivedEntries, err := test.WaitFromChannel(ingestOutput, 2*time.Second)
+	receivedEntry, err := test.WaitFromChannel(ingestOutput, 2*time.Second)
 	require.NoError(t, err)
 
 	// we remove timestamp for test stability
 	// Timereceived field is tested in the decodeJson tests
-	removeTimestamp(receivedEntries)
-
-	require.Equal(t, 1, len(receivedEntries))
-	require.Equal(t, test.DeserializeJSONToMap(t, string(fakeRecord)), receivedEntries[0])
-}
-
-func Test_MaxBatchLength(t *testing.T) {
-	ingestOutput := make(chan []config.GenericMap)
-	newIngest := initNewIngestKafka(t, testConfig1)
-	ingestKafka := newIngest.(*ingestKafka)
-
-	// change the ReadMessage function to the mock-up
-	fr := fakeKafkaReader{readToDo: 15}
-	ingestKafka.kafkaReader = &fr
-	ingestKafka.batchMaxLength = 10
-	ingestKafka.batchReadTimeout = 10000
-
-	// run Ingest in a separate thread
-	go func() {
-		ingestKafka.Ingest(ingestOutput)
-	}()
-
-	// wait for the data to have been processed
-	receivedEntries, err := test.WaitFromChannel(ingestOutput, 2*time.Second)
-	require.NoError(t, err)
-
-	require.Equal(t, 10, len(receivedEntries))
-}
-
-func Test_BatchTimeout(t *testing.T) {
-	ingestOutput := make(chan []config.GenericMap)
-	newIngest := initNewIngestKafka(t, testConfig1)
-	ingestKafka := newIngest.(*ingestKafka)
-
-	// change the ReadMessage function to the mock-up
-	fr := fakeKafkaReader{readToDo: 5}
-	ingestKafka.kafkaReader = &fr
-	ingestKafka.batchMaxLength = 1000
-	ingestKafka.batchReadTimeout = 100
-
-	beforeIngest := time.Now()
-	// run Ingest in a separate thread
-	go func() {
-		ingestKafka.Ingest(ingestOutput)
-	}()
-
-	require.Equal(t, 0, len(ingestOutput))
-	// wait for the data to have been processed
-	receivedEntries, err := test.WaitFromChannel(ingestOutput, 2*time.Second)
-	require.NoError(t, err)
-	require.Equal(t, 5, len(receivedEntries))
-
-	afterIngest := time.Now()
+	removeTimestamp(receivedEntry)
 
-	// We check that we get entries because of the timer
-	// Time must be above timer value but not too much, 20ms is our margin here
-	require.LessOrEqual(t, int64(100), afterIngest.Sub(beforeIngest).Milliseconds())
-	require.Greater(t, int64(120), afterIngest.Sub(beforeIngest).Milliseconds())
+	require.Equal(t, test.DeserializeJSONToMap(t, string(fakeRecord)), receivedEntry)
 }
 
 func Test_TLSConfigEmpty(t *testing.T) {
diff --git a/pkg/pipeline/pipeline_test.go b/pkg/pipeline/pipeline_test.go
index 34ee5d24..cecadff7 100644
--- a/pkg/pipeline/pipeline_test.go
+++ b/pkg/pipeline/pipeline_test.go
@@ -30,7 +30,6 @@ import (
 	test2 "github.com/mariomac/guara/pkg/test"
 	"github.com/netobserv/flowlogs-pipeline/pkg/config"
 	"github.com/netobserv/flowlogs-pipeline/pkg/operational"
-	"github.com/netobserv/flowlogs-pipeline/pkg/pipeline/ingest"
 	"github.com/netobserv/flowlogs-pipeline/pkg/pipeline/transform"
 	"github.com/netobserv/flowlogs-pipeline/pkg/pipeline/write"
 	"github.com/netobserv/flowlogs-pipeline/pkg/test"
@@ -54,12 +53,10 @@ parameters:
 `
 
 func Test_transformToLoki(t *testing.T) {
-	var transformed []config.GenericMap
-	input := []config.GenericMap{{"key": "value"}}
+	input := config.GenericMap{"key": "value"}
 	transform, err := transform.NewTransformNone()
 	require.NoError(t, err)
-	transformed = append(transformed, transform.Transform(input)...)
-
+	transformed, _ := transform.Transform(input)
 	v, cfg := test.InitConfig(t, yamlConfigNoParams)
 	require.NotNil(t, v)
 	loki, err := write.NewWriteLoki(operational.NewMetrics(&config.MetricsSettings{}), cfg.Parameters[0])
@@ -118,19 +115,19 @@ func Test_SimplePipeline(t *testing.T) {
 	// So we don't need to run it in a separate go-routine
 	mainPipeline.Run()
 	// What is there left to check? Check length of saved data of each stage in private structure.
-	ingester := mainPipeline.pipelineStages[0].Ingester.(*ingest.IngestFile)
 	writer := mainPipeline.pipelineStages[2].Writer.(*write.WriteNone)
-	require.Equal(t, len(ingester.PrevRecords), len(writer.PrevRecords))
 
 	// values checked from the first line of the ../../hack/examples/ocp-ipfix-flowlogs.json file
-	require.Equal(t, config.GenericMap{
-		"flp_bytes":   float64(20800),
-		"flp_dstAddr": "10.130.2.2",
-		"flp_dstPort": float64(36936),
-		"flp_packets": float64(400),
-		"flp_srcAddr": "10.130.2.13",
-		"flp_srcPort": float64(3100),
-	}, writer.PrevRecords[0])
+	test2.Eventually(t, 30*time.Second, func(t require.TestingT) {
+		require.Contains(t, writer.PrevRecords(), config.GenericMap{
+			"flp_bytes":   float64(20800),
+			"flp_dstAddr": "10.130.2.2",
+			"flp_dstPort": float64(36936),
+			"flp_packets": float64(400),
+			"flp_srcAddr": "10.130.2.13",
+			"flp_srcPort": float64(3100),
+		})
+	})
 }
 
 func TestGRPCProtobuf(t *testing.T) {
diff --git a/pkg/pipeline/transform/transform_filter_test.go b/pkg/pipeline/transform/transform_filter_test.go
index 45b581ae..7343f902 100644
--- a/pkg/pipeline/transform/transform_filter_test.go
+++ b/pkg/pipeline/transform/transform_filter_test.go
@@ -121,9 +121,10 @@ func TestNewTransformFilterRemoveField(t *testing.T) {
 	require.Len(t, transformFilter.Rules, 2)
 
 	input := test.GetIngestMockEntry(false)
-	output := transformFilter.Transform([]config.GenericMap{input})
+	output, ok := transformFilter.Transform(input)
+	require.True(t, ok)
 	expectedOutput := getFilterExpectedOutput()
-	require.Equal(t, expectedOutput, output[0])
+	require.Equal(t, expectedOutput, output)
 }
 
 func TestNewTransformFilterRemoveEntryIfExists(t *testing.T) {
@@ -132,8 +133,8 @@ func TestNewTransformFilterRemoveEntryIfExists(t *testing.T) {
 	require.Len(t, transformFilter.Rules, 1)
 
 	input := test.GetIngestMockEntry(false)
-	output := transformFilter.Transform([]config.GenericMap{input})
-	require.Equal(t, output, []config.GenericMap{})
+	_, ok := transformFilter.Transform(input)
+	require.False(t, ok)
 }
 
 func TestNewTransformFilterRemoveEntryIfDoesntExists(t *testing.T) {
@@ -142,8 +143,8 @@ func TestNewTransformFilterRemoveEntryIfDoesntExists(t *testing.T) {
 	require.Len(t, transformFilter.Rules, 1)
 
 	input := test.GetIngestMockEntry(false)
-	output := transformFilter.Transform([]config.GenericMap{input})
-	require.Equal(t, output, []config.GenericMap{})
+	_, ok := transformFilter.Transform(input)
+	require.False(t, ok)
 }
 
 func TestNewTransformFilterRemoveEntryIfEqual(t *testing.T) {
@@ -153,18 +154,17 @@ func TestNewTransformFilterRemoveEntryIfEqual(t *testing.T) {
 
 	input := test.GetIngestMockEntry(false)
 
-	output := transformFilter.Transform([]config.GenericMap{input})
-	require.Equal(t, 0, len(output))
+	_, ok := transformFilter.Transform(input)
+	require.False(t, ok)
 
 	input["message"] = "dummy message"
-	output = transformFilter.Transform([]config.GenericMap{input})
-	require.Equal(t, 1, len(output))
-	require.Contains(t, output[0], "message")
-	require.Equal(t, output[0]["message"], "dummy message")
+	output, ok := transformFilter.Transform(input)
+	require.True(t, ok)
+	require.Equal(t, output["message"], "dummy message")
 
 	input["value"] = 8.0
-	output = transformFilter.Transform([]config.GenericMap{input})
-	require.Equal(t, 0, len(output))
+	_, ok = transformFilter.Transform(input)
+	require.False(t, ok)
 }
 
 func TestNewTransformFilterRemoveEntryIfNotEqual(t *testing.T) {
@@ -174,14 +174,13 @@ func TestNewTransformFilterRemoveEntryIfNotEqual(t *testing.T) {
 
 	input := test.GetIngestMockEntry(false)
 
-	output := transformFilter.Transform([]config.GenericMap{input})
-	require.Equal(t, 1, len(output))
-	require.Contains(t, output[0], "message")
-	require.Equal(t, output[0]["message"], "test message")
+	output, ok := transformFilter.Transform(input)
+	require.True(t, ok)
+	require.Equal(t, output["message"], "test message")
 
 	input["message"] = "dummy message"
-	output = transformFilter.Transform([]config.GenericMap{input})
-	require.Equal(t, 0, len(output))
+	_, ok = transformFilter.Transform(input)
+	require.False(t, ok)
 }
 
 func InitNewTransformFilter(t *testing.T, configFile string) Transformer {
diff --git a/pkg/pipeline/transform/transform_generic_test.go b/pkg/pipeline/transform/transform_generic_test.go
index 3e989cda..85c1d145 100644
--- a/pkg/pipeline/transform/transform_generic_test.go
+++ b/pkg/pipeline/transform/transform_generic_test.go
@@ -113,9 +113,10 @@ func TestNewTransformGenericMaintainFalse(t *testing.T) {
 	require.Len(t, transformGeneric.rules, 6)
 
 	input := test.GetIngestMockEntry(false)
-	output := transformGeneric.Transform([]config.GenericMap{input})
+	output, ok := transformGeneric.Transform(input)
+	require.True(t, ok)
 	expectedOutput := getGenericExpectedOutputShort()
-	require.Equal(t, expectedOutput, output[0])
+	require.Equal(t, expectedOutput, output)
 }
 
 func TestNewTransformGenericMaintainTrue(t *testing.T) {
@@ -124,9 +125,10 @@ func TestNewTransformGenericMaintainTrue(t *testing.T) {
 	require.Len(t, transformGeneric.rules, 6)
 
 	input := test.GetIngestMockEntry(false)
-	output := transformGeneric.Transform([]config.GenericMap{input})
+	output, ok := transformGeneric.Transform(input)
+	require.True(t, ok)
 	expectedOutput := getGenericExpectedOutputLong()
-	require.Equal(t, expectedOutput, output[0])
+	require.Equal(t, expectedOutput, output)
 }
 
 func InitNewTransformGeneric(t *testing.T, configFile string) Transformer {
diff --git a/pkg/pipeline/transform/transform_network_test.go b/pkg/pipeline/transform/transform_network_test.go
index 9efd2b41..0eea5371 100644
--- a/pkg/pipeline/transform/transform_network_test.go
+++ b/pkg/pipeline/transform/transform_network_test.go
@@ -145,9 +145,9 @@ func Test_Transform(t *testing.T) {
 	err := location.InitLocationDB()
 	require.NoError(t, err)
 
-	output := networkTransform.Transform([]config.GenericMap{entry})
-
-	require.Equal(t, expectedOutput, output[0])
+	output, ok := networkTransform.Transform(entry)
+	require.True(t, ok)
+	require.Equal(t, expectedOutput, output)
 }
 
 func Test_TransformAddSubnetParseCIDRFailure(t *testing.T) {
@@ -169,9 +169,9 @@ func Test_TransformAddSubnetParseCIDRFailure(t *testing.T) {
 	err := location.InitLocationDB()
 	require.NoError(t, err)
 
-	output := networkTransform.Transform([]config.GenericMap{entry})
-
-	require.Equal(t, expectedOutput, output[0])
+	output, ok := networkTransform.Transform(entry)
+	require.True(t, ok)
+	require.Equal(t, expectedOutput, output)
 }
 
 func Test_NewTransformNetwork(t *testing.T) {
@@ -199,10 +199,11 @@ parameters:
 	require.NotNil(t, newNetworkTransform)
 
 	entry := test.GetIngestMockEntry(false)
-	output := newNetworkTransform.Transform([]config.GenericMap{entry})
+	output, ok := newNetworkTransform.Transform(entry)
+	require.True(t, ok)
 
-	require.Equal(t, "10.0.0.1", output[0]["srcIP"])
-	require.Equal(t, "10.0.0.0/24", output[0]["subnetSrcIP"])
+	require.Equal(t, "10.0.0.1", output["srcIP"])
+	require.Equal(t, "10.0.0.0/24", output["subnetSrcIP"])
 }
 
 func InitNewTransformNetwork(t *testing.T, configFile string) Transformer {
@@ -247,12 +248,13 @@ parameters:
 	require.NotNil(t, newNetworkTransform)
 
 	entry := test.GetIngestMockEntry(false)
-	output := newNetworkTransform.Transform([]config.GenericMap{entry})
+	output, ok := newNetworkTransform.Transform(entry)
+	require.True(t, ok)
 
-	require.Equal(t, "10.0.0.1", output[0]["srcIP"])
-	require.Equal(t, "10.0.0.0/24", output[0]["subnetSrcIP"])
-	require.Equal(t, "10.0.0.0/24", output[0]["match-10.0.*"])
-	require.NotEqual(t, "10.0.0.0/24", output[0]["match-11.0.*"])
+	require.Equal(t, "10.0.0.1", output["srcIP"])
+	require.Equal(t, "10.0.0.0/24", output["subnetSrcIP"])
+	require.Equal(t, "10.0.0.0/24", output["match-10.0.*"])
+	require.NotEqual(t, "10.0.0.0/24", output["match-11.0.*"])
 }
 
 func Test_Transform_AddIfScientificNotation(t *testing.T) {
@@ -294,28 +296,32 @@ func Test_Transform_AddIfScientificNotation(t *testing.T) {
 	entry = config.GenericMap{
 		"value": 1.2345e67,
 	}
-	output := newNetworkTransform.Transform([]config.GenericMap{entry})
-	require.Equal(t, true, output[0]["bigger_than_10_Evaluate"])
-	require.Equal(t, 1.2345e67, output[0]["bigger_than_10"])
+	output, ok := newNetworkTransform.Transform(entry)
+	require.True(t, ok)
+	require.Equal(t, true, output["bigger_than_10_Evaluate"])
+	require.Equal(t, 1.2345e67, output["bigger_than_10"])
 
 	entry = config.GenericMap{
 		"value": 1.2345e-67,
 	}
-	output = newNetworkTransform.Transform([]config.GenericMap{entry})
-	require.Equal(t, true, output[0]["smaller_than_10_Evaluate"])
-	require.Equal(t, 1.2345e-67, output[0]["smaller_than_10"])
+	output, ok = newNetworkTransform.Transform(entry)
+	require.True(t, ok)
+	require.Equal(t, true, output["smaller_than_10_Evaluate"])
+	require.Equal(t, 1.2345e-67, output["smaller_than_10"])
 
 	entry = config.GenericMap{
 		"value": 1,
 	}
-	output = newNetworkTransform.Transform([]config.GenericMap{entry})
-	require.Equal(t, true, output[0]["dir_Evaluate"])
-	require.Equal(t, "in", output[0]["dir"])
+	output, ok = newNetworkTransform.Transform(entry)
+	require.True(t, ok)
+	require.Equal(t, true, output["dir_Evaluate"])
+	require.Equal(t, "in", output["dir"])
 
 	entry = config.GenericMap{
 		"value": 0,
 	}
-	output = newNetworkTransform.Transform([]config.GenericMap{entry})
-	require.Equal(t, true, output[0]["dir_Evaluate"])
-	require.Equal(t, "out", output[0]["dir"])
+	output, ok = newNetworkTransform.Transform(entry)
+	require.True(t, ok)
+	require.Equal(t, true, output["dir_Evaluate"])
+	require.Equal(t, "out", output["dir"])
 }
diff --git a/pkg/pipeline/write/write_loki_test.go b/pkg/pipeline/write/write_loki_test.go
index 7dc4504d..2f76669b 100644
--- a/pkg/pipeline/write/write_loki_test.go
+++ b/pkg/pipeline/write/write_loki_test.go
@@ -333,7 +333,10 @@ func BenchmarkWriteLoki(b *testing.B) {
 	loki, err := NewWriteLoki(operational.NewMetrics(&config.MetricsSettings{}), config.StageParam{Write: &config.Write{Loki: &params}})
 	require.NoError(b, err)
 
+	hf := hundredFlows()
 	for i := 0; i < b.N; i++ {
-		loki.Write(hundredFlows())
+		for _, f := range hf {
+			loki.Write(f)
+		}
 	}
 }
diff --git a/pkg/pipeline/write/write_stdout_test.go b/pkg/pipeline/write/write_stdout_test.go
index c8ca8f6a..254d87a2 100644
--- a/pkg/pipeline/write/write_stdout_test.go
+++ b/pkg/pipeline/write/write_stdout_test.go
@@ -26,7 +26,7 @@ import (
 
 func Test_WriteStdout(t *testing.T) {
 	ws := writeStdout{}
-	ws.Write([]config.GenericMap{{"key": "test"}})
+	ws.Write(config.GenericMap{"key": "test"})
 }
 
 func Test_NewWriteStdout(t *testing.T) {
diff --git a/pkg/pipeline/write/write_test.go b/pkg/pipeline/write/write_test.go
index 28dc806b..afcdc140 100644
--- a/pkg/pipeline/write/write_test.go
+++ b/pkg/pipeline/write/write_test.go
@@ -26,7 +26,7 @@ import (
 
 func Test_Write(t *testing.T) {
 	wn := WriteNone{}
-	wn.Write([]config.GenericMap{{"key": "test"}})
+	wn.Write(config.GenericMap{"key": "test"})
 }
 
 func Test_NewWriteNone(t *testing.T) {
diff --git a/pkg/test/e2e/kafka/kafka_suite_test.go b/pkg/test/e2e/kafka/kafka_suite_test.go
index 754f3d9b..e9a35c43 100644
--- a/pkg/test/e2e/kafka/kafka_suite_test.go
+++ b/pkg/test/e2e/kafka/kafka_suite_test.go
@@ -18,10 +18,13 @@
 package e2e
 
 import (
+	"context"
 	"testing"
+	"time"
 
 	"github.com/netobserv/flowlogs-pipeline/pkg/test/e2e"
 	"sigs.k8s.io/e2e-framework/pkg/env"
+	"sigs.k8s.io/e2e-framework/pkg/envconf"
 )
 
 var TestEnv env.Environment
@@ -29,6 +32,11 @@ var TestEnv env.Environment
 var manifestDeployDefinitions = e2e.ManifestDeployDefinitions{
 	e2e.ManifestDeployDefinition{
 		YamlFile: "strimzi-cluster-operator-0.31.0.yaml",
+		PostFunction: func(ctx context.Context, cfg *envconf.Config, namespace string) error {
+			// Wait few seconds to allow the CRD to be registered so the next step does not fail
+			time.Sleep(5 * time.Second)
+			return nil
+		},
 	},
 	e2e.ManifestDeployDefinition{
 		YamlFile:     "kafka.strimzi.yaml",
diff --git a/pkg/test/utils.go b/pkg/test/utils.go
index 174b204e..5b55f704 100644
--- a/pkg/test/utils.go
+++ b/pkg/test/utils.go
@@ -167,7 +167,7 @@ func DumpToTemp(content string) (string, error, func()) {
 	}
 }
 
-func WaitFromChannel(in chan []config.GenericMap, timeout time.Duration) ([]config.GenericMap, error) {
+func WaitFromChannel(in chan config.GenericMap, timeout time.Duration) (config.GenericMap, error) {
 	timeoutReached := time.NewTicker(timeout)
 	select {
 	case record := <-in: