Repository: netobserv/flowlogs-pipeline
Pull request #318 (2022-09-30 15:18:36)

disable CGO
CGO it's not required and usually makes executables more resource-consuming. It also adds some extra requirements to the build/test pipeline (e.g. having to have GCC + libraries, etc...)

Reference read: https://dave.cheney.net/2016/01/18/cgo-is-not-go

codecov-commenter: # [Codecov](https://codecov.io/gh/netobserv/flowlogs-pipeline/pull/318?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=netobserv) Report
> Merging [#318](https://codecov.io/gh/netobserv/flowlogs-pipeline/pull/318?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=netobserv) (0c89d00) into [main](https://codecov.io/gh/netobserv/flowlogs-pipeline/commit/8cc4bc74b46c7ee11c34b12b7ccdca7f5dcf2ee8?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=netobserv) (8cc4bc7) will **not change** coverage.
> The diff coverage is `n/a`.

```diff
@@           Coverage Diff           @@
##             main     #318   +/-   ##
=======================================
  Coverage   68.89%   68.89%           
=======================================
  Files          84       84           
  Lines        4900     4900           
=======================================
  Hits         3376     3376           
  Misses       1308     1308           
  Partials      216      216           
```

| Flag | Coverage Δ | |
|---|---|---|
| unittests | `68.89% <ø> (ø)` | |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=netobserv#carryforward-flags-in-the-pull-request-comment) to find out more.


Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=netobserv). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=netobserv)


Patch:
diff --git a/Makefile b/Makefile
index ac44a764..987ce615 100644
--- a/Makefile
+++ b/Makefile
@@ -9,7 +9,6 @@ export GO111MODULE=on
 export CGO_ENABLED=0
 export GOOS=linux
 export GOARCH=amd64
-export CGO_ENABLED=1
 
 SHELL := /bin/bash
 DOCKER_TAG ?= latest
@@ -100,7 +99,8 @@ clean: ## Clean
 TEST_OPTS := -race -coverpkg=./... -covermode=atomic -coverprofile=/tmp/coverage.out
 .PHONY: tests-unit
 tests-unit: validate_go ## Unit tests
-	go test -p 1 $(TEST_OPTS) $$(go list ./... | grep -v /e2e)
+	# enabling CGO is required for -race flag
+	CGO_ENABLED=1 go test -p 1 $(TEST_OPTS) $$(go list ./... | grep -v /e2e)
 
 .PHONY: tests-fast
 tests-fast: TEST_OPTS=
@@ -108,7 +108,7 @@ tests-fast: tests-unit ## Fast unit tests (no race tests / coverage)
 
 .PHONY: tests-e2e
 tests-e2e: validate_go $(KIND)  ## End-to-end tests
-	go test -p 1 -v -timeout 20m -race $$(go list ./... | grep  /e2e)
+	go test -p 1 -v -timeout 20m $$(go list ./... | grep  /e2e)
 
 .PHONY: tests-all
 tests-all: validate_go tests-unit tests-e2e ## All tests
diff --git a/docs/operational-metrics.md b/docs/operational-metrics.md
index 161042ee..5a7e0b78 100644
--- a/docs/operational-metrics.md
+++ b/docs/operational-metrics.md
@@ -39,14 +39,38 @@ Each table below provides documentation for an exported flowlogs-pipeline operat
 | **Labels** | error, metric, key | 
 
 
+### ingest_batch_size_bytes
+| **Name** | ingest_batch_size_bytes | 
+|:---|:---|
+| **Description** | Ingested batch size distribution, in bytes | 
+| **Type** | summary | 
+| **Labels** | stage | 
+
+
+### ingest_errors
+| **Name** | ingest_errors | 
+|:---|:---|
+| **Description** | Counter of errors during ingestion | 
+| **Type** | counter | 
+| **Labels** | stage, type, code | 
+
+
 ### ingest_flows_processed
 | **Name** | ingest_flows_processed | 
 |:---|:---|
-| **Description** | Number of flow logs processed by the ingester | 
+| **Description** | Number of flows received by the ingester | 
 | **Type** | counter | 
 | **Labels** | stage | 
 
 
+### ingest_latency_ms
+| **Name** | ingest_latency_ms | 
+|:---|:---|
+| **Description** | Latency between flow end time and ingest time, in milliseconds | 
+| **Type** | histogram | 
+| **Labels** | stage | 
+
+
 ### metrics_processed
 | **Name** | metrics_processed | 
 |:---|:---|
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/operational/metrics.go b/pkg/operational/metrics.go
index f3fe22f3..ccf16b74 100644
--- a/pkg/operational/metrics.go
+++ b/pkg/operational/metrics.go
@@ -36,9 +36,10 @@ type MetricDefinition struct {
 
 type metricType string
 
-const TypeCounter = "counter"
-const TypeGauge = "gauge"
-const TypeHistogram = "histogram"
+const TypeCounter metricType = "counter"
+const TypeGauge metricType = "gauge"
+const TypeHistogram metricType = "histogram"
+const TypeSummary metricType = "summary"
 
 var allMetrics = []MetricDefinition{}
 
@@ -54,12 +55,6 @@ func DefineMetric(name, help string, t metricType, labels ...string) MetricDefin
 }
 
 var (
-	ingestFlowsProcessed = DefineMetric(
-		"ingest_flows_processed",
-		"Number of flow logs processed by the ingester",
-		TypeCounter,
-		"stage",
-	)
 	recordsWritten = DefineMetric(
 		"records_written",
 		"Number of output records written",
@@ -78,6 +73,12 @@ var (
 		TypeGauge,
 		"stage",
 	)
+	stageDuration = DefineMetric(
+		"stage_duration_ms",
+		"Pipeline stage duration in milliseconds",
+		TypeHistogram,
+		"stage",
+	)
 )
 
 func (def *MetricDefinition) mapLabels(labels []string) prometheus.Labels {
@@ -92,7 +93,8 @@ func (def *MetricDefinition) mapLabels(labels []string) prometheus.Labels {
 }
 
 type Metrics struct {
-	settings *config.MetricsSettings
+	settings           *config.MetricsSettings
+	stageDurationHisto *prometheus.HistogramVec
 }
 
 func NewMetrics(settings *config.MetricsSettings) *Metrics {
@@ -153,6 +155,18 @@ func (o *Metrics) NewGaugeFunc(def *MetricDefinition, f func() float64, labels .
 	o.register(c, fullName)
 }
 
+func (o *Metrics) NewHistogram(def *MetricDefinition, buckets []float64, labels ...string) prometheus.Histogram {
+	fullName := o.settings.Prefix + def.Name
+	c := prometheus.NewHistogram(prometheus.HistogramOpts{
+		Name:        fullName,
+		Help:        def.Help,
+		Buckets:     buckets,
+		ConstLabels: def.mapLabels(labels),
+	})
+	o.register(c, fullName)
+	return c
+}
+
 func (o *Metrics) NewHistogramVec(def *MetricDefinition, buckets []float64) *prometheus.HistogramVec {
 	fullName := o.settings.Prefix + def.Name
 	c := prometheus.NewHistogramVec(prometheus.HistogramOpts{
@@ -164,8 +178,20 @@ func (o *Metrics) NewHistogramVec(def *MetricDefinition, buckets []float64) *pro
 	return c
 }
 
-func (o *Metrics) CreateFlowsProcessedCounter(stage string) prometheus.Counter {
-	return o.NewCounter(&ingestFlowsProcessed, stage)
+func (o *Metrics) NewSummary(def *MetricDefinition, labels ...string) prometheus.Summary {
+	fullName := o.settings.Prefix + def.Name
+	c := prometheus.NewSummary(prometheus.SummaryOpts{
+		Name:        fullName,
+		Help:        def.Help,
+		ConstLabels: def.mapLabels(labels),
+		// arbitrary objectives for now
+		Objectives: map[float64]float64{
+			0.5:  0.02,
+			0.95: 0.01,
+		},
+	})
+	o.register(c, fullName)
+	return c
 }
 
 func (o *Metrics) CreateRecordsWrittenCounter(stage string) prometheus.Counter {
@@ -180,6 +206,13 @@ func (o *Metrics) CreateOutQueueSizeGauge(stage string, f func() int) {
 	o.NewGaugeFunc(&stageOutQueueSize, func() float64 { return float64(f()) }, stage)
 }
 
+func (o *Metrics) GetOrCreateStageDurationHisto() *prometheus.HistogramVec {
+	if o.stageDurationHisto == nil {
+		o.stageDurationHisto = o.NewHistogramVec(&stageDuration, []float64{.001, .01, .1, 1, 10, 100, 1000, 10000})
+	}
+	return o.stageDurationHisto
+}
+
 func GetDocumentation() string {
 	doc := ""
 	sort.Slice(allMetrics, func(i, j int) bool {
diff --git a/pkg/operational/timer.go b/pkg/operational/timer.go
new file mode 100644
index 00000000..6862326b
--- /dev/null
+++ b/pkg/operational/timer.go
@@ -0,0 +1,51 @@
+package operational
+
+import (
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+type Timer struct {
+	startTime *time.Time
+	observer  prometheus.Observer
+}
+
+func NewTimer(o prometheus.Observer) *Timer {
+	return &Timer{
+		observer: o,
+	}
+}
+
+// Start starts or restarts the timer, regardless if a previous call occurred without being observed first
+func (t *Timer) Start() time.Time {
+	now := time.Now()
+	t.startTime = &now
+	return now
+}
+
+// StartOnce starts the timer just the first time. Subsequent calls will be ignored,
+// until the timer is observed
+func (t *Timer) StartOnce() time.Time {
+	if t.startTime == nil {
+		now := time.Now()
+		t.startTime = &now
+	}
+	return *t.startTime
+}
+
+func (t *Timer) ObserveMilliseconds() {
+	t.observe(func(d time.Duration) float64 { return float64(d.Milliseconds()) })
+}
+
+func (t *Timer) ObserveSeconds() {
+	t.observe(time.Duration.Seconds)
+}
+
+func (t *Timer) observe(f func(d time.Duration) float64) {
+	if t.startTime != nil {
+		duration := time.Since(*t.startTime)
+		t.observer.Observe(f(duration))
+		t.startTime = nil
+	}
+}
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 2ab52901..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,24 +97,24 @@ func (w *TransportWrapper) Send(_, data []byte) error {
 }
 
 // Ingest ingests entries from a network collector using goflow2 library (https://github.com/netsampler/goflow2)
-func (ingestC *ingestCollector) Ingest(out chan<- []config.GenericMap) {
+func (c *ingestCollector) Ingest(out chan<- config.GenericMap) {
 	ctx := context.Background()
-	ingestC.metrics.createOutQueueLen(out)
+	c.metrics.createOutQueueLen(out)
 
 	// initialize background listeners (a.k.a.netflow+legacy collector)
-	ingestC.initCollectorListener(ctx)
+	c.initCollectorListener(ctx)
 
 	// forever process log lines received by collector
-	ingestC.processLogLines(out)
+	c.processLogLines(out)
 }
 
-func (ingestC *ingestCollector) initCollectorListener(ctx context.Context) {
-	transporter := NewWrapper(ingestC.in)
+func (c *ingestCollector) initCollectorListener(ctx context.Context) {
+	transporter := NewWrapper(c.in)
 	formatter, err := goflowFormat.FindFormat(ctx, "pb")
 	if err != nil {
 		log.Fatal(err)
 	}
-	if ingestC.port > 0 {
+	if c.port > 0 {
 		go func() {
 			sNF := &utils.StateNetFlow{
 				Format:    formatter,
@@ -127,13 +122,13 @@ func (ingestC *ingestCollector) initCollectorListener(ctx context.Context) {
 				Logger:    log.StandardLogger(),
 			}
 
-			log.Infof("listening for netflow on host %s, port = %d", ingestC.hostname, ingestC.port)
-			err = sNF.FlowRoutine(1, ingestC.hostname, ingestC.port, false)
+			log.Infof("listening for netflow on host %s, port = %d", c.hostname, c.port)
+			err = sNF.FlowRoutine(1, c.hostname, c.port, false)
 			log.Fatal(err)
 		}()
 	}
 
-	if ingestC.portLegacy > 0 {
+	if c.portLegacy > 0 {
 		go func() {
 			sLegacyNF := &utils.StateNFLegacy{
 				Format:    formatter,
@@ -141,47 +136,21 @@ func (ingestC *ingestCollector) initCollectorListener(ctx context.Context) {
 				Logger:    log.StandardLogger(),
 			}
 
-			log.Infof("listening for legacy netflow on host %s, port = %d", ingestC.hostname, ingestC.portLegacy)
-			err = sLegacyNF.FlowRoutine(1, ingestC.hostname, ingestC.portLegacy, false)
+			log.Infof("listening for legacy netflow on host %s, port = %d", c.hostname, c.portLegacy)
+			err = sLegacyNF.FlowRoutine(1, c.hostname, c.portLegacy, false)
 			log.Fatal(err)
 		}()
 	}
 }
 
-func (ingestC *ingestCollector) processLogLines(out chan<- []config.GenericMap) {
-	var records []config.GenericMap
-	// Maximum batch time for each batch
-	flushRecords := time.NewTicker(ingestC.batchFlushTime)
-	defer flushRecords.Stop()
+func (c *ingestCollector) processLogLines(out chan<- config.GenericMap) {
 	for {
 		select {
-		case <-ingestC.exitChan:
+		case <-c.exitChan:
 			log.Debugf("exiting ingestCollector because of signal")
 			return
-		case record := <-ingestC.in:
-			records = append(records, record)
-			if len(records) >= ingestC.batchMaxLength {
-				log.Debugf("ingestCollector sending %d entries, %d entries waiting", len(records), len(ingestC.in))
-				ingestC.metrics.flowsProcessed.Add(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(ingestC.in) > 0 {
-					for len(records) < ingestC.batchMaxLength && len(ingestC.in) > 0 {
-						record := <-ingestC.in
-						records = append(records, record)
-					}
-				}
-				log.Debugf("ingestCollector sending %d entries, %d entries waiting", len(records), len(ingestC.in))
-				ingestC.metrics.flowsProcessed.Add(float64(len(records)))
-				log.Debugf("ingestCollector records = %v", records)
-				out <- records
-				records = []config.GenericMap{}
-			}
+		case record := <-c.in:
+			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, func() int { return len(in) })
+	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 1a9aaa6e..283b2c19 100644
--- a/pkg/pipeline/ingest/ingest_grpc.go
+++ b/pkg/pipeline/ingest/ingest_grpc.go
@@ -3,9 +3,6 @@ package ingest
 import (
 	"context"
 	"fmt"
-	"strconv"
-	"strings"
-	"time"
 
 	"github.com/netobserv/flowlogs-pipeline/pkg/api"
 	"github.com/netobserv/flowlogs-pipeline/pkg/config"
@@ -14,28 +11,14 @@ import (
 	"github.com/netobserv/flowlogs-pipeline/pkg/pipeline/utils"
 	"github.com/netobserv/netobserv-ebpf-agent/pkg/grpc"
 	"github.com/netobserv/netobserv-ebpf-agent/pkg/pbflow"
-	flow "github.com/netsampler/goflow2/utils"
-	"github.com/prometheus/client_golang/prometheus"
 	log "github.com/sirupsen/logrus"
 	grpc2 "google.golang.org/grpc"
-	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/proto"
 )
 
 const (
 	defaultBufferLen = 100
-	decoderName      = "protobuf"
-	decoderVersion   = "protobuf1.0"
-)
-
-// Prometheus metrics describing the performance of the eBPF ingest
-// This metrics are internal to the goflow2 library, we update them here manually to align with the IPFIX ingester
-var (
-	flowDecoderCount    = flow.DecoderStats
-	processDelaySummary = flow.NetFlowTimeStatsSum.With(
-		prometheus.Labels{"version": decoderVersion, "router": ""})
-	flowTrafficBytesSum = flow.MetricPacketSizeSum
-	flowErrors          = flow.NetFlowErrors
 )
 
 // GRPCProtobuf ingests data from the NetObserv eBPF Agent, using Protocol Buffers over gRPC
@@ -58,10 +41,9 @@ func NewGRPCProtobuf(opMetrics *operational.Metrics, params config.StageParam) (
 		bufLen = defaultBufferLen
 	}
 	flowPackets := make(chan *pbflow.Records, bufLen)
-	metrics := newMetrics(opMetrics, params.Name, func() int { return len(flowPackets) })
-	counter := func(inc int) { metrics.flowsProcessed.Add(float64(inc)) }
+	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, counter))))
+		grpc.WithGRPCServerOptions(grpc2.UnaryInterceptor(instrumentGRPC(metrics))))
 	if err != nil {
 		return nil, err
 	}
@@ -72,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()
@@ -80,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)
+		}
 	}
 }
 
@@ -92,15 +75,15 @@ func (no *GRPCProtobuf) Close() error {
 	return err
 }
 
-func instrumentGRPC(port int, counter func(int)) grpc2.UnaryServerInterceptor {
-	localPort := strconv.Itoa(port)
+func instrumentGRPC(m *metrics) grpc2.UnaryServerInterceptor {
 	return func(
 		ctx context.Context,
 		req interface{},
 		info *grpc2.UnaryServerInfo,
 		handler grpc2.UnaryHandler,
 	) (resp interface{}, err error) {
-		timeReceived := time.Now()
+		timer := m.stageDurationTimer()
+		timeReceived := timer.Start()
 		if info.FullMethod != "/pbflow.Collector/Send" {
 			return handler(ctx, req)
 		}
@@ -109,41 +92,23 @@ func instrumentGRPC(port int, counter func(int)) grpc2.UnaryServerInterceptor {
 		// instrument difference between flow time and ingest time
 		for _, entry := range flowRecords.Entries {
 			delay := timeReceived.Sub(entry.TimeFlowEnd.AsTime()).Seconds()
-			processDelaySummary.Observe(delay)
+			m.latency.Observe(delay)
 		}
 
-		// instruments number of decoded flow messages
-		flowDecoderCount.With(
-			prometheus.Labels{"worker": "", "name": decoderName}).Inc()
-
-		// instruments number of processed individual flows
-		counter(len(flowRecords.Entries))
-
-		// extract sender IP address
-		remoteIP := "unknown"
-		if md, ok := metadata.FromIncomingContext(ctx); ok {
-			if auth := md.Get(":authority"); len(auth) > 0 {
-				if portIdx := strings.IndexByte(auth[0], ':'); portIdx > 0 {
-					remoteIP = auth[0][:portIdx]
-				} else {
-					remoteIP = auth[0]
-				}
-			}
-		}
-		trafficLabels := prometheus.Labels{
-			"type":       "protobuf",
-			"remote_ip":  remoteIP,
-			"local_ip":   "0.0.0.0",
-			"local_port": localPort,
-		}
+		// instrument flows processed counter
+		m.flowsProcessed.Add(float64(len(flowRecords.Entries)))
 
 		// instrument message bytes
-		flowTrafficBytesSum.With(trafficLabels).Observe(float64(proto.Size(flowRecords)))
+		m.batchSizeBytes.Observe(float64(proto.Size(flowRecords)))
 
 		resp, err = handler(ctx, req)
 		if err != nil {
-			flowErrors.With(prometheus.Labels{"router": "", "error": err.Error()}).Inc()
+			m.error(fmt.Sprint(status.Code(err)))
 		}
+
+		// Stage duration
+		timer.ObserveMilliseconds()
+
 		return resp, err
 	}
 }
diff --git a/pkg/pipeline/ingest/ingest_kafka.go b/pkg/pipeline/ingest/ingest_kafka.go
index 8bad31d7..6571f94e 100644
--- a/pkg/pipeline/ingest/ingest_kafka.go
+++ b/pkg/pipeline/ingest/ingest_kafka.go
@@ -26,8 +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"
-
-	"github.com/prometheus/client_golang/prometheus"
 	kafkago "github.com/segmentio/kafka-go"
 	log "github.com/sirupsen/logrus"
 	"golang.org/x/net/context"
@@ -39,14 +37,14 @@ type kafkaReadMessage interface {
 }
 
 type ingestKafka struct {
-	kafkaParams    api.IngestKafka
-	kafkaReader    kafkaReadMessage
-	decoder        decode.Decoder
-	in             chan []byte
-	exitChan       <-chan struct{}
-	batchMaxLength int
-	metrics        *metrics
-	canLogMessages bool
+	kafkaReader      kafkaReadMessage
+	decoder          decode.Decoder
+	in               chan []byte
+	exitChan         <-chan struct{}
+	batchReadTimeout int64
+	batchMaxLength   int
+	metrics          *metrics
+	canLogMessages   bool
 }
 
 const defaultBatchReadTimeout = int64(1000)
@@ -54,111 +52,81 @@ 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
-func (ingestK *ingestKafka) kafkaListener() {
+func (k *ingestKafka) kafkaListener() {
 	log.Debugf("entering kafkaListener")
 
 	go func() {
 		for {
 			// block until a message arrives
 			log.Debugf("before ReadMessage")
-			kafkaMessage, err := ingestK.kafkaReader.ReadMessage(context.Background())
+			kafkaMessage, err := k.kafkaReader.ReadMessage(context.Background())
 			if err != nil {
 				log.Errorln(err)
 				continue
 			}
-			if ingestK.canLogMessages {
+			if k.canLogMessages {
 				log.Debugf("string(kafkaMessage) = %s\n", string(kafkaMessage.Value))
 			}
+			k.metrics.flowsProcessed.Inc()
 			messageLen := len(kafkaMessage.Value)
+			k.metrics.batchSizeBytes.Observe(float64(messageLen) + float64(len(kafkaMessage.Key)))
 			if messageLen > 0 {
-				trafficLabels := prometheus.Labels{
-					"type":       ingestK.kafkaParams.Decoder.Type,
-					"remote_ip":  ingestK.kafkaParams.Brokers[0],
-					"local_ip":   "0.0.0.0",
-					"local_port": "0",
-				}
-				flowTrafficBytesSum.With(trafficLabels).Observe(float64(messageLen))
-				ingestK.in <- kafkaMessage.Value
+				// process message
+				k.in <- kafkaMessage.Value
 			}
 		}
 	}()
-
 }
 
-func processRecordDelay(record config.GenericMap) {
+func (k *ingestKafka) processRecordDelay(record config.GenericMap) {
 	TimeFlowEndInterface, ok := record["TimeFlowEnd"]
 	if !ok {
-		flowErrors.With(prometheus.Labels{"router": "", "error": "No TimeFlowEnd found"}).Inc()
+		k.metrics.error("TimeFlowEnd missing")
 		return
 	}
 	TimeFlowEnd, ok := TimeFlowEndInterface.(float64)
 	if !ok {
-		flowErrors.With(prometheus.Labels{"router": "", "error": "Cannot parse TimeFlowEnd"}).Inc()
+		k.metrics.error("Cannot parse TimeFlowEnd")
 		return
 	}
 	delay := time.Since(time.Unix(int64(TimeFlowEnd), 0)).Seconds()
-	processDelaySummary.Observe(delay)
+	k.metrics.latency.Observe(delay)
 }
 
-func (ingestK *ingestKafka) processBatch(out chan<- []config.GenericMap, records [][]byte) {
-	log.Debugf("ingestKafka sending %d records, %d entries waiting", len(records), len(ingestK.in))
-
+func (k *ingestKafka) processRecord(record []byte, out chan<- config.GenericMap) {
 	// Decode batch
-	decoded := ingestK.decoder.Decode(records)
-
-	// Update metrics
-	flowDecoderCount.With(
-		prometheus.Labels{"worker": "", "name": ingestK.kafkaParams.Decoder.Type}).Inc()
-	ingestK.metrics.flowsProcessed.Add(float64(len(records)))
-
-	for _, record := range decoded {
-		processRecordDelay(record)
+	decoded, err := k.decoder.Decode(record)
+	if err != nil {
+		log.WithError(err).Warnf("ignoring flow")
+		return
 	}
+	k.processRecordDelay(decoded)
 
 	// Send batch
 	out <- decoded
 }
 
 // read items from ingestKafka input channel, pool them, and send down the pipeline
-func (ingestK *ingestKafka) processLogLines(out chan<- []config.GenericMap) {
-	var records [][]byte
-	duration := time.Duration(ingestK.kafkaParams.BatchReadTimeout) * time.Millisecond
-	flushRecords := time.NewTicker(duration)
+func (k *ingestKafka) processLogLines(out chan<- config.GenericMap) {
 	for {
 		select {
-		case <-ingestK.exitChan:
+		case <-k.exitChan:
 			log.Debugf("exiting ingestKafka because of signal")
 			return
-		case record := <-ingestK.in:
-			records = append(records, record)
-			if len(records) >= ingestK.batchMaxLength {
-				ingestK.processBatch(out, records)
-				records = [][]byte{}
-			}
-		case <-flushRecords.C: // Maximum batch time for each batch
-			// Process batch of records (if not empty)
-			if len(records) > 0 {
-				if len(ingestK.in) > 0 {
-					for len(records) < ingestK.batchMaxLength && len(ingestK.in) > 0 {
-						record := <-ingestK.in
-						records = append(records, record)
-					}
-				}
-				ingestK.processBatch(out, records)
-				records = [][]byte{}
-			}
+		case record := <-k.in:
+			k.processRecord(record, out)
 		}
 	}
 }
@@ -199,10 +167,11 @@ func NewIngestKafka(opMetrics *operational.Metrics, params config.StageParam) (I
 		}
 	}
 
-	if jsonIngestKafka.BatchReadTimeout == 0 {
-		jsonIngestKafka.BatchReadTimeout = defaultBatchReadTimeout
+	batchReadTimeout := defaultBatchReadTimeout
+	if jsonIngestKafka.BatchReadTimeout != 0 {
+		batchReadTimeout = jsonIngestKafka.BatchReadTimeout
 	}
-	log.Infof("batchReadTimeout = %d", jsonIngestKafka.BatchReadTimeout)
+	log.Infof("batchReadTimeout = %d", batchReadTimeout)
 
 	commitInterval := int64(defaultKafkaCommitInterval)
 	if jsonIngestKafka.CommitInterval != 0 {
@@ -262,16 +231,16 @@ func NewIngestKafka(opMetrics *operational.Metrics, params config.StageParam) (I
 	}
 
 	in := make(chan []byte, 2*bml)
-	metrics := newMetrics(opMetrics, params.Name, func() int { return len(in) })
+	metrics := newMetrics(opMetrics, params.Name, params.Ingest.Type, func() int { return len(in) })
 
 	return &ingestKafka{
-		kafkaParams:    jsonIngestKafka,
-		kafkaReader:    kafkaReader,
-		decoder:        decoder,
-		exitChan:       utils.ExitChannel(),
-		in:             in,
-		batchMaxLength: bml,
-		metrics:        metrics,
-		canLogMessages: jsonIngestKafka.Decoder.Type == api.DecoderName("JSON"),
+		kafkaReader:      kafkaReader,
+		decoder:          decoder,
+		exitChan:         utils.ExitChannel(),
+		in:               in,
+		batchMaxLength:   bml,
+		batchReadTimeout: batchReadTimeout,
+		metrics:          metrics,
+		canLogMessages:   jsonIngestKafka.Decoder.Type == api.DecoderName("JSON"),
 	}, nil
 }
diff --git a/pkg/pipeline/ingest/metrics.go b/pkg/pipeline/ingest/metrics.go
index bc322bf3..d59f761f 100644
--- a/pkg/pipeline/ingest/metrics.go
+++ b/pkg/pipeline/ingest/metrics.go
@@ -6,21 +6,69 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 )
 
+var (
+	latencyHistogram = operational.DefineMetric(
+		"ingest_latency_ms",
+		"Latency between flow end time and ingest time, in milliseconds",
+		operational.TypeHistogram,
+		"stage",
+	)
+	flowsProcessedCounter = operational.DefineMetric(
+		"ingest_flows_processed",
+		"Number of flows received by the ingester",
+		operational.TypeCounter,
+		"stage",
+	)
+	batchSizeBytesSummary = operational.DefineMetric(
+		"ingest_batch_size_bytes",
+		"Ingested batch size distribution, in bytes",
+		operational.TypeSummary,
+		"stage",
+	)
+	errorsCounter = operational.DefineMetric(
+		"ingest_errors",
+		"Counter of errors during ingestion",
+		operational.TypeCounter,
+		"stage", "type", "code",
+	)
+)
+
 type metrics struct {
 	*operational.Metrics
 	stage          string
+	stageType      string
+	stageDuration  prometheus.Observer
+	latency        prometheus.Histogram
 	flowsProcessed prometheus.Counter
+	batchSizeBytes prometheus.Summary
+	errors         *prometheus.CounterVec
 }
 
-func newMetrics(opMetrics *operational.Metrics, stage string, inGaugeFunc func() int) *metrics {
+func newMetrics(opMetrics *operational.Metrics, stage, stageType string, inGaugeFunc func() int) *metrics {
 	opMetrics.CreateInQueueSizeGauge(stage, inGaugeFunc)
 	return &metrics{
 		Metrics:        opMetrics,
 		stage:          stage,
-		flowsProcessed: opMetrics.CreateFlowsProcessedCounter(stage),
+		stageType:      stageType,
+		latency:        opMetrics.NewHistogram(&latencyHistogram, []float64{.001, .01, .1, 1, 10, 100, 1000, 10000}, stage),
+		stageDuration:  opMetrics.GetOrCreateStageDurationHisto().WithLabelValues(stage),
+		flowsProcessed: opMetrics.NewCounter(&flowsProcessedCounter, stage),
+		batchSizeBytes: opMetrics.NewSummary(&batchSizeBytesSummary, stage),
+		errors:         opMetrics.NewCounterVec(&errorsCounter),
 	}
 }
 
-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) })
 }
+
+// Increment error counter
+// `code` should reflect any error code relative to this type. It can be a short string message,
+// but make sure to not include any dynamic value with high cardinality
+func (m *metrics) error(code string) {
+	m.errors.WithLabelValues(m.stage, m.stageType, code).Inc()
+}
+
+func (m *metrics) stageDurationTimer() *operational.Timer {
+	return operational.NewTimer(m.stageDuration)
+}
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 33e2ac9c..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,25 +65,29 @@ type pipelineEntry struct {
 	Writer      write.Writer
 }
 
-var (
-	stageDurationDef = operational.DefineMetric(
-		"stage_duration_ms",
-		"Pipeline stage duration in milliseconds",
-		operational.TypeHistogram,
-		"stage",
-	)
-)
+func newBuilder(cfg *config.ConfigFileStruct) *builder {
+	// Get global metrics settings
+	opMetrics := operational.NewMetrics(&cfg.MetricsSettings)
+	stageDuration := opMetrics.GetOrCreateStageDurationHisto()
 
-func newBuilder(params []config.StageParam, stages []config.Stage, opMetrics *operational.Metrics) *builder {
-	stageDuration := opMetrics.NewHistogramVec(&stageDurationDef, []float64{.001, .01, .1, 1, 10, 100, 1000, 10000})
+	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,
 	}
 }
 
@@ -247,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() {
@@ -258,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() {
@@ -269,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/kubernetes/kubernetes.go b/pkg/pipeline/transform/kubernetes/kubernetes.go
index 46c28c17..aed5f324 100644
--- a/pkg/pipeline/transform/kubernetes/kubernetes.go
+++ b/pkg/pipeline/transform/kubernetes/kubernetes.go
@@ -194,7 +194,7 @@ func (k *KubeData) initNodeInformer(informerFactory informers.SharedInformerFact
 		return &Info{
 			ObjectMeta: metav1.ObjectMeta{
 				Name:      node.Name,
-				Namespace: node.Namespace,
+				Namespace: "",
 				Labels:    node.Labels,
 			},
 			ips:  ips,
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 c66c2c11..7adcddf7 100644
--- a/pkg/pipeline/ingest/ingest_kafka_test.go
+++ b/pkg/pipeline/ingest/ingest_kafka_test.go
@@ -83,13 +83,13 @@ func initNewIngestKafka(t *testing.T, configTemplate string) Ingester {
 func Test_NewIngestKafka1(t *testing.T) {
 	newIngest := initNewIngestKafka(t, testConfig1)
 	ingestKafka := newIngest.(*ingestKafka)
-	require.Equal(t, "topic1", ingestKafka.kafkaParams.Topic)
-	require.Equal(t, "group1", ingestKafka.kafkaParams.GroupId)
+	require.Equal(t, "topic1", ingestKafka.kafkaReader.Config().Topic)
+	require.Equal(t, "group1", ingestKafka.kafkaReader.Config().GroupID)
 	expectedBrokers := []string{"1.1.1.1:9092"}
-	require.Equal(t, expectedBrokers, ingestKafka.kafkaParams.Brokers)
-	require.Equal(t, "FirstOffset", ingestKafka.kafkaParams.StartOffset)
+	require.Equal(t, expectedBrokers, ingestKafka.kafkaReader.Config().Brokers)
+	require.Equal(t, int64(-2), ingestKafka.kafkaReader.Config().StartOffset)
 	require.Equal(t, 2, len(ingestKafka.kafkaReader.Config().GroupBalancers))
-	require.Equal(t, int64(300), ingestKafka.kafkaParams.BatchReadTimeout)
+	require.Equal(t, int64(300), ingestKafka.batchReadTimeout)
 	require.Equal(t, int(500), ingestKafka.batchMaxLength)
 	require.Equal(t, time.Duration(500)*time.Millisecond, ingestKafka.kafkaReader.Config().CommitInterval)
 }
@@ -97,27 +97,25 @@ func Test_NewIngestKafka1(t *testing.T) {
 func Test_NewIngestKafka2(t *testing.T) {
 	newIngest := initNewIngestKafka(t, testConfig2)
 	ingestKafka := newIngest.(*ingestKafka)
-	require.Equal(t, "topic2", ingestKafka.kafkaParams.Topic)
-	require.Equal(t, "group2", ingestKafka.kafkaParams.GroupId)
+	require.Equal(t, "topic2", ingestKafka.kafkaReader.Config().Topic)
+	require.Equal(t, "group2", ingestKafka.kafkaReader.Config().GroupID)
 	expectedBrokers := []string{"1.1.1.2:9092"}
-	require.Equal(t, expectedBrokers, ingestKafka.kafkaParams.Brokers)
-	require.Equal(t, "LastOffset", ingestKafka.kafkaParams.StartOffset)
+	require.Equal(t, expectedBrokers, ingestKafka.kafkaReader.Config().Brokers)
+	require.Equal(t, int64(-1), ingestKafka.kafkaReader.Config().StartOffset)
 	require.Equal(t, 1, len(ingestKafka.kafkaReader.Config().GroupBalancers))
-	require.Equal(t, defaultBatchReadTimeout, ingestKafka.kafkaParams.BatchReadTimeout)
+	require.Equal(t, defaultBatchReadTimeout, ingestKafka.batchReadTimeout)
 	require.Equal(t, int(1000), ingestKafka.batchMaxLength)
 	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.kafkaParams.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.kafkaParams.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: