[receiver/netflow] Netflow receiver implementation - PR 2

.chloggen/netflow-receiver-implementation.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: new_component
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: netflowreceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Adds the implementation of the netflow receiver
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [32732, 97279]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext: Implement the UDP listener and the producer, as well as adding tests.
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

receiver/netflowreceiver/config_test.go

@@ -36,7 +36,47 @@ func TestLoadConfig(t *testing.T) {
 				Port:      2055,
 				Sockets:   1,
 				Workers:   1,
-				QueueSize: 1000000,
+				QueueSize: 1000,
+			},
+		},
+		{
+			id: component.NewIDWithName(metadata.Type, "zero_queue"),
+			expected: &Config{
+				Scheme:    "netflow",
+				Port:      2055,
+				Sockets:   1,
+				Workers:   1,
+				QueueSize: 1000,
+			},
+		},
+		{
+			id: component.NewIDWithName(metadata.Type, "zero_queue"),
+			expected: &Config{
+				Scheme:    "netflow",
+				Port:      2055,
+				Sockets:   1,
+				Workers:   1,
+				QueueSize: 1000,
+			},
+		},
+		{
+			id: component.NewIDWithName(metadata.Type, "sflow"),
+			expected: &Config{
+				Scheme:    "sflow",
+				Port:      2055,
+				Sockets:   1,
+				Workers:   1,
+				QueueSize: 1000,
+			},
+		},
+		{
+			id: component.NewIDWithName(metadata.Type, "flow"),
+			expected: &Config{
+				Scheme:    "flow",
+				Port:      2055,
+				Sockets:   1,
+				Workers:   1,
+				QueueSize: 1000,
 			},
 		},
 	}
@@ -74,6 +114,14 @@ func TestInvalidConfig(t *testing.T) {
 			id:  component.NewIDWithName(metadata.Type, "invalid_port"),
 			err: "port must be greater than 0",
 		},
+		{
+			id:  component.NewIDWithName(metadata.Type, "zero_sockets"),
+			err: "sockets must be greater than 0",
+		},
+		{
+			id:  component.NewIDWithName(metadata.Type, "zero_workers"),
+			err: "workers must be greater than 0",
+		},
 	}
 
 	for _, tt := range tests {

receiver/netflowreceiver/factory.go

@@ -16,7 +16,7 @@ import (
 const (
 	defaultSockets   = 1
 	defaultWorkers   = 2
-	defaultQueueSize = 1_000_000
+	defaultQueueSize = 1_000
 )
 
 // NewFactory creates a factory for netflow receiver.
@@ -27,6 +27,8 @@ func NewFactory() receiver.Factory {
 		receiver.WithLogs(createLogsReceiver, metadata.LogsStability))
 }
 
+// Config defines configuration for netflow receiver.
+// By default we listen for netflow traffic on port 2055
 func createDefaultConfig() component.Config {
 	return &Config{
 		Scheme:    "netflow",
@@ -37,14 +39,15 @@ func createDefaultConfig() component.Config {
 	}
 }
 
+// createLogsReceiver creates a netflow receiver.
+// We also create the UDP receiver, which is the piece of software that actually listens
+// for incoming netflow traffic on an UDP port.
 func createLogsReceiver(_ context.Context, params receiver.Settings, cfg component.Config, consumer consumer.Logs) (receiver.Logs, error) {
-	logger := params.Logger
-	conf := cfg.(*Config)
+	conf := *(cfg.(*Config))
 
-	nr := &netflowReceiver{
-		logger:      logger,
-		logConsumer: consumer,
-		config:      conf,
+	nr, err := newNetflowLogsReceiver(params, conf, consumer)
+	if err != nil {
+		return nil, err
 	}
 
 	return nr, nil

receiver/netflowreceiver/go.mod

@@ -3,12 +3,14 @@ module github.com/open-telemetry/opentelemetry-collector-contrib/receiver/netflo
 go 1.22.0
 
 require (
+	github.com/netsampler/goflow2/v2 v2.2.1
 	github.com/stretchr/testify v1.10.0
 	go.opentelemetry.io/collector/component v0.116.1-0.20241220212031-7c2639723f67
 	go.opentelemetry.io/collector/component/componenttest v0.116.1-0.20241220212031-7c2639723f67
 	go.opentelemetry.io/collector/confmap v1.22.1-0.20241220212031-7c2639723f67
 	go.opentelemetry.io/collector/consumer v1.22.1-0.20241220212031-7c2639723f67
 	go.opentelemetry.io/collector/consumer/consumertest v0.116.1-0.20241220212031-7c2639723f67
+	go.opentelemetry.io/collector/pdata v1.22.1-0.20241220212031-7c2639723f67
 	go.opentelemetry.io/collector/receiver v0.116.1-0.20241220212031-7c2639723f67
 	go.opentelemetry.io/collector/receiver/receivertest v0.116.1-0.20241220212031-7c2639723f67
 	go.uber.org/goleak v1.3.0
@@ -26,6 +28,7 @@ require (
 	github.com/knadh/koanf/maps v0.1.1 // indirect
 	github.com/knadh/koanf/providers/confmap v0.1.0 // indirect
 	github.com/knadh/koanf/v2 v2.1.2 // indirect
+	github.com/libp2p/go-reuseport v0.4.0 // indirect
 	github.com/mitchellh/copystructure v1.2.0 // indirect
 	github.com/mitchellh/reflectwalk v1.0.2 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
@@ -34,7 +37,6 @@ require (
 	go.opentelemetry.io/collector/config/configtelemetry v0.116.1-0.20241220212031-7c2639723f67 // indirect
 	go.opentelemetry.io/collector/consumer/consumererror v0.116.1-0.20241220212031-7c2639723f67 // indirect
 	go.opentelemetry.io/collector/consumer/xconsumer v0.116.1-0.20241220212031-7c2639723f67 // indirect
-	go.opentelemetry.io/collector/pdata v1.22.1-0.20241220212031-7c2639723f67 // indirect
 	go.opentelemetry.io/collector/pdata/pprofile v0.116.1-0.20241220212031-7c2639723f67 // indirect
 	go.opentelemetry.io/collector/pipeline v0.116.1-0.20241220212031-7c2639723f67 // indirect
 	go.opentelemetry.io/collector/receiver/xreceiver v0.116.1-0.20241220212031-7c2639723f67 // indirect

receiver/netflowreceiver/parser.go

@@ -0,0 +1,139 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package netflowreceiver // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/netflowreceiver"
+
+import (
+	"errors"
+	"net/netip"
+	"time"
+
+	"github.com/netsampler/goflow2/v2/producer"
+	protoproducer "github.com/netsampler/goflow2/v2/producer/proto"
+)
+
+var (
+	etypeName = map[uint32]string{
+		0x806:  "ARP",
+		0x800:  "IPv4",
+		0x86dd: "IPv6",
+	}
+	protoName = map[uint32]string{
+		1:   "ICMP",
+		6:   "TCP",
+		17:  "UDP",
+		58:  "ICMPv6",
+		132: "SCTP",
+	}
+
+	flowTypeName = map[int32]string{
+		0: "UNKNOWN",
+		1: "SFLOW_5",
+		2: "NETFLOW_V5",
+		3: "NETFLOW_V9",
+		4: "IPFIX",
+	}
+)
+
+type NetworkAddress struct {
+	Address string `json:"address,omitempty"`
+	Port    uint32 `json:"port,omitempty"`
+}
+
+type Flow struct {
+	Type           string    `json:"type,omitempty"`
+	TimeReceived   time.Time `json:"time_received,omitempty"`
+	Start          time.Time `json:"start,omitempty"`
+	End            time.Time `json:"end,omitempty"`
+	SequenceNum    uint32    `json:"sequence_num,omitempty"`
+	SamplingRate   uint64    `json:"sampling_rate,omitempty"`
+	SamplerAddress string    `json:"sampler_address,omitempty"`
+}
+
+type Protocol struct {
+	Name []byte `json:"name,omitempty"` // Layer 7
+}
+
+type NetworkIO struct {
+	Bytes   uint64 `json:"bytes,omitempty"`
+	Packets uint64 `json:"packets,omitempty"`
+}
+
+type OtelNetworkMessage struct {
+	Source      NetworkAddress `json:"source,omitempty"`
+	Destination NetworkAddress `json:"destination,omitempty"`
+	Transport   string         `json:"transport,omitempty"` // Layer 4
+	Type        string         `json:"type,omitempty"`      // Layer 3
+	IO          NetworkIO      `json:"io,omitempty"`
+	Flow        Flow           `json:"flow,omitempty"`
+}
+
+func getEtypeName(etype uint32) string {
+	if name, ok := etypeName[etype]; ok {
+		return name
+	}
+	return "unknown"
+}
+
+func getProtoName(proto uint32) string {
+	if name, ok := protoName[proto]; ok {
+		return name
+	}
+	return "unknown"
+}
+
+func getFlowTypeName(flowType int32) string {
+	if name, ok := flowTypeName[flowType]; ok {
+		return name
+	}
+	return "unknown"
+}
+
+// convertToOtel converts a ProtoProducerMessage to an OtelNetworkMessage
+func convertToOtel(m producer.ProducerMessage) (*OtelNetworkMessage, error) {
+	// we know msg is ProtoProducerMessage because that is the parent producer
+	pm, ok := m.(*protoproducer.ProtoProducerMessage)
+	if !ok {
+		return nil, errors.New("message is not ProtoProducerMessage")
+	}
+
+	// Parse IP addresses bytes to netip.Addr
+	srcAddr, _ := netip.AddrFromSlice(pm.SrcAddr)
+	dstAddr, _ := netip.AddrFromSlice(pm.DstAddr)
+	samplerAddr, _ := netip.AddrFromSlice(pm.SamplerAddress)
+
+	// Time the receiver received the message
+	receivedTime := time.Unix(0, int64(pm.TimeReceivedNs))
+	startTime := time.Unix(0, int64(pm.TimeFlowStartNs))
+	endTime := time.Unix(0, int64(pm.TimeFlowEndNs))
+
+	// Construct the actual log record based on the otel semantic conventions
+	// see https://opentelemetry.io/docs/specs/semconv/general/attributes/
+	otelMessage := OtelNetworkMessage{
+		Source: NetworkAddress{
+			Address: srcAddr.String(),
+			Port:    pm.SrcPort,
+		},
+		Destination: NetworkAddress{
+			Address: dstAddr.String(),
+			Port:    pm.DstPort,
+		},
+		Type:      getEtypeName(pm.Etype), // Layer 3
+		Transport: getProtoName(pm.Proto), // Layer 4
+		IO: NetworkIO{
+			Bytes:   pm.Bytes,
+			Packets: pm.Packets,
+		},
+		Flow: Flow{
+			Type:           getFlowTypeName(int32(pm.Type)),
+			TimeReceived:   receivedTime,
+			Start:          startTime,
+			End:            endTime,
+			SequenceNum:    pm.SequenceNum,
+			SamplingRate:   pm.SamplingRate,
+			SamplerAddress: samplerAddr.String(),
+		},
+	}
+
+	return &otelMessage, nil
+}