Repository: netobserv/flowlogs-pipeline
Pull request #992 (2025-06-11 13:41:36)

NETOBSERV-2285: opt: use uint64 hash instead of string for cache key
In timed_cache, use uint64 hash (fnv 64a) instead of string builder
Uses the same implementation as in prom go client

Also:
- [Bump agent, with improved pb decode](https://github.com/netobserv/flowlogs-pipeline/pull/992/commits/a8e13410001541b7ab48dcab3dac4a0e8e313b20)

![image](https://github.com/user-attachments/assets/6b84ca39-e267-411a-a836-726f637061c4)


- [reduce allocations in metrics encode](https://github.com/netobserv/flowlogs-pipeline/pull/992/commits/da2f2679d60fab7c2cca388004928305e4a3f18f)

![image](https://github.com/user-attachments/assets/05dcaa1e-d051-4185-943b-8e3b01340ed0)


openshift-ci[bot]: Skipping CI for Draft Pull Request.
If you want CI signal for your change, please convert it to an actual PR.
You can still manually trigger a test run with `/test all`
jotak: /test flp-node-density-heavy-25nodes
openshift-merge-robot: PR needs rebase.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository.
</details>
jotak: /test flp-node-density-heavy-25nodes
github-actions[bot]: New image:
quay.io/netobserv/flowlogs-pipeline:ae8e882

It will expire after two weeks.

To deploy this build, run from the operator repo, assuming the operator is running:
```bash
USER=netobserv VERSION=ae8e882 make set-flp-image
```

openshift-ci[bot]: @jotak: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
ci/prow/qe-e2e-tests | da2f2679d60fab7c2cca388004928305e4a3f18f | [link](https://prow.ci.openshift.org/view/gs/test-platform-results/pr-logs/pull/netobserv_flowlogs-pipeline/992/pull-ci-netobserv-flowlogs-pipeline-main-qe-e2e-tests/1932863634925424640) | false | `/test qe-e2e-tests`

[Full PR test history](https://prow.ci.openshift.org/pr-history?org=netobserv&repo=flowlogs-pipeline&pr=992). [Your PR dashboard](https://prow.ci.openshift.org/pr?query=is:pr+state:open+author:jotak).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->

Patch:
diff --git a/README.md b/README.md
index d95b2c96..c347d4c4 100644
--- a/README.md
+++ b/README.md
@@ -52,7 +52,7 @@ Flags:
       --config string              config file (default is $HOME/.flowlogs-pipeline)  
       --dynamicParameters string   json of configmap location for dynamic parameters  
       --health.address string      Health server address (default "0.0.0.0")  
-      --health.port string         Health server port (default "8080")  
+      --health.port int            Health server port (default: disable health server)   
   -h, --help                       help for flowlogs-pipeline  
       --log-level string           Log level: debug, info, warning, error (default "error")  
       --metricsSettings string     json for global metrics settings  
@@ -876,7 +876,7 @@ A sample is the following:
 
 ```
 metricsSettings:
-  suppressGoMetrics: true
+  suppressDefaultMetrics: true
   prefix: flp_operational_
   port: 9102
 
@@ -887,7 +887,7 @@ In addition, there are default metrics reported by `Go`, which are also directed
 The port to which these metrics are made available is specified in the `port` configuration parameter.
 If a `prefix` is specified, then the specified prefix is prepended to each of the operational metrics generated by FLP.
 A different `prefix` may be specified on an `encode prom` stage to be prepended to the prometheus metrics defined in that stage.
-The `suppressGoMetrics` parameter may be set to `true` in order to suppress the reporting of the `Go` and process metrics in the prometheus client interface.
+The `suppressDefaultMetrics` parameter may be set to `true` in order to suppress the reporting of the default Prometheus metrics, such as `Go` and process metrics.
 
 # Development
 
diff --git a/cmd/flowlogs-pipeline/main.go b/cmd/flowlogs-pipeline/main.go
index 4a70552f..a6c571e7 100644
--- a/cmd/flowlogs-pipeline/main.go
+++ b/cmd/flowlogs-pipeline/main.go
@@ -143,7 +143,7 @@ func initFlags() {
 	rootCmd.PersistentFlags().StringVar(&cfgFile, "config", "", fmt.Sprintf("config file (default is $HOME/%s)", defaultLogFileName))
 	rootCmd.PersistentFlags().StringVar(&logLevel, "log-level", "error", "Log level: debug, info, warning, error")
 	rootCmd.PersistentFlags().StringVar(&opts.Health.Address, "health.address", "0.0.0.0", "Health server address")
-	rootCmd.PersistentFlags().StringVar(&opts.Health.Port, "health.port", "8080", "Health server port")
+	rootCmd.PersistentFlags().IntVar(&opts.Health.Port, "health.port", 0, "Health server port (default: disable health server) ")
 	rootCmd.PersistentFlags().IntVar(&opts.Profile.Port, "profile.port", 0, "Go pprof tool port (default: disabled)")
 	rootCmd.PersistentFlags().StringVar(&opts.PipeLine, "pipeline", "", "json of config file pipeline field")
 	rootCmd.PersistentFlags().StringVar(&opts.Parameters, "parameters", "", "json of config file parameters field")
@@ -199,7 +199,10 @@ func run() {
 	}
 
 	// Start health report server
-	healthServer := operational.NewHealthServer(&opts, mainPipeline.IsAlive, mainPipeline.IsReady)
+	var healthServer *http.Server
+	if opts.Health.Port > 0 {
+		healthServer = operational.NewHealthServer(&opts, mainPipeline.IsAlive, mainPipeline.IsReady)
+	}
 
 	// Starts the flows pipeline
 	mainPipeline.Run()
@@ -207,7 +210,9 @@ func run() {
 	if promServer != nil {
 		_ = promServer.Shutdown(context.Background())
 	}
-	_ = healthServer.Shutdown(context.Background())
+	if healthServer != nil {
+		_ = healthServer.Shutdown(context.Background())
+	}
 
 	// Give all threads a chance to exit and then exit the process
 	time.Sleep(time.Second)
diff --git a/docs/api.md b/docs/api.md
index e6e19c71..4b3e3f85 100644
--- a/docs/api.md
+++ b/docs/api.md
@@ -34,7 +34,7 @@ Following is the supported API format for prometheus encode:
                  buckets: histogram buckets
                  valueScale: scale factor of the value (MetricVal := FlowVal / Scale)
          prefix: prefix added to each metric name
-         expiryTime: time duration of no-flow to wait before deleting prometheus data item
+         expiryTime: time duration of no-flow to wait before deleting prometheus data item (default: 2m)
          maxMetrics: maximum number of metrics to report (default: unlimited)
 </pre>
 ## Kafka encode API
@@ -90,6 +90,8 @@ Following is the supported API format for the NetFlow / IPFIX collector:
          port: the port number to listen on, for IPFIX/NetFlow v9. Omit or set to 0 to disable IPFIX/NetFlow v9 ingestion
          portLegacy: the port number to listen on, for legacy NetFlow v5. Omit or set to 0 to disable NetFlow v5 ingestion
          batchMaxLen: the number of accumulated flows before being forwarded for processing
+         workers: the number of netflow/ipfix decoding workers
+         sockets: the number of listening sockets
 </pre>
 ## Ingest Kafka API
 Following is the supported API format for the kafka ingest:
@@ -319,6 +321,8 @@ Following is the supported API format for writing to loki:
          clientConfig: clientConfig
          timestampLabel: label to use for time indexing
          timestampScale: timestamp units scale (e.g. for UNIX = 1s)
+         format: the format of each line: printf (writes using golang's default map printing), fields (writes one key and value field per line) or json (default)
+         reorder: reorder json map keys
 </pre>
 ## Write Standard Output
 Following is the supported API format for writing to standard output:
@@ -458,8 +462,8 @@ Following is the supported API format for writing metrics to an OpenTelemetry co
                  flatten: list fields to be flattened
                  buckets: histogram buckets
                  valueScale: scale factor of the value (MetricVal := FlowVal / Scale)
-         pushTimeInterval: how often should metrics be sent to collector:
-         expiryTime: time duration of no-flow to wait before deleting data item
+         pushTimeInterval: how often should metrics be sent to collector (default: 20s)
+         expiryTime: time duration of no-flow to wait before deleting data item (default: 2m)
 </pre>
 ## OpenTelemetry Traces API
 Following is the supported API format for writing traces to an OpenTelemetry collector:
diff --git a/go.mod b/go.mod
index 7501d50c..77489f76 100644
--- a/go.mod
+++ b/go.mod
@@ -18,7 +18,7 @@ require (
 	github.com/mitchellh/mapstructure v1.5.0
 	github.com/netobserv/gopipes v0.3.0
 	github.com/netobserv/loki-client-go v0.0.0-20250425113517-526b43e51847
-	github.com/netobserv/netobserv-ebpf-agent v1.9.0-crc0.0.20250528064221-6f34e5b85d2c
+	github.com/netobserv/netobserv-ebpf-agent v1.9.0-crc0.0.20250610144135-d64c5d99f2da
 	github.com/netsampler/goflow2 v1.3.7
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.22.0
diff --git a/go.sum b/go.sum
index c44afd51..d326ca1f 100644
--- a/go.sum
+++ b/go.sum
@@ -258,8 +258,8 @@ github.com/netobserv/gopipes v0.3.0 h1:IYmPnnAVCdSK7VmHmpFhrVBOEm45qpgbZmJz1sSW+
 github.com/netobserv/gopipes v0.3.0/go.mod h1:N7/Gz05EOF0CQQSKWsv3eof22Cj2PB08Pbttw98YFYU=
 github.com/netobserv/loki-client-go v0.0.0-20250425113517-526b43e51847 h1:hjzhVZSSKIOmAzHbGUV4JhVIPkgKs/UtrWDx6JSVKMw=
 github.com/netobserv/loki-client-go v0.0.0-20250425113517-526b43e51847/go.mod h1:Zb/jtD3Lnu88Poo+jnhTASzxYnvncmHOoZaT93xQjJ8=
-github.com/netobserv/netobserv-ebpf-agent v1.9.0-crc0.0.20250528064221-6f34e5b85d2c h1:HhObTHfakq06NAPkljppnT1UT9MV2o5obH2dPD0jxmk=
-github.com/netobserv/netobserv-ebpf-agent v1.9.0-crc0.0.20250528064221-6f34e5b85d2c/go.mod h1:sqvri+RCx8QefV2g+pbJktbNuO3ebB1TE1Zl7M2qqDo=
+github.com/netobserv/netobserv-ebpf-agent v1.9.0-crc0.0.20250610144135-d64c5d99f2da h1:fahbLlD/5BidF+7rcvydhro3Tu0pY81OUGOkxmaY07A=
+github.com/netobserv/netobserv-ebpf-agent v1.9.0-crc0.0.20250610144135-d64c5d99f2da/go.mod h1:IfxvtBeSfhJaCO/7ie3R7mna+j8sAer2vqtbwaBTlzA=
 github.com/netsampler/goflow2 v1.3.7 h1:XZaTy8kkMnGXpJ9hS3KbO1McyrFTpVNhVFEx9rNhMmc=
 github.com/netsampler/goflow2 v1.3.7/go.mod h1:4UZsVGVAs//iMCptUHn3WNScztJeUhZH7kDW2+/vDdQ=
 github.com/nxadm/tail v1.4.4/go.mod h1:kenIhsEOeOJmVchQTgglprH7qJGnHDVpk1VPCcaMI8A=
diff --git a/hack/examples/docker-ipfix-config.yaml b/hack/examples/docker-ipfix-config.yaml
index a7963bfb..acfe2da7 100644
--- a/hack/examples/docker-ipfix-config.yaml
+++ b/hack/examples/docker-ipfix-config.yaml
@@ -1,4 +1,9 @@
 log-level: info
+health:
+  port: 8080
+metricsSettings:
+  suppressDefaultMetrics: true
+  prefix: flp_
 pipeline:
   - name: ingest
   - name: conntrack
diff --git a/pkg/api/encode_otlp.go b/pkg/api/encode_otlp.go
index 86e3c32c..0338ad85 100644
--- a/pkg/api/encode_otlp.go
+++ b/pkg/api/encode_otlp.go
@@ -30,8 +30,8 @@ type EncodeOtlpMetrics struct {
 	*OtlpConnectionInfo `json:",inline" doc:"OpenTelemetry connection info; includes:"`
 	Prefix              string       `yaml:"prefix,omitempty" json:"prefix,omitempty" doc:"prefix added to each metric name"`
 	Metrics             MetricsItems `yaml:"metrics,omitempty" json:"metrics,omitempty" doc:"list of metric definitions, each includes:"`
-	PushTimeInterval    Duration     `yaml:"pushTimeInterval,omitempty" json:"pushTimeInterval,omitempty" doc:"how often should metrics be sent to collector:"`
-	ExpiryTime          Duration     `yaml:"expiryTime,omitempty" json:"expiryTime,omitempty" doc:"time duration of no-flow to wait before deleting data item"`
+	PushTimeInterval    Duration     `yaml:"pushTimeInterval,omitempty" json:"pushTimeInterval,omitempty" doc:"how often should metrics be sent to collector (default: 20s)"`
+	ExpiryTime          Duration     `yaml:"expiryTime,omitempty" json:"expiryTime,omitempty" doc:"time duration of no-flow to wait before deleting data item (default: 2m)"`
 }
 
 type OtlpConnectionInfo struct {
diff --git a/pkg/api/encode_prom.go b/pkg/api/encode_prom.go
index 6780a43a..4c04b636 100644
--- a/pkg/api/encode_prom.go
+++ b/pkg/api/encode_prom.go
@@ -26,7 +26,7 @@ type PromEncode struct {
 	*PromConnectionInfo `json:",inline,omitempty" doc:"Prometheus connection info (optional); includes:"`
 	Metrics             MetricsItems `yaml:"metrics,omitempty" json:"metrics,omitempty" doc:"list of prometheus metric definitions, each includes:"`
 	Prefix              string       `yaml:"prefix,omitempty" json:"prefix,omitempty" doc:"prefix added to each metric name"`
-	ExpiryTime          Duration     `yaml:"expiryTime,omitempty" json:"expiryTime,omitempty" doc:"time duration of no-flow to wait before deleting prometheus data item"`
+	ExpiryTime          Duration     `yaml:"expiryTime,omitempty" json:"expiryTime,omitempty" doc:"time duration of no-flow to wait before deleting prometheus data item (default: 2m)"`
 	MaxMetrics          int          `yaml:"maxMetrics,omitempty" json:"maxMetrics,omitempty" doc:"maximum number of metrics to report (default: unlimited)"`
 }
 
diff --git a/pkg/api/ingest_collector.go b/pkg/api/ingest_collector.go
index 72fa97a1..090b3fdb 100644
--- a/pkg/api/ingest_collector.go
+++ b/pkg/api/ingest_collector.go
@@ -22,4 +22,6 @@ type IngestCollector struct {
 	Port        int    `yaml:"port,omitempty" json:"port,omitempty" doc:"the port number to listen on, for IPFIX/NetFlow v9. Omit or set to 0 to disable IPFIX/NetFlow v9 ingestion"`
 	PortLegacy  int    `yaml:"portLegacy,omitempty" json:"portLegacy,omitempty" doc:"the port number to listen on, for legacy NetFlow v5. Omit or set to 0 to disable NetFlow v5 ingestion"`
 	BatchMaxLen int    `yaml:"batchMaxLen,omitempty" json:"batchMaxLen,omitempty" doc:"the number of accumulated flows before being forwarded for processing"`
+	Workers     int    `yaml:"workers,omitempty" json:"workers,omitempty" doc:"the number of netflow/ipfix decoding workers"`
+	Sockets     int    `yaml:"sockets,omitempty" json:"sockets,omitempty" doc:"the number of listening sockets"`
 }
diff --git a/pkg/api/write_loki.go b/pkg/api/write_loki.go
index 61885751..344c8edf 100644
--- a/pkg/api/write_loki.go
+++ b/pkg/api/write_loki.go
@@ -44,6 +44,8 @@ type WriteLoki struct {
 	// scales of '1ms' (one millisecond) or just '1' (one nanosecond)
 	// Default value is '1s'
 	TimestampScale string `yaml:"timestampScale,omitempty" json:"timestampScale,omitempty" doc:"timestamp units scale (e.g. for UNIX = 1s)"`
+	Format         string `yaml:"format,omitempty" json:"format,omitempty" doc:"the format of each line: printf (writes using golang's default map printing), fields (writes one key and value field per line) or json (default)"`
+	Reorder        bool   `yaml:"reorder,omitempty" json:"reorder,omitempty" doc:"reorder json map keys"`
 }
 
 func (w *WriteLoki) SetDefaults() {
@@ -71,6 +73,9 @@ func (w *WriteLoki) SetDefaults() {
 	if w.TimestampScale == "" {
 		w.TimestampScale = "1s"
 	}
+	if w.Format == "" {
+		w.Format = "json"
+	}
 }
 
 func (w *WriteLoki) Validate() error {
diff --git a/pkg/config/config.go b/pkg/config/config.go
index e7311f70..d08b1471 100644
--- a/pkg/config/config.go
+++ b/pkg/config/config.go
@@ -60,7 +60,7 @@ type HotReloadStruct struct {
 
 type Health struct {
 	Address string
-	Port    string
+	Port    int
 }
 
 type Profile struct {
@@ -75,7 +75,7 @@ type MetricsSettings struct {
 	DisableGlobalServer    bool   `yaml:"disableGlobalServer,omitempty" json:"disableGlobalServer,omitempty" doc:"disabling the global metrics server makes operational metrics unavailable. If prometheus-encoding stages are defined, they need to contain their own metrics server parameters."`
 	Prefix                 string `yaml:"prefix,omitempty" json:"prefix,omitempty" doc:"prefix for names of the operational metrics"`
 	NoPanic                bool   `yaml:"noPanic,omitempty" json:"noPanic,omitempty"`
-	SuppressGoMetrics      bool   `yaml:"suppressGoMetrics,omitempty" json:"suppressGoMetrics,omitempty" doc:"filter out Go and process metrics"`
+	SuppressDefaultMetrics bool   `yaml:"suppressDefaultMetrics,omitempty" json:"suppressDefaultMetrics,omitempty" doc:"filter out default Prometheus metrics such as Go and process metrics"`
 }
 
 // PerfSettings allows setting some internal configuration parameters
diff --git a/pkg/operational/health.go b/pkg/operational/health.go
index c4510a69..b60a5917 100644
--- a/pkg/operational/health.go
+++ b/pkg/operational/health.go
@@ -18,7 +18,7 @@
 package operational
 
 import (
-	"net"
+	"fmt"
 	"net/http"
 	"time"
 
@@ -30,7 +30,7 @@ import (
 
 func NewHealthServer(opts *config.Options, isAlive healthcheck.Check, isReady healthcheck.Check) *http.Server {
 	handler := healthcheck.NewHandler()
-	address := net.JoinHostPort(opts.Health.Address, opts.Health.Port)
+	address := fmt.Sprintf("%s:%d", opts.Health.Address, opts.Health.Port)
 	handler.AddLivenessCheck("PipelineCheck", isAlive)
 	handler.AddReadinessCheck("PipelineCheck", isReady)
 
diff --git a/pkg/pipeline/encode/encode_prom.go b/pkg/pipeline/encode/encode_prom.go
index 052839aa..99827de3 100644
--- a/pkg/pipeline/encode/encode_prom.go
+++ b/pkg/pipeline/encode/encode_prom.go
@@ -66,7 +66,7 @@ func (e *EncodeProm) ProcessCounter(m interface{}, labels map[string]string, val
 	return nil
 }
 
-func (e *EncodeProm) ProcessGauge(m interface{}, labels map[string]string, value float64, _ string) error {
+func (e *EncodeProm) ProcessGauge(m interface{}, _ string, labels map[string]string, value float64, _ []string) error {
 	gauge := m.(*prometheus.GaugeVec)
 	mm, err := gauge.GetMetricWith(labels)
 	if err != nil {
diff --git a/pkg/pipeline/encode/metrics_common.go b/pkg/pipeline/encode/metrics_common.go
index be9252f8..425fd6aa 100644
--- a/pkg/pipeline/encode/metrics_common.go
+++ b/pkg/pipeline/encode/metrics_common.go
@@ -18,7 +18,6 @@
 package encode
 
 import (
-	"strings"
 	"time"
 
 	"github.com/netobserv/flowlogs-pipeline/pkg/api"
@@ -53,7 +52,7 @@ type MetricsCommonStruct struct {
 type MetricsCommonInterface interface {
 	GetCacheEntry(entryLabels map[string]string, m interface{}) interface{}
 	ProcessCounter(m interface{}, labels map[string]string, value float64) error
-	ProcessGauge(m interface{}, labels map[string]string, value float64, key string) error
+	ProcessGauge(m interface{}, name string, labels map[string]string, value float64, lvs []string) error
 	ProcessHist(m interface{}, labels map[string]string, value float64) error
 	ProcessAggHist(m interface{}, labels map[string]string, value []float64) error
 }
@@ -132,7 +131,7 @@ func (m *MetricsCommonStruct) MetricCommonEncode(mci MetricsCommonInterface, met
 			continue
 		}
 		for _, labels := range labelSets {
-			err := mci.ProcessGauge(mInfo.genericMetric, labels.lMap, value, labels.key)
+			err := mci.ProcessGauge(mInfo.genericMetric, mInfo.info.Name, labels.lMap, value, labels.values)
 			if err != nil {
 				log.Errorf("labels registering error on %s: %v", mInfo.info.Name, err)
 				m.errorsCounter.WithLabelValues("LabelsRegisteringError", mInfo.info.Name, "").Inc()
@@ -198,20 +197,17 @@ func (m *MetricsCommonStruct) prepareMetric(mci MetricsCommonInterface, flow con
 	}
 
 	labelSets := extractLabels(flow, flatParts, info)
-	lkms := make([]labelsKeyAndMap, 0, len(labelSets))
 	for _, ls := range labelSets {
 		// Update entry for expiry mechanism (the entry itself is its own cleanup function)
-		lkm := ls.toKeyAndMap(info)
-		lkms = append(lkms, lkm)
-		ok := m.mCache.UpdateCacheEntry(lkm.key, func() interface{} {
-			return mci.GetCacheEntry(lkm.lMap, mv)
+		ok := m.mCache.UpdateCacheEntry(ls.values, func() interface{} {
+			return mci.GetCacheEntry(ls.lMap, mv)
 		})
 		if !ok {
 			m.metricsDropped.Inc()
 			return nil, 0
 		}
 	}
-	return lkms, floatVal
+	return labelSets, floatVal
 }
 
 func (m *MetricsCommonStruct) prepareAggHisto(mci MetricsCommonInterface, flow config.GenericMap, info *metrics.Preprocessed, mc interface{}) ([]labelsKeyAndMap, []float64) {
@@ -232,20 +228,17 @@ func (m *MetricsCommonStruct) prepareAggHisto(mci MetricsCommonInterface, flow c
 	}
 
 	labelSets := extractLabels(flow, flatParts, info)
-	lkms := make([]labelsKeyAndMap, 0, len(labelSets))
 	for _, ls := range labelSets {
 		// Update entry for expiry mechanism (the entry itself is its own cleanup function)
-		lkm := ls.toKeyAndMap(info)
-		lkms = append(lkms, lkm)
-		ok := m.mCache.UpdateCacheEntry(lkm.key, func() interface{} {
-			return mci.GetCacheEntry(lkm.lMap, mc)
+		ok := m.mCache.UpdateCacheEntry(ls.values, func() interface{} {
+			return mci.GetCacheEntry(ls.lMap, mc)
 		})
 		if !ok {
 			m.metricsDropped.Inc()
 			return nil, nil
 		}
 	}
-	return lkms, values
+	return labelSets, values
 }
 
 func (m *MetricsCommonStruct) extractGenericValue(flow config.GenericMap, info *metrics.Preprocessed) interface{} {
@@ -261,59 +254,44 @@ func (m *MetricsCommonStruct) extractGenericValue(flow config.GenericMap, info *
 	return val
 }
 
-type label struct {
-	key   string
-	value string
-}
-
-type labelSet []label
-
 type labelsKeyAndMap struct {
-	key  string
-	lMap map[string]string
-}
-
-func (l labelSet) toKeyAndMap(info *metrics.Preprocessed) labelsKeyAndMap {
-	key := strings.Builder{}
-	key.Grow(256) // pre-allocate a decent buffer
-	key.WriteString(info.Name)
-	key.WriteRune('|')
-	m := make(map[string]string, len(l))
-	for _, kv := range l {
-		key.WriteString(kv.value)
-		key.WriteRune('|')
-		m[kv.key] = kv.value
-	}
-	return labelsKeyAndMap{key: key.String(), lMap: m}
+	values []string
+	lMap   map[string]string
 }
 
 // extractLabels takes the flow and a single metric definition as input.
 // It returns the flat labels maps (label names and values).
 // Most of the time it will return a single map; it may return several of them when the parsed flow fields are lists (e.g. "interfaces").
-func extractLabels(flow config.GenericMap, flatParts []config.GenericMap, info *metrics.Preprocessed) []labelSet {
-	common := newLabelSet(flow, info.MappedLabels)
+func extractLabels(flow config.GenericMap, flatParts []config.GenericMap, info *metrics.Preprocessed) []labelsKeyAndMap {
+	common := newLabelKeyAndMap(info.Name, flow, info.MappedLabels)
 	if len(flatParts) == 0 {
-		return []labelSet{common}
+		return []labelsKeyAndMap{common}
 	}
-	var all []labelSet
+	all := make([]labelsKeyAndMap, 0, len(flatParts))
 	for _, fp := range flatParts {
-		ls := newLabelSet(fp, info.FlattenedLabels)
-		ls = append(ls, common...)
+		ls := newLabelKeyAndMap(info.Name, fp, info.FlattenedLabels)
+		ls.values = append(ls.values, common.values...)
+		for k, v := range common.lMap {
+			ls.lMap[k] = v
+		}
 		all = append(all, ls)
 	}
 	return all
 }
 
-func newLabelSet(part config.GenericMap, labels []metrics.MappedLabel) labelSet {
-	ls := make(labelSet, 0, len(labels))
+func newLabelKeyAndMap(name string, part config.GenericMap, labels []metrics.MappedLabel) labelsKeyAndMap {
+	values := make([]string, 0, len(labels)+1)
+	values = append(values, name)
+	m := make(map[string]string, len(labels))
 	for _, t := range labels {
-		label := label{key: t.Target, value: ""}
+		value := ""
 		if v, ok := part[t.Source]; ok {
-			label.value = utils.ConvertToString(v)
+			value = utils.ConvertToString(v)
 		}
-		ls = append(ls, label)
+		values = append(values, value)
+		m[t.Target] = value
 	}
-	return ls
+	return labelsKeyAndMap{values: values, lMap: m}
 }
 
 func (m *MetricsCommonStruct) cleanupExpiredEntriesLoop(callback putils.CacheCallback) {
diff --git a/pkg/pipeline/encode/opentelemetry/encode_otlpmetrics.go b/pkg/pipeline/encode/opentelemetry/encode_otlpmetrics.go
index 8dde163f..dac8fb32 100644
--- a/pkg/pipeline/encode/opentelemetry/encode_otlpmetrics.go
+++ b/pkg/pipeline/encode/opentelemetry/encode_otlpmetrics.go
@@ -19,6 +19,7 @@ package opentelemetry
 
 import (
 	"context"
+	"strings"
 	"time"
 
 	"github.com/netobserv/flowlogs-pipeline/pkg/api"
@@ -64,10 +65,22 @@ func (e *EncodeOtlpMetrics) ProcessCounter(m interface{}, labels map[string]stri
 	return nil
 }
 
-func (e *EncodeOtlpMetrics) ProcessGauge(m interface{}, labels map[string]string, value float64, key string) error {
+func createKey(name string, keys []string) string {
+	key := strings.Builder{}
+	key.WriteString(name)
+	key.WriteRune('|')
+	for _, k := range keys {
+		key.WriteString(k)
+		key.WriteRune('|')
+	}
+	return key.String()
+}
+
+func (e *EncodeOtlpMetrics) ProcessGauge(m interface{}, name string, labels map[string]string, value float64, lvs []string) error {
 	obs := m.(Float64Gauge)
 	// set attributes using the labels
 	attributes := obtainAttributesFromLabels(labels)
+	key := createKey(name, lvs)
 	obs.Set(key, value, attributes)
 	return nil
 }
diff --git a/pkg/pipeline/extract/aggregate/aggregate.go b/pkg/pipeline/extract/aggregate/aggregate.go
index 6c6ecd5c..5e45311c 100644
--- a/pkg/pipeline/extract/aggregate/aggregate.go
+++ b/pkg/pipeline/extract/aggregate/aggregate.go
@@ -43,7 +43,7 @@ const (
 )
 
 type Labels map[string]string
-type NormalizedValues string
+type NormalizedValues []string
 
 type Aggregate struct {
 	definition *api.AggregateDefinition
@@ -78,7 +78,7 @@ func (aggregate *Aggregate) LabelsFromEntry(entry config.GenericMap) (Labels, bo
 }
 
 func (labels Labels) getNormalizedValues() NormalizedValues {
-	var normalizedAsString string
+	var normalized NormalizedValues
 
 	keys := make([]string, 0, len(labels))
 	for k := range labels {
@@ -87,20 +87,16 @@ func (labels Labels) getNormalizedValues() NormalizedValues {
 	sort.Strings(keys)
 
 	for _, k := range keys {
-		normalizedAsString += labels[k] + ","
+		normalized = append(normalized, labels[k])
 	}
 
-	if len(normalizedAsString) > 0 {
-		normalizedAsString = normalizedAsString[:len(normalizedAsString)-1]
-	}
-
-	return NormalizedValues(normalizedAsString)
+	return normalized
 }
 
 func (aggregate *Aggregate) filterEntry(entry config.GenericMap) (NormalizedValues, Labels, error) {
 	labels, allLabelsFound := aggregate.LabelsFromEntry(entry)
 	if !allLabelsFound {
-		return "", nil, fmt.Errorf("missing keys in entry")
+		return nil, nil, fmt.Errorf("missing keys in entry")
 	}
 
 	normalizedValues := labels.getNormalizedValues()
@@ -128,7 +124,7 @@ func (aggregate *Aggregate) UpdateByEntry(entry config.GenericMap, normalizedVal
 	defer aggregate.mutex.Unlock()
 
 	var groupState *GroupState
-	oldEntry, ok := aggregate.cache.GetCacheEntry(string(normalizedValues))
+	oldEntry, ok := aggregate.cache.GetCacheEntry(normalizedValues)
 	if !ok {
 		groupState = &GroupState{normalizedValues: normalizedValues, labels: labels}
 		initVal := getInitValue(string(aggregate.definition.OperationType))
@@ -140,7 +136,7 @@ func (aggregate *Aggregate) UpdateByEntry(entry config.GenericMap, normalizedVal
 	} else {
 		groupState = oldEntry.(*GroupState)
 	}
-	aggregate.cache.UpdateCacheEntry(string(normalizedValues), func() interface{} {
+	aggregate.cache.UpdateCacheEntry(normalizedValues, func() interface{} {
 		return groupState
 	})
 
@@ -212,20 +208,21 @@ func (aggregate *Aggregate) GetMetrics() []config.GenericMap {
 	var metrics []config.GenericMap
 
 	// iterate over the items in the cache
-	aggregate.cache.Iterate(func(_ string, value interface{}) {
+	aggregate.cache.Iterate(func(_ uint64, value interface{}) {
 		group := value.(*GroupState)
+		nv := strings.Join(group.normalizedValues, ",")
 		newEntry := config.GenericMap{
 			"name":              aggregate.definition.Name,
 			"operation_type":    aggregate.definition.OperationType,
 			"operation_key":     aggregate.definition.OperationKey,
 			"by":                strings.Join(aggregate.definition.GroupByKeys, ","),
-			"aggregate":         string(group.normalizedValues),
+			"aggregate":         nv,
 			"total_value":       group.totalValue,
 			"total_count":       group.totalCount,
 			"recent_raw_values": group.recentRawValues,
 			"recent_op_value":   group.recentOpValue,
 			"recent_count":      group.recentCount,
-			strings.Join(aggregate.definition.GroupByKeys, "_"): string(group.normalizedValues),
+			strings.Join(aggregate.definition.GroupByKeys, "_"): nv,
 		}
 		// add the items in aggregate.definition.GroupByKeys individually to the entry
 		for _, key := range aggregate.definition.GroupByKeys {
diff --git a/pkg/pipeline/ingest/ingest_collector.go b/pkg/pipeline/ingest/ingest_collector.go
index 0c9a7c0f..f6d4b246 100644
--- a/pkg/pipeline/ingest/ingest_collector.go
+++ b/pkg/pipeline/ingest/ingest_collector.go
@@ -47,6 +47,8 @@ type ingestCollector struct {
 	hostname   string
 	port       int
 	portLegacy int
+	workers    int
+	sockets    int
 	in         chan map[string]interface{}
 	exitChan   <-chan struct{}
 	metrics    *metrics
@@ -126,30 +128,34 @@ func (c *ingestCollector) initCollectorListener(ctx context.Context) {
 		}
 		defer tpl.Close(ctx)
 
-		go func() {
-			sNF := utils.NewStateNetFlow()
-			sNF.Format = formatter
-			sNF.Transport = transporter
-			sNF.Logger = log.StandardLogger()
-			sNF.TemplateSystem = tpl
-
-			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)
-		}()
+		log.Infof("listening for netflow on host %s, port = %d", c.hostname, c.port)
+		for i := 0; i < c.sockets; i++ {
+			go func() {
+				sNF := utils.NewStateNetFlow()
+				sNF.Format = formatter
+				sNF.Transport = transporter
+				sNF.Logger = log.StandardLogger()
+				sNF.TemplateSystem = tpl
+
+				err = sNF.FlowRoutine(c.workers, c.hostname, c.port, c.sockets > 1)
+				log.Fatal(err)
+			}()
+		}
 	}
 
 	if c.portLegacy > 0 {
-		go func() {
-			sLegacyNF := utils.NewStateNFLegacy()
-			sLegacyNF.Format = formatter
-			sLegacyNF.Transport = transporter
-			sLegacyNF.Logger = log.StandardLogger()
-
-			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)
-		}()
+		log.Infof("listening for legacy netflow on host %s, port = %d", c.hostname, c.portLegacy)
+		for i := 0; i < c.sockets; i++ {
+			go func() {
+				sLegacyNF := utils.NewStateNFLegacy()
+				sLegacyNF.Format = formatter
+				sLegacyNF.Transport = transporter
+				sLegacyNF.Logger = log.StandardLogger()
+
+				err = sLegacyNF.FlowRoutine(c.workers, c.hostname, c.portLegacy, c.sockets > 1)
+				log.Fatal(err)
+			}()
+		}
 	}
 }
 
@@ -177,10 +183,18 @@ func NewIngestCollector(opMetrics *operational.Metrics, params config.StageParam
 	if jsonIngestCollector.Port == 0 && jsonIngestCollector.PortLegacy == 0 {
 		return nil, fmt.Errorf("no ingest port specified")
 	}
+	if jsonIngestCollector.Workers == 0 {
+		jsonIngestCollector.Workers = 1
+	}
+	if jsonIngestCollector.Sockets == 0 {
+		jsonIngestCollector.Sockets = 1
+	}
 
 	log.Infof("hostname = %s", jsonIngestCollector.HostName)
 	log.Infof("port = %d", jsonIngestCollector.Port)
 	log.Infof("portLegacy = %d", jsonIngestCollector.PortLegacy)
+	log.Infof("workers = %d", jsonIngestCollector.Workers)
+	log.Infof("sockets = %d", jsonIngestCollector.Sockets)
 
 	in := make(chan map[string]interface{}, channelSize)
 	metrics := newMetrics(opMetrics, params.Name, params.Ingest.Type, func() int { return len(in) })
@@ -189,6 +203,8 @@ func NewIngestCollector(opMetrics *operational.Metrics, params config.StageParam
 		hostname:   jsonIngestCollector.HostName,
 		port:       jsonIngestCollector.Port,
 		portLegacy: jsonIngestCollector.PortLegacy,
+		workers:    jsonIngestCollector.Workers,
+		sockets:    jsonIngestCollector.Sockets,
 		exitChan:   pUtils.ExitChannel(),
 		in:         in,
 		metrics:    metrics,
diff --git a/pkg/pipeline/transform/transform_network.go b/pkg/pipeline/transform/transform_network.go
index 9547917b..6d6f9f74 100644
--- a/pkg/pipeline/transform/transform_network.go
+++ b/pkg/pipeline/transform/transform_network.go
@@ -127,10 +127,11 @@ func (n *Network) Transform(inputEntry config.GenericMap) (config.GenericMap, bo
 			}
 			if anyIP, ok := outputEntry[rule.AddSubnetLabel.Input]; ok {
 				if strIP, ok := anyIP.(string); ok {
-					lbl, ok := n.ipLabelCache.GetCacheEntry(strIP)
+					keys := []string{strIP}
+					lbl, ok := n.ipLabelCache.GetCacheEntry(keys)
 					if !ok {
 						lbl = n.applySubnetLabel(strIP)
-						n.ipLabelCache.UpdateCacheEntry(strIP, func() interface{} { return lbl })
+						n.ipLabelCache.UpdateCacheEntry(keys, func() interface{} { return lbl })
 					}
 					if lbl != "" {
 						outputEntry[rule.AddSubnetLabel.Output] = lbl
@@ -139,7 +140,7 @@ func (n *Network) Transform(inputEntry config.GenericMap) (config.GenericMap, bo
 			}
 		case api.NetworkDecodeTCPFlags:
 			if anyFlags, ok := outputEntry[rule.DecodeTCPFlags.Input]; ok && anyFlags != nil {
-				if flags, ok := anyFlags.(uint16); ok {
+				if flags, err := util.ConvertToUint(anyFlags); err == nil {
 					flags := util.DecodeTCPFlags(flags)
 					outputEntry[rule.DecodeTCPFlags.Output] = flags
 				}
diff --git a/pkg/pipeline/utils/fnv.go b/pkg/pipeline/utils/fnv.go
new file mode 100644
index 00000000..d9bfd2b7
--- /dev/null
+++ b/pkg/pipeline/utils/fnv.go
@@ -0,0 +1,44 @@
+// Copyright 2018 The Prometheus Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Borrowed from https://github.com/prometheus/client_golang/blob/7c924b7c82b5a61e1e7e2be165bff781700b5365/prometheus/fnv.go
+
+package utils
+
+// Inline and byte-free variant of hash/fnv's fnv64a.
+
+const (
+	offset64 = 14695981039346656037
+	prime64  = 1099511628211
+)
+
+// hashNew initializies a new fnv64a hash value.
+func hashNew() uint64 {
+	return offset64
+}
+
+// hashAdd adds a string to a fnv64a hash value, returning the updated hash.
+func hashAdd(h uint64, s string) uint64 {
+	for i := 0; i < len(s); i++ {
+		h ^= uint64(s[i])
+		h *= prime64
+	}
+	return h
+}
+
+// hashAddByte adds a byte to a fnv64a hash value, returning the updated hash.
+func hashAddByte(h uint64, b byte) uint64 {
+	h ^= uint64(b)
+	h *= prime64
+	return h
+}
diff --git a/pkg/pipeline/utils/timed_cache.go b/pkg/pipeline/utils/timed_cache.go
index 13e3f79d..5264196d 100644
--- a/pkg/pipeline/utils/timed_cache.go
+++ b/pkg/pipeline/utils/timed_cache.go
@@ -35,13 +35,13 @@ var log = logrus.WithField("component", "utils.TimedCache")
 type CacheCallback func(entry interface{})
 
 type cacheEntry struct {
-	key             string
+	key             uint64
 	lastUpdatedTime time.Time
 	e               *list.Element
 	SourceEntry     interface{}
 }
 
-type TimedCacheMap map[string]*cacheEntry
+type TimedCacheMap map[uint64]*cacheEntry
 
 // If maxEntries is non-zero, this limits the number of entries in the cache to the number specified.
 // If maxEntries is zero, the cache has no size limit.
@@ -53,25 +53,36 @@ type TimedCache struct {
 	cacheLenMetric prometheus.Gauge
 }
 
-func (tc *TimedCache) GetCacheEntry(key string) (interface{}, bool) {
+var sep byte = '|'
+
+func computeHash(keys []string) uint64 {
+	h := hashNew()
+	for _, k := range keys {
+		h = hashAdd(h, k)
+		h = hashAddByte(h, sep)
+	}
+	return h
+}
+
+func (tc *TimedCache) GetCacheEntry(keys []string) (interface{}, bool) {
+	h := computeHash(keys)
 	tc.mu.RLock()
 	defer tc.mu.RUnlock()
-	cEntry, ok := tc.cacheMap[key]
+	cEntry, ok := tc.cacheMap[h]
 	if ok {
 		return cEntry.SourceEntry, ok
 	}
 	return nil, ok
 }
 
-var uclog = log.WithField("method", "UpdateCacheEntry")
-
 // If cache entry exists, update its timestamp; if it does not exist, create it if there is room.
 // If we exceed the size of the cache, then do not allocate new entry
-func (tc *TimedCache) UpdateCacheEntry(key string, entryProvider func() interface{}) bool {
+func (tc *TimedCache) UpdateCacheEntry(keys []string, entryProvider func() interface{}) bool {
 	nowInSecs := time.Now()
+	h := computeHash(keys)
 	tc.mu.Lock()
 	defer tc.mu.Unlock()
-	cEntry, ok := tc.cacheMap[key]
+	cEntry, ok := tc.cacheMap[h]
 	if ok {
 		// item already exists in cache; update the element and move to end of list
 		cEntry.lastUpdatedTime = nowInSecs
@@ -84,13 +95,13 @@ func (tc *TimedCache) UpdateCacheEntry(key string, entryProvider func() interfac
 		}
 		cEntry = &cacheEntry{
 			lastUpdatedTime: nowInSecs,
-			key:             key,
+			key:             h,
 			SourceEntry:     entryProvider(),
 		}
-		uclog.Tracef("adding entry: %#v", cEntry)
+		log.Tracef("adding entry: %#v", cEntry)
 		// place at end of list
 		cEntry.e = tc.cacheList.PushBack(cEntry)
-		tc.cacheMap[key] = cEntry
+		tc.cacheMap[h] = cEntry
 		if tc.cacheLenMetric != nil {
 			tc.cacheLenMetric.Inc()
 		}
@@ -107,7 +118,7 @@ func (tc *TimedCache) GetCacheLen() int {
 // We expect that the function calling Iterate might make updates to the entries by calling UpdateCacheEntry()
 // We therefore cannot take the lock at this point since it will conflict with the call in UpdateCacheEntry()
 // TODO: If the callback needs to update the cache, then we need a method to perform it without taking the lock again.
-func (tc *TimedCache) Iterate(f func(key string, value interface{})) {
+func (tc *TimedCache) Iterate(f func(key uint64, value interface{})) {
 	tc.mu.RLock()
 	defer tc.mu.RUnlock()
 	for k, v := range tc.cacheMap {
diff --git a/pkg/pipeline/write/write_loki.go b/pkg/pipeline/write/write_loki.go
index 95e7899e..ec760763 100644
--- a/pkg/pipeline/write/write_loki.go
+++ b/pkg/pipeline/write/write_loki.go
@@ -30,7 +30,6 @@ import (
 	"github.com/netobserv/flowlogs-pipeline/pkg/utils"
 
 	logAdapter "github.com/go-kit/kit/log/logrus"
-	jsonIter "github.com/json-iterator/go"
 	"github.com/netobserv/loki-client-go/loki"
 	"github.com/netobserv/loki-client-go/pkg/backoff"
 	"github.com/netobserv/loki-client-go/pkg/urlutil"
@@ -38,8 +37,6 @@ import (
 	"github.com/sirupsen/logrus"
 )
 
-var jsonEncodingConfig = jsonIter.Config{}.Froze()
-
 var (
 	keyReplacer = strings.NewReplacer("/", "_", ".", "_", "-", "_")
 )
@@ -61,6 +58,7 @@ type Loki struct {
 	timeNow        func() time.Time
 	exitChan       <-chan struct{}
 	metrics        *metrics
+	formatter      func(config.GenericMap) string
 }
 
 func buildLokiConfig(c *api.WriteLoki) (loki.Config, error) {
@@ -110,13 +108,10 @@ func buildLokiConfig(c *api.WriteLoki) (loki.Config, error) {
 func (l *Loki) ProcessRecord(in config.GenericMap) error {
 	labels, lines := l.splitLabelsLines(in)
 
-	js, err := jsonEncodingConfig.Marshal(lines)
-	if err != nil {
-		return err
-	}
-
+	output := l.formatter(lines)
 	timestamp := l.extractTimestamp(lines)
-	err = l.client.Handle(labels, timestamp, string(js))
+
+	err := l.client.Handle(labels, timestamp, output)
 	if err == nil {
 		l.metrics.recordsWritten.Inc()
 	}
@@ -256,6 +251,7 @@ func NewWriteLoki(opMetrics *operational.Metrics, params config.StageParam) (*Lo
 		ignoreList[label] = nil
 	}
 
+	f := formatter(lokiConfigIn.Format, lokiConfigIn.Reorder)
 	l := &Loki{
 		lokiConfig:     lokiConfig,
 		apiConfig:      lokiConfigIn,
@@ -266,6 +262,7 @@ func NewWriteLoki(opMetrics *operational.Metrics, params config.StageParam) (*Lo
 		timeNow:        time.Now,
 		exitChan:       pUtils.ExitChannel(),
 		metrics:        newMetrics(opMetrics, params.Name),
+		formatter:      f,
 	}
 
 	return l, nil
diff --git a/pkg/pipeline/write/write_stdout.go b/pkg/pipeline/write/write_stdout.go
index 5c228305..cb24db27 100644
--- a/pkg/pipeline/write/write_stdout.go
+++ b/pkg/pipeline/write/write_stdout.go
@@ -18,50 +18,66 @@
 package write
 
 import (
-	"encoding/json"
 	"fmt"
-	"os"
 	"sort"
+	"strings"
 	"text/tabwriter"
 	"time"
 
+	jsonIter "github.com/json-iterator/go"
 	"github.com/netobserv/flowlogs-pipeline/pkg/config"
 	"github.com/sirupsen/logrus"
 )
 
 type writeStdout struct {
-	format string
+	formatter func(config.GenericMap) string
 }
 
 // Write writes a flow before being stored
 func (t *writeStdout) Write(v config.GenericMap) {
 	logrus.Tracef("entering writeStdout Write")
-	if t.format == "json" {
-		txt, _ := json.Marshal(v)
-		fmt.Println(string(txt))
-	} else if t.format == "fields" {
-		var order sort.StringSlice
-		for fieldName := range v {
-			order = append(order, fieldName)
+	fmt.Println(t.formatter(v))
+}
+
+func formatter(format string, reorder bool) func(config.GenericMap) string {
+	if format == "json" {
+		jconf := jsonIter.Config{
+			SortMapKeys: reorder,
+		}.Froze()
+		return func(v config.GenericMap) string {
+			b, _ := jconf.Marshal(v)
+			return string(b)
 		}
-		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])
+	} else if format == "fields" {
+		return func(v config.GenericMap) string {
+			var sb strings.Builder
+			var order sort.StringSlice
+			for fieldName := range v {
+				order = append(order, fieldName)
+			}
+			order.Sort()
+			w := tabwriter.NewWriter(&sb, 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()
+			return sb.String()
 		}
-		w.Flush()
-	} else {
-		fmt.Printf("%s: %v\n", time.Now().Format(time.StampMilli), v)
+	}
+	return func(v config.GenericMap) string {
+		return fmt.Sprintf("%v", v)
 	}
 }
 
 // NewWriteStdout create a new write
 func NewWriteStdout(params config.StageParam) (Writer, error) {
 	logrus.Debugf("entering NewWriteStdout")
-	writeStdout := &writeStdout{}
-	if params.Write != nil && params.Write.Stdout != nil {
-		writeStdout.format = params.Write.Stdout.Format
+	var format string
+	if params.Write.Stdout != nil {
+		format = params.Write.Stdout.Format
 	}
-	return writeStdout, nil
+	return &writeStdout{
+		formatter: formatter(format, false),
+	}, nil
 }
diff --git a/pkg/prometheus/prom_server.go b/pkg/prometheus/prom_server.go
index 6353c180..5c05087f 100644
--- a/pkg/prometheus/prom_server.go
+++ b/pkg/prometheus/prom_server.go
@@ -73,9 +73,12 @@ func InitializePrometheus(settings *config.MetricsSettings) *PromServer {
 		return nil
 	}
 	r := prom.DefaultGatherer
-	if settings.SuppressGoMetrics {
+	if settings.SuppressDefaultMetrics {
 		// set up private prometheus registry
-		r = prom.NewRegistry()
+		nr := prom.NewRegistry()
+		prom.DefaultRegisterer = nr
+		prom.DefaultGatherer = nr
+		r = nr
 	}
 	SharedServer = StartServerAsync(&settings.PromConnectionInfo, "", r)
 	return SharedServer
diff --git a/pkg/utils/convert.go b/pkg/utils/convert.go
index 40128bb5..fb752319 100644
--- a/pkg/utils/convert.go
+++ b/pkg/utils/convert.go
@@ -217,6 +217,21 @@ func ConvertToInt(unk interface{}) (int, error) {
 	}
 }
 
+func ConvertToUint(unk interface{}) (uint, error) {
+	switch i := unk.(type) {
+	case uint64:
+		return uint(i), nil
+	case uint32:
+		return uint(i), nil
+	case uint16:
+		return uint(i), nil
+	case uint:
+		return uint(i), nil
+	default:
+		return 0, fmt.Errorf("can't convert %v to uint", i)
+	}
+}
+
 func ConvertToBool(unk interface{}) (bool, error) {
 	switch i := unk.(type) {
 	case string:
diff --git a/pkg/utils/tcp_flags.go b/pkg/utils/tcp_flags.go
index c07d3f37..4bcc55d0 100644
--- a/pkg/utils/tcp_flags.go
+++ b/pkg/utils/tcp_flags.go
@@ -1,7 +1,7 @@
 package utils
 
 type tcpFlag struct {
-	value uint16
+	value uint
 	name  string
 }
 
@@ -19,7 +19,7 @@ var tcpFlags = []tcpFlag{
 	{value: 1024, name: "RST_ACK"},
 }
 
-func DecodeTCPFlags(bitfield uint16) []string {
+func DecodeTCPFlags(bitfield uint) []string {
 	var values []string
 	for _, flag := range tcpFlags {
 		if bitfield&flag.value != 0 {
diff --git a/vendor/github.com/netobserv/netobserv-ebpf-agent/pkg/model/packet_record.go b/vendor/github.com/netobserv/netobserv-ebpf-agent/pkg/model/packet_record.go
index 85af8613..eed8bd84 100644
--- a/vendor/github.com/netobserv/netobserv-ebpf-agent/pkg/model/packet_record.go
+++ b/vendor/github.com/netobserv/netobserv-ebpf-agent/pkg/model/packet_record.go
@@ -18,12 +18,12 @@ type PacketRecord struct {
 // NewPacketRecord contains packet bytes
 func NewPacketRecord(
 	stream []byte,
-	len uint32,
+	length uint32,
 	ts time.Time,
 ) *PacketRecord {
 	pr := PacketRecord{}
 	pr.Time = ts
-	pr.Stream = make([]byte, len)
+	pr.Stream = make([]byte, length)
 	pr.Stream = stream
 	return &pr
 }
diff --git a/vendor/github.com/netobserv/netobserv-ebpf-agent/pkg/model/record.go b/vendor/github.com/netobserv/netobserv-ebpf-agent/pkg/model/record.go
index 255cf21e..1727142a 100644
--- a/vendor/github.com/netobserv/netobserv-ebpf-agent/pkg/model/record.go
+++ b/vendor/github.com/netobserv/netobserv-ebpf-agent/pkg/model/record.go
@@ -39,11 +39,11 @@ type Direction uint8
 // (same behavior as Go's net.IP type)
 type IPAddr [net.IPv6len]uint8
 
-type InterfaceNamer func(ifIndex int) string
+type InterfaceNamer func(ifIndex int, mac MacAddr) string
 
 var (
 	agentIP        net.IP
-	interfaceNamer InterfaceNamer = func(ifIndex int) string { return fmt.Sprintf("[namer unset] %d", ifIndex) }
+	interfaceNamer InterfaceNamer = func(ifIndex int, _ MacAddr) string { return fmt.Sprintf("[namer unset] %d", ifIndex) }
 )
 
 func SetGlobals(ip net.IP, ifaceNamer InterfaceNamer) {
@@ -89,13 +89,17 @@ func NewRecord(
 		TimeFlowEnd:   currentTime.Add(-endDelta),
 		AgentIP:       agentIP,
 	}
-	record.Interfaces = []IntfDirUdn{NewIntfDirUdn(interfaceNamer(int(metrics.IfIndexFirstSeen)),
+	lMAC := metrics.SrcMac
+	if metrics.DirectionFirstSeen == 0 {
+		lMAC = metrics.DstMac
+	}
+	record.Interfaces = []IntfDirUdn{NewIntfDirUdn(interfaceNamer(int(metrics.IfIndexFirstSeen), lMAC),
 		int(metrics.DirectionFirstSeen),
 		udnsCache)}
 
 	for i := uint8(0); i < record.Metrics.NbObservedIntf; i++ {
 		record.Interfaces = append(record.Interfaces, NewIntfDirUdn(
-			interfaceNamer(int(metrics.ObservedIntf[i])),
+			interfaceNamer(int(metrics.ObservedIntf[i]), lMAC),
 			int(metrics.ObservedDirection[i]),
 			udnsCache,
 		))
@@ -185,7 +189,8 @@ func (ia *IPAddr) MarshalJSON() ([]byte, error) {
 }
 
 func (m *MacAddr) String() string {
-	return fmt.Sprintf("%02X:%02X:%02X:%02X:%02X:%02X", m[0], m[1], m[2], m[3], m[4], m[5])
+	t := net.HardwareAddr(m[:])
+	return t.String()
 }
 
 func (m *MacAddr) MarshalJSON() ([]byte, error) {
diff --git a/vendor/modules.txt b/vendor/modules.txt
index d6464085..d1dd2e42 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -317,7 +317,7 @@ github.com/netobserv/loki-client-go/pkg/labelutil
 github.com/netobserv/loki-client-go/pkg/logproto
 github.com/netobserv/loki-client-go/pkg/metric
 github.com/netobserv/loki-client-go/pkg/urlutil
-# github.com/netobserv/netobserv-ebpf-agent v1.9.0-crc0.0.20250528064221-6f34e5b85d2c
+# github.com/netobserv/netobserv-ebpf-agent v1.9.0-crc0.0.20250610144135-d64c5d99f2da
 ## explicit; go 1.23.0
 github.com/netobserv/netobserv-ebpf-agent/pkg/decode
 github.com/netobserv/netobserv-ebpf-agent/pkg/ebpf


Test patch:
diff --git a/cmd/flowlogs-pipeline/main_test.go b/cmd/flowlogs-pipeline/main_test.go
index 2f0a3c8a..6ce73025 100644
--- a/cmd/flowlogs-pipeline/main_test.go
+++ b/cmd/flowlogs-pipeline/main_test.go
@@ -54,7 +54,7 @@ func TestPipelineConfigSetup(t *testing.T) {
     "PipeLine": "[{\"name\":\"grpc\"},{\"follows\":\"grpc\",\"name\":\"enrich\"},{\"follows\":\"enrich\",\"name\":\"loki\"},{\"follows\":\"enrich\",\"name\":\"prometheus\"}]",
     "Parameters": "[{\"ingest\":{\"grpc\":{\"port\":2055},\"type\":\"grpc\"},\"name\":\"grpc\"},{\"name\":\"enrich\",\"transform\":{\"network\":{\"rules\":[{\"kubernetes\":{\"ipField\":\"SrcAddr\",\"output\":\"SrcK8S\"},\"type\":\"add_kubernetes\"},{\"kubernetes\":{\"ipField\":\"DstAddr\",\"output\":\"DstK8S\"},\"type\":\"add_kubernetes\"},{\"add_service\":{\"input\":\"DstPort\",\"output\":\"Service\",\"protocol\":\"Proto\"},\"type\":\"add_service\"},{\"add_subnet\":{\"input\":\"SrcAddr\",\"output\":\"SrcSubnet\",\"subnet_mask\":\"/16\"},\"type\":\"add_subnet\"}]},\"type\":\"network\"}},{\"name\":\"loki\",\"write\":{\"loki\":{\"batchSize\":102400,\"batchWait\":\"1s\",\"clientConfig\":{\"follow_redirects\":false,\"proxy_url\":null,\"tls_config\":{\"insecure_skip_verify\":false}},\"labels\":[\"SrcK8S_Namespace\",\"SrcK8S_OwnerName\",\"DstK8S_Namespace\",\"DstK8S_OwnerName\",\"FlowDirection\"],\"maxBackoff\":\"5m0s\",\"maxRetries\":10,\"minBackoff\":\"1s\",\"staticLabels\":{\"app\":\"netobserv-flowcollector\"},\"tenantID\":\"netobserv\",\"timeout\":\"10s\",\"timestampLabel\":\"TimeFlowEndMs\",\"timestampScale\":\"1ms\",\"url\":\"http://loki.netobserv.svc:3100/\"},\"type\":\"loki\"}},{\"encode\":{\"prom\":{\"metrics\":[{\"buckets\":null,\"labels\":[\"Service\",\"SrcK8S_Namespace\"],\"name\":\"bandwidth_per_network_service_per_namespace\",\"type\":\"counter\",\"valueKey\":\"Bytes\"},{\"buckets\":null,\"labels\":[\"SrcSubnet\"],\"name\":\"bandwidth_per_source_subnet\",\"type\":\"counter\",\"valueKey\":\"Bytes\"},{\"buckets\":null,\"labels\":[\"Service\"],\"name\":\"network_service_total\",\"type\":\"counter\",\"valueKey\":\"\"}],\"prefix\":\"netobserv_\"},\"type\":\"prom\"},\"name\":\"prometheus\"}]",
     "Health": {
-        "Port": "8080"
+        "Port": 8080
     },
     "Profile": {
         "Port": 0
diff --git a/pkg/pipeline/encode/encode_prom_test.go b/pkg/pipeline/encode/encode_prom_test.go
index 7bc77aa9..38ed5c67 100644
--- a/pkg/pipeline/encode/encode_prom_test.go
+++ b/pkg/pipeline/encode/encode_prom_test.go
@@ -819,14 +819,6 @@ func buildFlow() config.GenericMap {
 	}
 }
 
-func thousandsFlows() []config.GenericMap {
-	flows := make([]config.GenericMap, 1000)
-	for i := 0; i < 1000; i++ {
-		flows[i] = buildFlow()
-	}
-	return flows
-}
-
 func BenchmarkPromEncode(b *testing.B) {
 	var expiryTimeDuration api.Duration
 	expiryTimeDuration.Duration = time.Duration(60 * time.Second)
@@ -869,9 +861,8 @@ func BenchmarkPromEncode(b *testing.B) {
 	require.NoError(b, err)
 
 	for i := 0; i < b.N; i++ {
-		for _, metric := range thousandsFlows() {
-			prom.Encode(metric)
-		}
+		m := buildFlow()
+		prom.Encode(m)
 	}
 }
 
diff --git a/pkg/pipeline/extract/aggregate/aggregate_test.go b/pkg/pipeline/extract/aggregate/aggregate_test.go
index da1a0e85..2b70324c 100644
--- a/pkg/pipeline/extract/aggregate/aggregate_test.go
+++ b/pkg/pipeline/extract/aggregate/aggregate_test.go
@@ -62,7 +62,7 @@ func getMockLabels(reverseOrder bool) Labels {
 }
 
 func Test_getNormalizedValues(t *testing.T) {
-	expectedLabels := NormalizedValues("20.0.0.2,10.0.0.1")
+	expectedLabels := NormalizedValues{"20.0.0.2", "10.0.0.1"}
 
 	labels := getMockLabels(false)
 
@@ -103,7 +103,7 @@ func Test_FilterEntry(t *testing.T) {
 	normalizedLabels, labels, err := aggregate.filterEntry(entry)
 	require.Equal(t, err, nil)
 	require.Equal(t, Labels{"srcIP": "10.0.0.1", "dstIP": "20.0.0.2"}, labels)
-	require.Equal(t, NormalizedValues("20.0.0.2,10.0.0.1"), normalizedLabels)
+	require.Equal(t, NormalizedValues{"20.0.0.2", "10.0.0.1"}, normalizedLabels)
 
 	entry = test.GetIngestMockEntry(true)
 
@@ -125,7 +125,7 @@ func Test_Evaluate(t *testing.T) {
 
 	require.Equal(t, nil, err)
 	require.Equal(t, 1, aggregate.cache.GetCacheLen())
-	cacheEntry, found := aggregate.cache.GetCacheEntry(string(normalizedValues))
+	cacheEntry, found := aggregate.cache.GetCacheEntry(normalizedValues)
 	gState := cacheEntry.(*GroupState)
 	require.Equal(t, true, found)
 	require.Equal(t, 2, gState.totalCount)
diff --git a/pkg/pipeline/health_test.go b/pkg/pipeline/health_test.go
index af16c42b..c6db2aa1 100644
--- a/pkg/pipeline/health_test.go
+++ b/pkg/pipeline/health_test.go
@@ -35,7 +35,7 @@ func TestNewHealthServer(t *testing.T) {
 
 	type args struct {
 		pipeline Pipeline
-		port     string
+		port     int
 		address  string
 	}
 	type want struct {
@@ -47,15 +47,15 @@ func TestNewHealthServer(t *testing.T) {
 		args args
 		want want
 	}{
-		{name: "pipeline running", args: args{pipeline: Pipeline{IsRunning: true}, port: "7000", address: "0.0.0.0"}, want: want{statusCode: 200}},
-		{name: "pipeline not running", args: args{pipeline: Pipeline{IsRunning: false}, port: "7001", address: "0.0.0.0"}, want: want{statusCode: 503}},
+		{name: "pipeline running", args: args{pipeline: Pipeline{IsRunning: true}, port: 7000, address: "0.0.0.0"}, want: want{statusCode: 200}},
+		{name: "pipeline not running", args: args{pipeline: Pipeline{IsRunning: false}, port: 7001, address: "0.0.0.0"}, want: want{statusCode: 503}},
 	}
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 
 			opts := config.Options{Health: config.Health{Port: tt.args.port, Address: tt.args.address}}
-			expectedAddr := fmt.Sprintf("%s:%s", opts.Health.Address, opts.Health.Port)
+			expectedAddr := fmt.Sprintf("%s:%d", opts.Health.Address, opts.Health.Port)
 			server := operational.NewHealthServer(&opts, tt.args.pipeline.IsAlive, tt.args.pipeline.IsReady)
 			require.NotNil(t, server)
 			require.Equal(t, expectedAddr, server.Addr)
diff --git a/pkg/pipeline/transform/transform_network_test.go b/pkg/pipeline/transform/transform_network_test.go
index c371aa1c..b1ad903d 100644
--- a/pkg/pipeline/transform/transform_network_test.go
+++ b/pkg/pipeline/transform/transform_network_test.go
@@ -543,3 +543,41 @@ func Test_ValidateReinterpretDirection(t *testing.T) {
 		"FlowDirection": 1,
 	}, output)
 }
+
+func Test_DecodeTCPFlags(t *testing.T) {
+	tr, err := NewTransformNetwork(config.StageParam{
+		Transform: &config.Transform{
+			Network: &api.TransformNetwork{
+				Rules: []api.NetworkTransformRule{
+					{
+						Type: "decode_tcp_flags",
+						DecodeTCPFlags: &api.NetworkGenericRule{
+							Input:  "TcpFlagsU16",
+							Output: "TcpFlagsU16String",
+						},
+					},
+					{
+						Type: "decode_tcp_flags",
+						DecodeTCPFlags: &api.NetworkGenericRule{
+							Input:  "TcpFlagsU32",
+							Output: "TcpFlagsU32String",
+						},
+					},
+				},
+			},
+		},
+	}, nil)
+	require.NoError(t, err)
+
+	output, ok := tr.Transform(config.GenericMap{
+		"TcpFlagsU16": uint16(17),
+		"TcpFlagsU32": uint32(18),
+	})
+	require.True(t, ok)
+	require.Equal(t, config.GenericMap{
+		"TcpFlagsU16":       uint16(17),
+		"TcpFlagsU16String": []string{"FIN", "ACK"},
+		"TcpFlagsU32":       uint32(18),
+		"TcpFlagsU32String": []string{"SYN", "ACK"},
+	}, output)
+}
diff --git a/pkg/pipeline/write/write_loki_test.go b/pkg/pipeline/write/write_loki_test.go
index 49f84ef0..4c526c45 100644
--- a/pkg/pipeline/write/write_loki_test.go
+++ b/pkg/pipeline/write/write_loki_test.go
@@ -41,19 +41,24 @@ const timeout = 5 * time.Second
 
 type fakeEmitter struct {
 	mock.Mock
+	reorderJSON bool
 }
 
 func (f *fakeEmitter) Handle(labels model.LabelSet, timestamp time.Time, record string) error {
-	// sort alphabetically records just for simplifying testing verification with JSON strings
-	recordMap := map[string]interface{}{}
-	if err := json.Unmarshal([]byte(record), &recordMap); err != nil {
-		panic("expected JSON: " + err.Error())
-	}
-	recordBytes, err := json.Marshal(recordMap)
-	if err != nil {
-		panic("error unmarshaling: " + err.Error())
+	if f.reorderJSON {
+		// sort alphabetically records just for simplifying testing verification with JSON strings
+		recordMap := map[string]interface{}{}
+		if err := json.Unmarshal([]byte(record), &recordMap); err != nil {
+			panic("expected JSON: " + err.Error())
+		}
+		recordBytes, err := json.Marshal(recordMap)
+		if err != nil {
+			panic("error unmarshaling: " + err.Error())
+		}
+		a := f.Mock.Called(labels, timestamp, string(recordBytes))
+		return a.Error(0)
 	}
-	a := f.Mock.Called(labels, timestamp, string(recordBytes))
+	a := f.Mock.Called(labels, timestamp, record)
 	return a.Error(0)
 }
 
@@ -122,32 +127,17 @@ parameters:
 }
 
 func TestLoki_ProcessRecord(t *testing.T) {
-	var yamlConfig = `
-log-level: debug
-pipeline:
-  - name: write1
-parameters:
-  - name: write1
-    write:
-      type: loki
-      loki:
-        url: http://loki:3100/
-        timestampLabel: ts
-        ignoreList:
-        - ignored
-        staticLabels:
-          static: label
-        labels:
-          - foo
-          - bar
-`
-	v, cfg := test.InitConfig(t, yamlConfig)
-	require.NotNil(t, v)
-
-	loki, err := NewWriteLoki(operational.NewMetrics(&config.MetricsSettings{}), cfg.Parameters[0])
+	params := api.WriteLoki{
+		URL:            "http://loki:3100/",
+		TimestampLabel: "ts",
+		IgnoreList:     []string{"ignored"},
+		StaticLabels:   model.LabelSet{"static": "label"},
+		Labels:         []string{"foo", "bar"},
+	}
+	loki, err := NewWriteLoki(operational.NewMetrics(&config.MetricsSettings{}), config.StageParam{Write: &config.Write{Loki: &params}})
 	require.NoError(t, err)
 
-	fe := fakeEmitter{}
+	fe := fakeEmitter{reorderJSON: true}
 	fe.On("Handle", mock.Anything, mock.Anything, mock.Anything).Return(nil)
 	loki.client = &fe
 
@@ -171,6 +161,49 @@ parameters:
 	}, time.Unix(124567, 0), `{"other":"val","ts":124567,"value":5678}`)
 }
 
+func TestLoki_ProcessRecordOrdered(t *testing.T) {
+	params := api.WriteLoki{
+		URL:            "http://loki:3100/",
+		TimestampLabel: "ts",
+		Reorder:        true,
+	}
+	loki, err := NewWriteLoki(operational.NewMetrics(&config.MetricsSettings{}), config.StageParam{Write: &config.Write{Loki: &params}})
+	require.NoError(t, err)
+
+	fe := fakeEmitter{reorderJSON: false}
+	fe.On("Handle", mock.Anything, mock.Anything, mock.Anything).Return(nil)
+	loki.client = &fe
+
+	require.NoError(t, loki.ProcessRecord(map[string]interface{}{"ts": 123456, "c": "c", "e": "e", "a": "a", "d": "d", "b": "b"}))
+
+	fe.AssertCalled(t, "Handle", model.LabelSet{}, time.Unix(123456, 0), `{"a":"a","b":"b","c":"c","d":"d","e":"e","ts":123456}`)
+}
+
+func TestLoki_ProcessRecordGoFormat(t *testing.T) {
+	params := api.WriteLoki{
+		URL:            "http://loki:3100/",
+		TimestampLabel: "ts",
+		Labels:         []string{"foo"},
+		Format:         "printf",
+	}
+	loki, err := NewWriteLoki(operational.NewMetrics(&config.MetricsSettings{}), config.StageParam{Write: &config.Write{Loki: &params}})
+	require.NoError(t, err)
+
+	fe := fakeEmitter{}
+	fe.On("Handle", mock.Anything, mock.Anything, mock.Anything).Return(nil)
+	loki.client = &fe
+
+	// WHEN it processes input records
+	require.NoError(t, loki.ProcessRecord(map[string]interface{}{
+		"ts": 123456, "foo": "fooLabel", "bar": "barLabel", "value": 1234}))
+	require.NoError(t, loki.ProcessRecord(map[string]interface{}{
+		"ts": 124567, "foo": "fooLabel2", "bar": "barLabel2", "value": 5678, "other": "val"}))
+
+	// THEN it forwards the records extracting the timestamp and labels from the configuration
+	fe.AssertCalled(t, "Handle", model.LabelSet{"foo": "fooLabel"}, time.Unix(123456, 0), `map[bar:barLabel ts:123456 value:1234]`)
+	fe.AssertCalled(t, "Handle", model.LabelSet{"foo": "fooLabel2"}, time.Unix(124567, 0), `map[bar:barLabel2 other:val ts:124567 value:5678]`)
+}
+
 func TestTimestampScale(t *testing.T) {
 	// verifies that the unix residual time (below 1-second precision) is properly
 	// incorporated into the timestamp whichever scale it is
@@ -262,26 +295,12 @@ func TestTimestampExtraction_LocalTime(t *testing.T) {
 // Tests that labels are sanitized before being sent to loki.
 // Labels that are invalid even if sanitized are ignored
 func TestSanitizedLabels(t *testing.T) {
-	var yamlConfig = `
-log-level: debug
-pipeline:
-  - name: write1
-parameters:
-  - name: write1
-    write:
-      type: loki
-      loki:
-        url: http://loki:3100/
-        labels:
-          - "fo.o"
-          - "ba-r"
-          - "ba/z"
-          - "ignored?"
-`
-	v, cfg := test.InitConfig(t, yamlConfig)
-	require.NotNil(t, v)
-
-	loki, err := NewWriteLoki(operational.NewMetrics(&config.MetricsSettings{}), cfg.Parameters[0])
+	params := api.WriteLoki{
+		URL:            "http://loki:3100/",
+		TimestampLabel: "ts",
+		Labels:         []string{"fo.o", "ba-r", "ba/z", "ignored?"},
+	}
+	loki, err := NewWriteLoki(operational.NewMetrics(&config.MetricsSettings{}), config.StageParam{Write: &config.Write{Loki: &params}})
 	require.NoError(t, err)
 
 	fe := fakeEmitter{}
diff --git a/pkg/pipeline/write/write_stdout_test.go b/pkg/pipeline/write/write_stdout_test.go
index 90c305b4..74c40d11 100644
--- a/pkg/pipeline/write/write_stdout_test.go
+++ b/pkg/pipeline/write/write_stdout_test.go
@@ -20,17 +20,13 @@ package write
 import (
 	"testing"
 
+	"github.com/netobserv/flowlogs-pipeline/pkg/api"
 	"github.com/netobserv/flowlogs-pipeline/pkg/config"
 	"github.com/stretchr/testify/require"
 )
 
-func Test_WriteStdout(_ *testing.T) {
-	ws := writeStdout{}
-	ws.Write(config.GenericMap{"key": "test"})
-}
-
 func Test_NewWriteStdout(t *testing.T) {
-	writer, err := NewWriteStdout(config.StageParam{})
-	require.Nil(t, err)
-	require.Equal(t, writer, &writeStdout{})
+	writer, err := NewWriteStdout(config.StageParam{Write: &config.Write{Stdout: &api.WriteStdout{}}})
+	require.NoError(t, err)
+	writer.Write(config.GenericMap{"key": "test"})
 }