[filebeat] convert netflow input to API v2 (#37901)

* feat: migrate netflow input to v2.Plugin

* fix: replace deprecated ioutil usage

* fix: remove unnecessary type conversions

* doc: update CHANGELOG.next.asciidoc

* ci: install missing libpcap-dev for linux lint step

* fix: utilise require.Eventually to wait for conditions to become true in netflow_test.go

* fix: proper log verbosity for errors during stopping netflow input

* fix: remove redundant config creation

* fix: utilise a std context in statsLoop

* fix: add server in udp regarding log messages

* fix: remove redundant empty line in CHANGELOG.next.asciidoc

* fix: enrich more the messages of eventual condition in netflow tests

Author: pkoutsovasilis

.github/workflows/golangci-lint.yml

@@ -32,6 +32,9 @@ jobs:
         with:
           go-version-file: .go-version
 
+      - name: Install Apt Package
+        run: sudo apt-get update && sudo apt-get install -y libpcap-dev
+
       - name: golangci-lint
         env:
           GOOS: ${{ matrix.GOOS }}

CHANGELOG.next.asciidoc

@@ -24,6 +24,8 @@ fields added to events containing the Beats version. {pull}37553[37553]
 
 *Filebeat*
 
+- Convert netflow input to API v2 and disable event normalisation {pull}37901[37901]
+
 
 *Heartbeat*
 

x-pack/dockerlogbeat/pipelinemock/pipelines.go

@@ -85,3 +85,11 @@ func (pc *MockPipelineConnector) ConnectWith(beat.ClientConfig) (beat.Client, er
 
 	return c, nil
 }
+
+// HasConnectedClients returns true if there are clients connected.
+func (pc *MockPipelineConnector) HasConnectedClients() bool {
+	pc.mtx.Lock()
+	defer pc.mtx.Unlock()
+
+	return len(pc.clients) > 0
+}

x-pack/dockerlogbeat/pipelinemock/reader.go

@@ -8,7 +8,6 @@ import (
 	"bytes"
 	"encoding/binary"
 	"io"
-	"io/ioutil"
 	"testing"
 
 	"github.com/docker/docker/api/types/plugins/logdriver"
@@ -33,7 +32,7 @@ func CreateTestInputFromLine(t *testing.T, line string) io.ReadCloser {
 	writer := new(bytes.Buffer)
 
 	encodeLog(t, writer, exampleStruct)
-	return ioutil.NopCloser(writer)
+	return io.NopCloser(writer)
 }
 
 func encodeLog(t *testing.T, out io.Writer, entry *logdriver.LogEntry) {

x-pack/filebeat/input/default-inputs/inputs_other.go

@@ -20,6 +20,7 @@ import (
 	"github.com/elastic/beats/v7/x-pack/filebeat/input/http_endpoint"
 	"github.com/elastic/beats/v7/x-pack/filebeat/input/httpjson"
 	"github.com/elastic/beats/v7/x-pack/filebeat/input/lumberjack"
+	"github.com/elastic/beats/v7/x-pack/filebeat/input/netflow"
 	"github.com/elastic/beats/v7/x-pack/filebeat/input/o365audit"
 	"github.com/elastic/beats/v7/x-pack/filebeat/input/shipper"
 	"github.com/elastic/beats/v7/x-pack/filebeat/input/websocket"
@@ -41,5 +42,6 @@ func xpackInputs(info beat.Info, log *logp.Logger, store beater.StateStore) []v2
 		lumberjack.Plugin(),
 		shipper.Plugin(log, store),
 		websocket.Plugin(log, store),
+		netflow.Plugin(log),
 	}
 }

x-pack/filebeat/input/netflow/case.go

@@ -60,9 +60,8 @@ func CamelCaseToSnakeCase(in string) string {
 	}
 
 	out := make([]rune, 0, len(in)+4)
-	runes := []rune(in)
 	upperCount := 1
-	for _, r := range runes {
+	for _, r := range in {
 		lr := unicode.ToLower(r)
 		isUpper := lr != r
 		if isUpper {

x-pack/filebeat/input/netflow/definitions.go

@@ -7,7 +7,7 @@ package netflow
 import (
 	"errors"
 	"fmt"
-	"io/ioutil"
+	"io"
 	"math"
 	"os"
 	"strconv"
@@ -95,7 +95,7 @@ func LoadFieldDefinitionsFromFile(path string) (defs fields.FieldDict, err error
 		return nil, err
 	}
 	defer file.Close()
-	contents, err := ioutil.ReadAll(file)
+	contents, err := io.ReadAll(file)
 	if err != nil {
 		return nil, err
 	}
@@ -169,7 +169,7 @@ func loadFields(def map[interface{}]interface{}, pem uint32, dest fields.FieldDi
 			return fmt.Errorf("bad field ID %d: should have two items (type, name) or one (:skip) (Got %+v)", fieldID, list)
 		}
 		key := fields.Key{
-			EnterpriseID: uint32(pem),
+			EnterpriseID: pem,
 			FieldID:      uint16(fieldID),
 		}
 		if _, exists := dest[key]; exists {