Repository: netobserv/flowlogs-pipeline (go)

Issue:
loki.write order of fields is different/random compared to the stdout format
Hello,

I collect IPFIX flows and compared it with "stdout" and "loki.write"

stdout. the logs have the same order of labels and values for every flow/log
```
May  2 22:50:31.379: map[AsPath:[] BgpCommunities:[] BgpNextHop:[] BiFlowDirection:0 Bytes:168 CustomBytes_1:[] CustomBytes_2:[] CustomBytes_3:[] CustomBytes_4:[] CustomBytes_5:[] CustomInteger_1:0 CustomInteger_2:0 CustomInteger_3:0 CustomInteger_4:0 CustomInteger_5:0 CustomList_1:[] DstAddr:10.30.98.1 DstAs:0 DstMac:00:00:00:00:00:00 DstNet:0 DstPort:0 DstVlan:0 EgressVrfId:0 Etype:2048 FlowDirection:0 ForwardingStatus:0 FragmentId:0 FragmentOffset:0 HasMpls:false IcmpCode:0 IcmpType:0 InIf:168 IngressVrfId:0 IpTos:0 IpTtl:0 Ipv6FlowLabel:0 MplsCount:0 MplsLabelIp:[] MplsLastLabel:0 MplsLastTtl:0 Mpls_1Label:0 Mpls_1Ttl:0 Mpls_2Label:0 Mpls_2Ttl:0 Mpls_3Label:0 Mpls_3Ttl:0 NextHop:[] NextHopAs:0 ObservationDomainId:0 ObservationPointId:0 OutIf:168 Packets:2 Proto:1 SamplerAddress:[10 30 102 1] SamplingRate:0 SequenceNum:2366368352 SrcAddr:10.30.98.19 SrcAs:0 SrcMac:00:00:00:00:00:00 SrcNet:0 SrcPort:25742 SrcVlan:0 TcpFlags:0 TimeFlowEnd:1746218922 TimeFlowEndMs:1746218922000 TimeFlowStart:1746218921 TimeFlowStartMs:1746218921000 TimeReceived:1746219031 Type:IPFIX VlanId:0]
May  2 22:50:31.379: map[AsPath:[] BgpCommunities:[] BgpNextHop:[] BiFlowDirection:0 Bytes:4046597 CustomBytes_1:[] CustomBytes_2:[] CustomBytes_3:[] CustomBytes_4:[] CustomBytes_5:[] CustomInteger_1:0 CustomInteger_2:0 CustomInteger_3:0 CustomInteger_4:0 CustomInteger_5:0 CustomList_1:[] DstAddr:10.30.67.101 DstAs:0 DstMac:00:00:00:00:00:00 DstNet:0 DstPort:443 DstVlan:0 EgressVrfId:0 Etype:2048 FlowDirection:0 ForwardingStatus:0 FragmentId:0 FragmentOffset:0 HasMpls:false IcmpCode:0 IcmpType:0 InIf:170 IngressVrfId:0 IpTos:0 IpTtl:0 Ipv6FlowLabel:0 MplsCount:0 MplsLabelIp:[] MplsLastLabel:0 MplsLastTtl:0 Mpls_1Label:0 Mpls_1Ttl:0 Mpls_2Label:0 Mpls_2Ttl:0 Mpls_3Label:0 Mpls_3Ttl:0 NextHop:[] NextHopAs:0 ObservationDomainId:0 ObservationPointId:0 OutIf:164 Packets:3598 Proto:6 SamplerAddress:[10 30 102 1] SamplingRate:0 SequenceNum:2366368352 SrcAddr:10.30.93.83 SrcAs:0 SrcMac:00:00:00:00:00:00 SrcNet:0 SrcPort:50414 SrcVlan:0 TcpFlags:17 TimeFlowEnd:1746218921 TimeFlowEndMs:1746218921000 TimeFlowStart:1746218911 TimeFlowStartMs:1746218911000 TimeReceived:1746219031 Type:IPFIX VlanId:0]
May  2 22:50:31.379: map[AsPath:[] BgpCommunities:[] BgpNextHop:[] BiFlowDirection:0 Bytes:426594 CustomBytes_1:[] CustomBytes_2:[] CustomBytes_3:[] CustomBytes_4:[] CustomBytes_5:[] CustomInteger_1:0 CustomInteger_2:0 CustomInteger_3:0 CustomInteger_4:0 CustomInteger_5:0 CustomList_1:[] DstAddr:10.30.93.83 DstAs:0 DstMac:00:00:00:00:00:00 DstNet:0 DstPort:50414 DstVlan:0 EgressVrfId:0 Etype:2048 FlowDirection:0 ForwardingStatus:0 FragmentId:0 FragmentOffset:0 HasMpls:false IcmpCode:0 IcmpType:0 InIf:164 IngressVrfId:0 IpTos:0 IpTtl:0 Ipv6FlowLabel:0 MplsCount:0 MplsLabelIp:[] MplsLastLabel:0 MplsLastTtl:0 Mpls_1Label:0 Mpls_1Ttl:0 Mpls_2Label:0 Mpls_2Ttl:0 Mpls_3Label:0 Mpls_3Ttl:0 NextHop:[] NextHopAs:0 ObservationDomainId:0 ObservationPointId:0 OutIf:170 Packets:1252 Proto:6 SamplerAddress:[10 30 102 1] SamplingRate:0 SequenceNum:2366368352 SrcAddr:10.30.67.101 SrcAs:0 SrcMac:00:00:00:00:00:00 SrcNet:0 SrcPort:443 SrcVlan:0 TcpFlags:17 TimeFlowEnd:1746218921 TimeFlowEndMs:1746218921000 TimeFlowStart:1746218911 TimeFlowStartMs:1746218911000 TimeReceived:1746219031 Type:IPFIX VlanId:0]
May  2 22:50:31.379: map[AsPath:[] BgpCommunities:[] BgpNextHop:[] BiFlowDirection:0 Bytes:193 CustomBytes_1:[] CustomBytes_2:[] CustomBytes_3:[] CustomBytes_4:[] CustomBytes_5:[] CustomInteger_1:0 CustomInteger_2:0 CustomInteger_3:0 CustomInteger_4:0 CustomInteger_5:0 CustomList_1:[] DstAddr:10.30.75.11 DstAs:0 DstMac:00:00:00:00:00:00 DstNet:0 DstPort:22 DstVlan:0 EgressVrfId:0 Etype:2048 FlowDirection:0 ForwardingStatus:0 FragmentId:0 FragmentOffset:0 HasMpls:false IcmpCode:0 IcmpType:0 InIf:170 IngressVrfId:0 IpTos:0 IpTtl:0 Ipv6FlowLabel:0 MplsCount:0 MplsLabelIp:[] MplsLastLabel:0 MplsLastTtl:0 Mpls_1Label:0 Mpls_1Ttl:0 Mpls_2Label:0 Mpls_2Ttl:0 Mpls_3Label:0 Mpls_3Ttl:0 NextHop:[] NextHopAs:0 ObservationDomainId:0 ObservationPointId:0 OutIf:167 Packets:4 Proto:6 SamplerAddress:[10 30 102 1] SamplingRate:0 SequenceNum:2366368352 SrcAddr:10.204.34.21 SrcAs:0 SrcMac:00:00:00:00:00:00 SrcNet:0 SrcPort:46143 SrcVlan:0 TcpFlags:17 TimeFlowEnd:1746218921 TimeFlowEndMs:1746218921000 TimeFlowStart:1746218921 TimeFlowStartMs:1746218921000 TimeReceived:1746219031 Type:IPFIX VlanId:0]
May  2 22:50:31.380: map[AsPath:[] BgpCommunities:[] BgpNextHop:[] BiFlowDirection:0 Bytes:212 CustomBytes_1:[] CustomBytes_2:[] CustomBytes_3:[] CustomBytes_4:[] CustomBytes_5:[] CustomInteger_1:0 CustomInteger_2:0 CustomInteger_3:0 CustomInteger_4:0 CustomInteger_5:0 CustomList_1:[] DstAddr:10.204.34.21 DstAs:0 DstMac:00:00:00:00:00:00 DstNet:0 DstPort:46143 DstVlan:0 EgressVrfId:0 Etype:2048 FlowDirection:0 ForwardingStatus:0 FragmentId:0 FragmentOffset:0 HasMpls:false IcmpCode:0 IcmpType:0 InIf:167 IngressVrfId:0 IpTos:0 IpTtl:0 Ipv6FlowLabel:0 MplsCount:0 MplsLabelIp:[] MplsLastLabel:0 MplsLastTtl:0 Mpls_1Label:0 Mpls_1Ttl:0 Mpls_2Label:0 Mpls_2Ttl:0 Mpls_3Label:0 Mpls_3Ttl:0 NextHop:[] NextHopAs:0 ObservationDomainId:0 ObservationPointId:0 OutIf:170 Packets:5 Proto:6 SamplerAddress:[10 30 102 1] SamplingRate:0 SequenceNum:2366368352 SrcAddr:10.30.75.11 SrcAs:0 SrcMac:00:00:00:00:00:00 SrcNet:0 SrcPort:22 SrcVlan:0 TcpFlags:24 TimeFlowEnd:1746218921 TimeFlowEndMs:1746218921000 TimeFlowStart:1746218921 TimeFlowStartMs:1746218921000 TimeReceived:1746219031 Type:IPFIX VlanId:0]
```

If I compare this with the results in grafana loki the order is totally random:

![Image](https://github.com/user-attachments/assets/970a7824-9838-41f3-9569-5d7c6d6fa9e7)


I would like to have the logs in loki the same as in the stdout output. at least in the same order for all values.

Grafana Loki allows the "Pattern Parser" which is fast and easy to create. I would like to create Dashboards with predefines "pattern parser" queries which allows the user to type in IPs etc. For this I need a consistent log format for all flows.

https://grafana.com/blog/2021/08/09/new-in-loki-2.3-logql-pattern-parser-makes-it-easier-to-extract-data-from-unstructured-logs/

Pull request:
Implement more format on Loki writer
- Implement same formats as stdout writer (printf, json, fields)
- Allow reordering json keys
- Refactor and add more tests

Fixes #941 

Examples:
To send logs as plain go structures:

```yaml
    write:
      type: loki
      loki:
      # ...
        format: printf
```

To send logs as ordered json

```yaml
    write:
      type: loki
      loki:
      # ...
        format: json
        reorder: true
```


Patch:
diff --git a/docs/api.md b/docs/api.md
index e60a31ba..4b3e3f85 100644
--- a/docs/api.md
+++ b/docs/api.md
@@ -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:
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/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/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
 }


Test patch:
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"})
 }