[Netflow] Migrate from log to the logp logging library

The rest of beats is using github.com/elastic/elastic-agent-libs/logp, so make Netflow use it, too.
Cleaned up the test code as well

x-pack/filebeat/input/netflow/decoder/config/config.go

@@ -5,10 +5,10 @@
 package config
 
 import (
-	"io"
 	"time"
 
 	"github.com/elastic/beats/v7/x-pack/filebeat/input/netflow/decoder/fields"
+	"github.com/elastic/elastic-agent-libs/logp"
 )
 
 type ActiveSessionsMetric interface {
@@ -19,7 +19,7 @@ type ActiveSessionsMetric interface {
 // Config stores the configuration used by the NetFlow Collector.
 type Config struct {
 	protocols            []string
-	logOutput            io.Writer
+	logOutput            *logp.Logger
 	expiration           time.Duration
 	detectReset          bool
 	fields               fields.FieldDict
@@ -28,21 +28,22 @@ type Config struct {
 	activeSessionsMetric ActiveSessionsMetric
 }
 
-var defaultCfg = Config{
-	protocols:       []string{},
-	logOutput:       io.Discard,
-	expiration:      time.Hour,
-	detectReset:     true,
-	sharedTemplates: false,
-	withCache:       false,
-}
-
 // Defaults returns a configuration object with defaults settings:
 // - no protocols are enabled.
-// - log output is discarded
+// - log output is set to the logger that is passed in.
 // - session expiration is checked once every hour.
-func Defaults() Config {
-	return defaultCfg
+// - resets are detected.
+// - templates are not shared.
+// - cache is disabled.
+func Defaults(logger *logp.Logger) Config {
+	return Config{
+		protocols:       []string{},
+		logOutput:       logger,
+		expiration:      time.Hour,
+		detectReset:     true,
+		sharedTemplates: false,
+		withCache:       false,
+	}
 }
 
 // WithProtocols modifies an existing configuration object to enable the
@@ -52,12 +53,6 @@ func (c *Config) WithProtocols(protos ...string) *Config {
 	return c
 }
 
-// WithLogOutput sets the output io.Writer for logging.
-func (c *Config) WithLogOutput(output io.Writer) *Config {
-	c.logOutput = output
-	return c
-}
-
 // WithExpiration configures the expiration timeout for sessions and templates.
 // A value of zero disables expiration.
 func (c *Config) WithExpiration(timeout time.Duration) *Config {
@@ -121,7 +116,7 @@ func (c *Config) Protocols() []string {
 }
 
 // LogOutput returns the io.Writer where logs are to be written.
-func (c *Config) LogOutput() io.Writer {
+func (c *Config) LogOutput() *logp.Logger {
 	return c.logOutput
 }
 

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

@@ -14,6 +14,8 @@ import (
 	"net"
 	"sync"
 
+	"github.com/elastic/elastic-agent-libs/logp"
+
 	"github.com/elastic/beats/v7/x-pack/filebeat/input/netflow/decoder/config"
 	"github.com/elastic/beats/v7/x-pack/filebeat/input/netflow/decoder/protocol"
 	"github.com/elastic/beats/v7/x-pack/filebeat/input/netflow/decoder/record"
@@ -93,8 +95,8 @@ func (p *Decoder) Read(buf *bytes.Buffer, source net.Addr) (records []record.Rec
 }
 
 // NewConfig returns a new configuration structure to be passed to NewDecoder.
-func NewConfig() *config.Config {
-	cfg := config.Defaults()
+func NewConfig(logger *logp.Logger) *config.Config {
+	cfg := config.Defaults(logger)
 	return &cfg
 }
 

x-pack/filebeat/input/netflow/decoder/examples/go-netflow-example.go

@@ -8,51 +8,52 @@ import (
 	"bytes"
 	"encoding/json"
 	"fmt"
-	"log"
 	"net"
-	"os"
+
+	"github.com/elastic/elastic-agent-libs/logp"
 
 	"github.com/elastic/beats/v7/x-pack/filebeat/input/netflow/decoder"
 )
 
 func main() {
-	decoder, err := decoder.NewDecoder(decoder.NewConfig().
-		WithLogOutput(os.Stderr).
+	logger := logp.L().Named("netflow")
+
+	decoder, err := decoder.NewDecoder(decoder.NewConfig(logger).
 		WithProtocols("v1", "v5", "v9", "ipfix"))
 	if err != nil {
-		log.Fatal("Failed creating decoder:", err)
+		logger.Fatal("Failed creating decoder:", err)
 	}
 
 	addr, err := net.ResolveUDPAddr("udp", ":2055")
 	if err != nil {
-		log.Fatal("Failed to resolve address:", err)
+		logger.Fatal("Failed to resolve address:", err)
 	}
 
 	server, err := net.ListenUDP("udp", addr)
 	if err != nil {
-		log.Fatalf("Failed to listen on %v: %v", addr, err)
+		logger.Fatalf("Failed to listen on %v: %v", addr, err)
 	}
 	defer server.Close()
 
 	if err = server.SetReadBuffer(1 << 16); err != nil {
-		log.Fatalf("Failed to set read buffer size for socket: %v", err)
+		logger.Fatalf("Failed to set read buffer size for socket: %v", err)
 	}
 
-	log.Println("Listening on ", server.LocalAddr())
+	logger.Debug("Listening on ", server.LocalAddr())
 	buf := make([]byte, 8192)
 	decBuf := new(bytes.Buffer)
 	for {
 		size, remote, err := server.ReadFromUDP(buf)
 		if err != nil {
-			log.Println("Error reading from socket:", err)
+			logger.Debug("Error reading from socket:", err)
 			continue
 		}
 
 		decBuf.Reset()
 		decBuf.Write(buf[:size])
 		records, err := decoder.Read(decBuf, remote)
 		if err != nil {
-			log.Printf("warn: Failed reading records from %v: %v\n", remote, err)
+			logger.Debugf("warn: Failed reading records from %v: %v\n", remote, err)
 		}
 
 		for _, r := range records {
@@ -63,7 +64,7 @@ func main() {
 				"data":       r.Fields,
 			})
 			if err != nil {
-				log.Fatal(err)
+				logger.Fatal(err)
 			}
 			fmt.Println(string(evt))
 		}

x-pack/filebeat/input/netflow/decoder/ipfix/ipfix.go

@@ -5,8 +5,6 @@
 package ipfix
 
 import (
-	"log"
-
 	"github.com/elastic/beats/v7/x-pack/filebeat/input/netflow/decoder/config"
 	"github.com/elastic/beats/v7/x-pack/filebeat/input/netflow/decoder/protocol"
 	v9 "github.com/elastic/beats/v7/x-pack/filebeat/input/netflow/decoder/v9"
@@ -29,7 +27,7 @@ func init() {
 }
 
 func New(config config.Config) protocol.Protocol {
-	logger := log.New(config.LogOutput(), LogPrefix, 0)
+	logger := config.LogOutput().Named(LogPrefix)
 	decoder := DecoderIPFIX{
 		DecoderV9: v9.DecoderV9{Logger: logger, Fields: config.Fields()},
 	}

x-pack/filebeat/input/netflow/decoder/ipfix/ipfix_test.go

@@ -12,13 +12,19 @@ import (
 
 	"github.com/stretchr/testify/assert"
 
+	"github.com/elastic/elastic-agent-libs/logp"
+
 	"github.com/elastic/beats/v7/x-pack/filebeat/input/netflow/decoder/config"
 	"github.com/elastic/beats/v7/x-pack/filebeat/input/netflow/decoder/fields"
 	"github.com/elastic/beats/v7/x-pack/filebeat/input/netflow/decoder/record"
 	"github.com/elastic/beats/v7/x-pack/filebeat/input/netflow/decoder/test"
 	v9 "github.com/elastic/beats/v7/x-pack/filebeat/input/netflow/decoder/v9"
 )
 
+func init() {
+	logp.TestingSetup()
+}
+
 func TestMessageWithOptions(t *testing.T) {
 	rawString := "" +
 		"000a01e45bf435e1000000a500000000000200480400001000080004000c0004" +
@@ -67,7 +73,7 @@ func TestMessageWithOptions(t *testing.T) {
 			"version":      uint64(10),
 		},
 	}
-	proto := New(config.Defaults())
+	proto := New(config.Defaults(logp.L()))
 	flows, err := proto.OnPacket(bytes.NewBuffer(raw), test.MakeAddress(t, "127.0.0.1:1234"))
 	assert.NoError(t, err)
 	if assert.Len(t, flows, 7) {
@@ -84,7 +90,7 @@ func TestOptionTemplates(t *testing.T) {
 	key := v9.MakeSessionKey(addr, 1234, false)
 
 	t.Run("Single options template", func(t *testing.T) {
-		proto := New(config.Defaults())
+		proto := New(config.Defaults(logp.L()))
 		flows, err := proto.OnPacket(test.MakePacket([]uint16{
 			// Header
 			// Version, Length, Ts, SeqNo, Source
@@ -113,7 +119,7 @@ func TestOptionTemplates(t *testing.T) {
 	})
 
 	t.Run("Multiple options template", func(t *testing.T) {
-		proto := New(config.Defaults())
+		proto := New(config.Defaults(logp.L()))
 		raw := test.MakePacket([]uint16{
 			// Header
 			// Version, Count, Ts, SeqNo, Source
@@ -151,7 +157,7 @@ func TestOptionTemplates(t *testing.T) {
 	})
 
 	t.Run("records discarded", func(t *testing.T) {
-		proto := New(config.Defaults())
+		proto := New(config.Defaults(logp.L()))
 		raw := test.MakePacket([]uint16{
 			// Header
 			// Version, Count, Ts, SeqNo, Source
@@ -193,7 +199,7 @@ func TestOptionTemplates(t *testing.T) {
 func TestCustomFields(t *testing.T) {
 	addr := test.MakeAddress(t, "127.0.0.1:12345")
 
-	conf := config.Defaults()
+	conf := config.Defaults(logp.L())
 	conf.WithCustomFields(fields.FieldDict{
 		fields.Key{EnterpriseID: 0x12345678, FieldID: 33}: &fields.Field{Name: "customField", Decoder: fields.String},
 	})

x-pack/filebeat/input/netflow/decoder/protocol/registry_test.go

@@ -11,10 +11,16 @@ import (
 
 	"github.com/stretchr/testify/assert"
 
+	"github.com/elastic/elastic-agent-libs/logp"
+
 	"github.com/elastic/beats/v7/x-pack/filebeat/input/netflow/decoder/config"
 	"github.com/elastic/beats/v7/x-pack/filebeat/input/netflow/decoder/record"
 )
 
+func init() {
+	logp.TestingSetup()
+}
+
 type testProto int
 
 func (testProto) Version() uint16 {
@@ -61,7 +67,7 @@ func TestRegistry_Get(t *testing.T) {
 		assert.NoError(t, err)
 		gen, err := registry.Get("my_proto")
 		assert.NoError(t, err)
-		assert.Equal(t, testProto(0), gen(config.Defaults()))
+		assert.Equal(t, testProto(0), gen(config.Defaults(logp.L())))
 	})
 	t.Run("two protocols", func(t *testing.T) {
 		registry := ProtocolRegistry{}
@@ -71,10 +77,10 @@ func TestRegistry_Get(t *testing.T) {
 		assert.NoError(t, err)
 		gen, err := registry.Get("my_proto")
 		assert.NoError(t, err)
-		assert.Equal(t, testProto(1), gen(config.Defaults()))
+		assert.Equal(t, testProto(1), gen(config.Defaults(logp.L())))
 		gen, err = registry.Get("other_proto")
 		assert.NoError(t, err)
-		assert.Equal(t, testProto(2), gen(config.Defaults()))
+		assert.Equal(t, testProto(2), gen(config.Defaults(logp.L())))
 	})
 	t.Run("not registered", func(t *testing.T) {
 		registry := ProtocolRegistry{}

x-pack/filebeat/input/netflow/decoder/test/helper.go

@@ -20,11 +20,6 @@ type TestLogWriter struct {
 	testing.TB
 }
 
-func (t TestLogWriter) Write(buf []byte) (int, error) {
-	t.Log(string(buf))
-	return len(buf), nil
-}
-
 func MakeAddress(t testing.TB, ipPortPair string) net.Addr {
 	ip, portS, err := net.SplitHostPort(ipPortPair)
 	if err != nil {

x-pack/filebeat/input/netflow/decoder/v1/v1.go

@@ -9,7 +9,6 @@ import (
 	"encoding/binary"
 	"fmt"
 	"io"
-	"log"
 	"net"
 	"time"
 
@@ -18,6 +17,7 @@ import (
 	"github.com/elastic/beats/v7/x-pack/filebeat/input/netflow/decoder/protocol"
 	"github.com/elastic/beats/v7/x-pack/filebeat/input/netflow/decoder/record"
 	"github.com/elastic/beats/v7/x-pack/filebeat/input/netflow/decoder/template"
+	"github.com/elastic/elastic-agent-libs/logp"
 )
 
 const (
@@ -52,21 +52,23 @@ var templateV1 = template.Template{
 type ReadHeaderFn func(*bytes.Buffer, net.Addr) (int, time.Time, record.Map, error)
 
 type NetflowProtocol struct {
-	logger       *log.Logger
+	logger       *logp.Logger
 	flowTemplate *template.Template
 	version      uint16
 	readHeader   ReadHeaderFn
 }
 
 func init() {
-	protocol.Registry.Register(ProtocolName, New)
+	if err := protocol.Registry.Register(ProtocolName, New); err != nil {
+		panic(err)
+	}
 }
 
 func New(config config.Config) protocol.Protocol {
-	return NewProtocol(ProtocolID, &templateV1, readV1Header, log.New(config.LogOutput(), LogPrefix, 0))
+	return NewProtocol(ProtocolID, &templateV1, readV1Header, config.LogOutput().Named(LogPrefix))
 }
 
-func NewProtocol(version uint16, template *template.Template, readHeader ReadHeaderFn, logger *log.Logger) protocol.Protocol {
+func NewProtocol(version uint16, template *template.Template, readHeader ReadHeaderFn, logger *logp.Logger) protocol.Protocol {
 	return &NetflowProtocol{
 		logger:       logger,
 		flowTemplate: template,
@@ -90,7 +92,7 @@ func (NetflowProtocol) Stop() error {
 func (p *NetflowProtocol) OnPacket(buf *bytes.Buffer, source net.Addr) (flows []record.Record, err error) {
 	numFlows, timestamp, metadata, err := p.readHeader(buf, source)
 	if err != nil {
-		p.logger.Printf("Failed parsing packet: %v", err)
+		p.logger.Debugf("Failed parsing packet: %v", err)
 		return nil, fmt.Errorf("error reading netflow header: %w", err)
 	}
 	flows, err = p.flowTemplate.Apply(buf, numFlows)

x-pack/filebeat/input/netflow/decoder/v1/v1_test.go

@@ -17,18 +17,23 @@ import (
 	"github.com/elastic/beats/v7/x-pack/filebeat/input/netflow/decoder/record"
 	template2 "github.com/elastic/beats/v7/x-pack/filebeat/input/netflow/decoder/template"
 	"github.com/elastic/beats/v7/x-pack/filebeat/input/netflow/decoder/test"
+	"github.com/elastic/elastic-agent-libs/logp"
 )
 
+func init() {
+	logp.TestingSetup()
+}
+
 func TestNetflowProtocol_New(t *testing.T) {
-	proto := New(config.Defaults())
+	proto := New(config.Defaults(logp.L()))
 
 	assert.Nil(t, proto.Start())
 	assert.Equal(t, uint16(1), proto.Version())
 	assert.Nil(t, proto.Stop())
 }
 
 func TestNetflowProtocol_OnPacket(t *testing.T) {
-	proto := New(config.Defaults())
+	proto := New(config.Defaults(logp.L()))
 
 	rawS := "00010002000000015bf689f605946fb0" +
 		"acd910e5c0a8017b00000000000000000000000e00002cfa" +
@@ -105,7 +110,7 @@ func TestNetflowProtocol_OnPacket(t *testing.T) {
 }
 
 func TestNetflowProtocol_BadPacket(t *testing.T) {
-	proto := New(config.Defaults())
+	proto := New(config.Defaults(logp.L()))
 
 	rawS := "00010002000000015bf689f605"
 	raw, err := hex.DecodeString(rawS)