Add record ordering

Author: phaag

example/reader/main.go

@@ -45,7 +45,7 @@ func main() {
 	fmt.Printf("nffile:\n%v", nffile)
 
 	// Dump flow records
-	recordChannel, _ := nffile.AllRecords()
+	recordChannel, _ := nffile.AllRecords().Get()
 	cnt := 0
 	for record := range recordChannel {
 		cnt++

example/sorter/main.go

@@ -0,0 +1,54 @@
+// Copyright © 2024 Peter Haag [email protected]
+// All rights reserved.
+//
+// Use of this source code is governed by the license that can be
+// found in the LICENSE file.
+
+package main
+
+import (
+	"flag"
+	"fmt"
+	"os"
+
+	nfdump "github.com/phaag/go-nfdump"
+)
+
+var (
+	fileName = flag.String("r", "", "nfdump file to read")
+)
+
+func main() {
+
+	flag.CommandLine.Usage = func() {
+		fmt.Fprintf(os.Stderr, "Usage of %s [flags]\n", os.Args[0])
+		flag.PrintDefaults()
+	}
+
+	flag.Parse()
+
+	if len(*fileName) == 0 {
+		fmt.Printf("Filename required\n")
+		flag.PrintDefaults()
+		os.Exit(255)
+	}
+
+	// get empty new nffile object
+	nffile := nfdump.New()
+
+	// open the flow file
+	if err := nffile.Open(*fileName); err != nil {
+		fmt.Printf("Failed to open nf file: %v\n", err)
+		os.Exit(255)
+	}
+
+	// Read all flow records and append the OrderBy() processing
+	// finally get the flows and print them
+	if recordChannel, err := nffile.AllRecords().OrderBy("bytes", nfdump.DESCENDING).Get(); err != nil {
+		fmt.Printf("Failed to process flows: %v\n", err)
+	} else {
+		for record := range recordChannel {
+			record.PrintLine()
+		}
+	}
+}

go.mod

@@ -6,4 +6,5 @@ require (
 	github.com/klauspost/compress v1.17.8
 	github.com/pierrec/lz4/v4 v4.1.17
 	github.com/rasky/go-lzo v0.0.0-20200203143853-96a758eda86e
+	github.com/twotwotwo/sorts v0.0.0-20160814051341-bf5c1f2b8553
 )

go.sum

@@ -4,3 +4,5 @@ github.com/pierrec/lz4/v4 v4.1.17 h1:kV4Ip+/hUBC+8T6+2EgburRtkE9ef4nbY3f4dFhGjMc
 github.com/pierrec/lz4/v4 v4.1.17/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
 github.com/rasky/go-lzo v0.0.0-20200203143853-96a758eda86e h1:dCWirM5F3wMY+cmRda/B1BiPsFtmzXqV9b0hLWtVBMs=
 github.com/rasky/go-lzo v0.0.0-20200203143853-96a758eda86e/go.mod h1:9leZcVcItj6m9/CfHY5Em/iBrCz7js8LcRQGTKEEv2M=
+github.com/twotwotwo/sorts v0.0.0-20160814051341-bf5c1f2b8553 h1:DRC1ubdb3ZmyyIeCSTxjZIQAnpLPfKVgYrLETQuOPjo=
+github.com/twotwotwo/sorts v0.0.0-20160814051341-bf5c1f2b8553/go.mod h1:Rj7Csq/tZ/egz+Ltc2IVpsA5309AmSMEswjkTZmq2Xc=

nffile.go

@@ -265,7 +265,8 @@ func (nfFile *NfFile) ReadDataBlocks() (chan DataBlock, error) {
 // AllRecord takes an NfFile object and returns a channel of FlowRecordV3
 // it reads and uncompresses the data blocks with ReadDataBlocks
 // Iterating over the channel reads all flow records
-func (nfFile *NfFile) AllRecords() (chan *FlowRecordV3, error) {
+// returns a record chain type
+func (nfFile *NfFile) AllRecords() *RecordChain {
 	recordChannel := make(chan *FlowRecordV3, 32)
 	go func() {
 		blockChannel, _ := nfFile.ReadDataBlocks()
@@ -306,5 +307,9 @@ func (nfFile *NfFile) AllRecords() (chan *FlowRecordV3, error) {
 		}
 		close(recordChannel)
 	}()
-	return recordChannel, nil
+
+	chain := new(RecordChain)
+	chain.recordChan = recordChannel
+	chain.err = nil
+	return chain
 }

orderby.go

@@ -0,0 +1,176 @@
+// Copyright © 2024 Peter Haag [email protected]
+// All rights reserved.
+//
+// Use of this source code is governed by the license that can be
+// found in the LICENSE file.
+
+package nfdump
+
+import (
+	"fmt"
+
+	"github.com/twotwotwo/sorts"
+)
+
+type sortRecord struct {
+	index uint32
+	value uint64
+}
+
+type sortType []sortRecord
+
+var sortArray []sortRecord
+var recordArray []*FlowRecordV3
+
+// sort direction ASCENDING
+const ASCENDING = 1
+
+// sort direction DESCENDING
+const DESCENDING = 2
+
+// implement sorts interface
+// return len of sorting array
+func (a sortType) Len() int { return len(a) }
+
+// swap 2 elements
+func (a sortType) Swap(i, j int) { a[i], a[j] = a[j], a[i] }
+
+// return key of element i
+func (a sortType) Key(i int) uint64 {
+	return a[i].value
+}
+
+// compare tw values for less
+func (a sortType) Less(i, j int) bool {
+	return a[i].value < a[j].value
+}
+
+// value functions
+// return appropriate values
+type valueFuncType func(record *FlowRecordV3) uint64
+
+// tstart in msec
+func getTstart(record *FlowRecordV3) uint64 {
+	var value uint64
+	if genericFlow := record.GenericFlow(); genericFlow != nil {
+		value = genericFlow.MsecFirst
+	}
+	return value
+}
+
+// tend in msec
+func getTend(record *FlowRecordV3) uint64 {
+	var value uint64
+	if genericFlow := record.GenericFlow(); genericFlow != nil {
+		value = genericFlow.MsecFirst
+	}
+	return value
+}
+
+// packets
+func getPackets(record *FlowRecordV3) uint64 {
+	var value uint64
+	if genericFlow := record.GenericFlow(); genericFlow != nil {
+		value = genericFlow.InPackets
+	}
+	return value
+}
+
+// bytes
+func getBytes(record *FlowRecordV3) uint64 {
+	var value uint64
+	if genericFlow := record.GenericFlow(); genericFlow != nil {
+		value = genericFlow.InBytes
+	}
+	return value
+}
+
+// order option - name and function
+type orderOption struct {
+	name      string
+	orderFunc valueFuncType
+}
+
+// list all possible orderBy options
+var orderTable = []orderOption{
+	orderOption{"tstart", getTstart},
+	orderOption{"tend", getTend},
+	orderOption{"packets", getPackets},
+	orderOption{"bytes", getBytes},
+}
+
+// function used recordChain as input, sort the records by orderBy
+// accepts and orderBy, defined in the order table as name
+// direction is einer ASCENDING or DESCENDING
+// returns chain element with channel of sorted records
+func (recordChain *RecordChain) OrderBy(orderBy string, direction int) *RecordChain {
+	// propagate error
+	if recordChain.err != nil {
+		return &RecordChain{recordChan: nil, err: recordChain.err}
+	}
+
+	var valueFunc valueFuncType
+	for i := 0; i < len(orderTable); i++ {
+		if orderBy == orderTable[i].name {
+			valueFunc = orderTable[i].orderFunc
+		}
+	}
+
+	if valueFunc == nil {
+		return &RecordChain{recordChan: nil, err: fmt.Errorf("Unknown orderBy: %s", orderBy)}
+	}
+
+	writeChan := make(chan *FlowRecordV3, 64)
+
+	// store all flow records into an array for later printing
+	recordArray = make([]*FlowRecordV3, 1024*1024)
+
+	// store value to be sorted and index of appropriate flow record of
+	// recordArray. Keeps sortArray smaller - cache friendly
+	sortArray = make([]sortRecord, 1024*1024)
+
+	go func(readChan chan *FlowRecordV3) {
+
+		var arrayLen = len(sortArray)
+		// use direct access [cnt] to slice to speed up instead of append()
+		// increase array if needed
+		var cnt uint32 = 0
+		for record := range readChan {
+			if uint32(arrayLen)-cnt == 0 {
+				// double array, if exhausted
+				sortArray = append(make([]sortRecord, 2*arrayLen), sortArray...)
+				recordArray = append(make([]*FlowRecordV3, 2*arrayLen), recordArray...)
+
+				// use new len of array. Go may assign more memory than requested
+				// so use actual len
+				arrayLen = len(sortArray)
+			}
+			recordArray[cnt] = record
+			value := valueFunc(record)
+			sortArray[cnt] = sortRecord{cnt, value}
+			cnt++
+		}
+
+		// sort array
+		// the interface makes use of len() - therefore cut slice pointer to real size
+		sorts.ByUint64(sortType(sortArray[0:cnt]))
+
+		if direction == ASCENDING {
+			for i := 0; i < int(cnt); i++ {
+				index := sortArray[i].index
+				record := recordArray[index]
+				writeChan <- record
+			}
+		} else {
+			for i := int(cnt) - 1; i >= 0; i-- {
+				index := sortArray[i].index
+				record := recordArray[index]
+				writeChan <- record
+			}
+		}
+		close(writeChan)
+
+	}(recordChain.recordChan)
+
+	return &RecordChain{recordChan: writeChan, err: nil}
+} // End of OrderBy

printLine.go

@@ -0,0 +1,41 @@
+// Copyright © 2024 Peter Haag [email protected]
+// All rights reserved.
+//
+// Use of this source code is governed by the license that can be
+// found in the LICENSE file.
+
+package nfdump
+
+import (
+	"fmt"
+	"time"
+)
+
+func fmtDuration(d uint64) string {
+	msec := d % 1000
+	d = (d - msec) / 1000
+	sec := d % 60
+	d = (d - sec) / 60
+	min := d % 60
+	d = (d - min) / 60
+	hour := d % 24
+	days := (d - hour) / 24
+	if days == 0 {
+		return fmt.Sprintf("   %02d:%02d:%02d.%03d", hour, min, sec, msec)
+	} else {
+		return fmt.Sprintf("%02dd %02d:%02d:%02d.%03d", days, hour, min, sec, msec)
+	}
+}
+
+// Return generic extension
+func (flowRecord *FlowRecordV3) PrintLine() {
+	if genericFlow := flowRecord.GenericFlow(); genericFlow != nil {
+		tTime := time.UnixMilli(int64(genericFlow.MsecFirst))
+		duration := genericFlow.MsecLast - genericFlow.MsecFirst
+		ipAddr := flowRecord.IP()
+		if ipAddr != nil {
+			fmt.Printf("%s %s %3d %15v:%-5v -> %15v:%-5v %5d %7d\n", tTime.Format("2006-01-02 15:04:05.000"), fmtDuration(duration), genericFlow.Proto,
+				ipAddr.SrcIP, genericFlow.SrcPort, ipAddr.DstIP, genericFlow.DstPort, genericFlow.InPackets, genericFlow.InBytes)
+		}
+	}
+}

record.go

@@ -40,7 +40,18 @@ type FlowRecordV3 struct {
 	extOffset      [MAXEXTENSIONS]elementParam
 }
 
-// Extract next flow record from []byte stream
+// type to return/accept in the flow processing record chain
+type RecordChain struct {
+	recordChan chan *FlowRecordV3
+	err        error
+}
+
+// at the end of flow processing return a channel of all records
+func (recordChain *RecordChain) Get() (chan *FlowRecordV3, error) {
+	return recordChain.recordChan, recordChain.err
+}
+
+// Extract the next flow record from []byte stream
 func NewRecord(record []byte) (*FlowRecordV3, error) {
 
 	offset := 0