[receiver/netflow] Add the netflow receiver - PR 1

.github/ISSUE_TEMPLATE/bug_report.yaml

@@ -246,6 +246,7 @@ body:
       - receiver/mongodbatlas
       - receiver/mysql
       - receiver/namedpipe
+      - receiver/netflow
       - receiver/nginx
       - receiver/nsxt
       - receiver/opencensus

.github/ISSUE_TEMPLATE/feature_request.yaml

@@ -240,6 +240,7 @@ body:
       - receiver/mongodbatlas
       - receiver/mysql
       - receiver/namedpipe
+      - receiver/netflow
       - receiver/nginx
       - receiver/nsxt
       - receiver/opencensus

.github/ISSUE_TEMPLATE/other.yaml

@@ -240,6 +240,7 @@ body:
       - receiver/mongodbatlas
       - receiver/mysql
       - receiver/namedpipe
+      - receiver/netflow
       - receiver/nginx
       - receiver/nsxt
       - receiver/opencensus

.github/ISSUE_TEMPLATE/unmaintained.yaml

@@ -245,6 +245,7 @@ body:
       - receiver/mongodbatlas
       - receiver/mysql
       - receiver/namedpipe
+      - receiver/netflow
       - receiver/nginx
       - receiver/nsxt
       - receiver/opencensus

receiver/netflowreceiver/Makefile

@@ -0,0 +1 @@
+include ../../Makefile.Common

receiver/netflowreceiver/README.md

@@ -0,0 +1,96 @@
+# Netflow receiver
+<!-- status autogenerated section -->
+| Status        |           |
+| ------------- |-----------|
+| Stability     | [development]: logs   |
+| Distributions | [contrib] |
+| Issues        | [![Open issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector-contrib?query=is%3Aissue%20is%3Aopen%20label%3Areceiver%2Fnetflow%20&label=open&color=orange&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues?q=is%3Aopen+is%3Aissue+label%3Areceiver%2Fnetflow) [![Closed issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector-contrib?query=is%3Aissue%20is%3Aclosed%20label%3Areceiver%2Fnetflow%20&label=closed&color=blue&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues?q=is%3Aclosed+is%3Aissue+label%3Areceiver%2Fnetflow) |
+| [Code Owners](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/CONTRIBUTING.md#becoming-a-code-owner)    | [@evan-bradley](https://www.github.com/evan-bradley), [@dlopes7](https://www.github.com/dlopes7) |
+
+[development]: https://github.com/open-telemetry/opentelemetry-collector#development
+[contrib]: https://github.com/open-telemetry/opentelemetry-collector-releases/tree/main/distributions/otelcol-contrib
+<!-- end autogenerated section -->
+
+The netflow receiver can listen for [netflow](https://en.wikipedia.org/wiki/NetFlow), [sflow](https://en.wikipedia.org/wiki/SFlow), and [ipfix](https://en.wikipedia.org/wiki/IP_Flow_Information_Export) data and convert it to OpenTelemetry logs. The receiver is based on the [goflow2](https://github.com/netsampler/goflow2) project.
+
+This gives Opentelemetry users the capability of monitoring network traffic, and answer questions like:
+
+* Which protocols are passing through the network?
+* Which servers and clients are producing the highest amount of traffic?
+* What ports are involved in these network calls?
+* How many bytes and packets are being sent and received?
+
+
+## Getting started
+
+By default the receiver will listen for ipfix and netflow on port `2055` and sflow on port `6343`. The receiver can be configured to listen on different ports and protocols.
+
+Example configuration:
+
+```yaml
+receivers:
+  netflow:
+    listeners:
+      - scheme: netflow
+        port: 2055
+        sockets: 16
+        workers: 32
+
+processors:
+  batch:
+    send_batch_size: 2000
+    timeout: 30s
+
+exporters:
+  debug:
+    verbosity: detailed
+
+service:
+  pipelines:
+    logs:
+      receivers: [netflow]
+      processors: [batch]
+      exporters: [debug]
+  telemetry:
+    logs:
+      level: debug
+```
+
+We recommend using the batch processor to reduce the number of log requests being sent to the exporter. The batch processor will batch log records together and send them in a single request to the exporter.
+
+You would then configure your network devices to send netflow, sflow, or ipfix data to the collector on the specified ports.
+
+## Configuration
+
+| Field | Description | Examples |
+|-------|-------------|--------|
+| scheme | The type of flow data that the listener will receive | `sflow`, `netflow`, `flow` |
+| hostname | The hostname or IP address that the listener will bind to | `localhost` |
+| port | The port that the listener will bind to | `2055` |
+| sockets | The number of sockets that the listener will use | 1 |
+| workers | The number of workers that the listener will use to decode incoming flow messages | 2 |
+| queue_size | The size of the queue that the listener will use | 1000 |
+
+## Data format
+
+The netflow data is standardized for the different schemas and is converted to OpenTelemetry logs. The output will look like this:
+
+```json
+{
+    "type": "SFLOW_5",
+    "time_received_ns": 1681583295157626000,
+    "sequence_num": 2999,
+    "sampling_rate": 100,
+    "sampler_address": "192.168.0.1",
+    "time_flow_start_ns": 1681583295157626000,
+    "time_flow_end_ns": 1681583295157626000,
+    "bytes": 1500,
+    "packets": 1,
+    "src_addr": "fd01::1",
+    "dst_addr": "fd01::2",
+    "etype": "IPv6",
+    "proto": "TCP",
+    "src_port": 443,
+    "dst_port": 50001
+}
+```

receiver/netflowreceiver/config.go

@@ -0,0 +1,73 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package netflowreceiver // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/netflowreceiver"
+
+import "fmt"
+
+// Config represents the receiver config settings within the collector's config.yaml
+type Config struct {
+	Listeners []ListenerConfig `mapstructure:"listeners"`
+}
+
+type ListenerConfig struct {
+
+	// The scheme defines the type of flow data that the listener will receive
+	// The scheme must be one of sflow, netflow, or flow
+	Scheme string `mapstructure:"scheme"`
+
+	// The hostname or IP address that the listener will bind to
+	Hostname string `mapstructure:"hostname"`
+
+	// The port that the listener will bind to
+	Port int `mapstructure:"port"`
+
+	// The number of sockets that the listener will use
+	Sockets int `mapstructure:"sockets"`
+
+	// The number of workers that the listener will use to decode incoming flow messages
+	// By default it will be two times the number of sockets
+	// Ideally set this to the number of CPU cores
+	Workers int `mapstructure:"workers"`
+
+	// The size of the queue that the listener will use
+	// This is a buffer that will hold flow messages before they are processed by a worker
+	QueueSize int `mapstructure:"queue_size"`
+}
+
+// Validate checks if the receiver configuration is valid
+func (cfg *Config) Validate() error {
+	validSchemes := [3]string{"sflow", "netflow", "flow"}
+
+	for _, listener := range cfg.Listeners {
+
+		validScheme := false
+		for _, scheme := range validSchemes {
+			if listener.Scheme == scheme {
+				validScheme = true
+				break
+			}
+		}
+		if !validScheme {
+			return fmt.Errorf("scheme must be one of sflow, netflow, or flow")
+		}
+
+		if listener.Sockets <= 0 {
+			return fmt.Errorf("sockets must be greater than 0")
+		}
+
+		if listener.Workers <= 0 {
+			return fmt.Errorf("workers must be greater than 0")
+		}
+
+		if listener.QueueSize <= 0 {
+			listener.QueueSize = defaultQueueSize
+		}
+
+		if listener.Port <= 0 {
+			return fmt.Errorf("port must be greater than 0")
+		}
+	}
+
+	return nil
+}

receiver/netflowreceiver/config_test.go

@@ -0,0 +1,118 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package netflowreceiver
+
+import (
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/confmap/confmaptest"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/netflowreceiver/internal/metadata"
+)
+
+func TestLoadConfig(t *testing.T) {
+	t.Parallel()
+
+	cm, err := confmaptest.LoadConf(filepath.Join("testdata", "config.yaml"))
+	require.NoError(t, err)
+
+	tests := []struct {
+		id       component.ID
+		expected component.Config
+	}{
+		{
+			id:       component.NewIDWithName(metadata.Type, "defaults"),
+			expected: createDefaultConfig(),
+		},
+		{
+			id: component.NewIDWithName(metadata.Type, "two_listeners"),
+			expected: &Config{
+				Listeners: []ListenerConfig{
+					{
+						Scheme:    "netflow",
+						Port:      2055,
+						Sockets:   1,
+						Workers:   4,
+						QueueSize: 0,
+					},
+					{
+						Scheme:    "sflow",
+						Port:      6443,
+						Sockets:   1,
+						Workers:   2,
+						QueueSize: 1000,
+					},
+				},
+			},
+		},
+		{
+			id: component.NewIDWithName(metadata.Type, "one_listener"),
+			expected: &Config{
+				Listeners: []ListenerConfig{
+					{
+						Scheme:    "netflow",
+						Port:      2055,
+						Sockets:   1,
+						Workers:   1,
+						QueueSize: 0,
+					},
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.id.String(), func(t *testing.T) {
+			factory := NewFactory()
+			cfg := factory.CreateDefaultConfig()
+
+			sub, err := cm.Sub(tt.id.String())
+			require.NoError(t, err)
+			require.NoError(t, sub.Unmarshal(cfg))
+
+			assert.NoError(t, component.ValidateConfig(cfg))
+			assert.Equal(t, tt.expected, cfg)
+		})
+	}
+}
+
+func TestInvalidConfig(t *testing.T) {
+	t.Parallel()
+
+	cm, err := confmaptest.LoadConf(filepath.Join("testdata", "config.yaml"))
+	require.NoError(t, err)
+
+	tests := []struct {
+		id  component.ID
+		err string
+	}{
+		{
+			id:  component.NewIDWithName(metadata.Type, "invalid_schema"),
+			err: "scheme must be one of sflow, netflow, or flow",
+		},
+		{
+			id:  component.NewIDWithName(metadata.Type, "invalid_port"),
+			err: "port must be greater than 0",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.id.String(), func(t *testing.T) {
+			factory := NewFactory()
+			cfg := factory.CreateDefaultConfig()
+
+			sub, err := cm.Sub(tt.id.String())
+			require.NoError(t, err)
+			require.NoError(t, sub.Unmarshal(cfg))
+
+			err = component.ValidateConfig(cfg)
+			assert.ErrorContains(t, err, tt.err)
+		})
+	}
+
+}

receiver/netflowreceiver/doc.go

@@ -0,0 +1,6 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+//go:generate mdatagen metadata.yaml
+
+package netflowreceiver // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/netflowreceiver"

receiver/netflowreceiver/factory.go

@@ -0,0 +1,64 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package netflowreceiver // import "github.com/open-telemetry/opentelemetry-collector-contrib/receiver/netflowreceiver"
+
+import (
+	"context"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/receiver"
+)
+
+var (
+	typeStr = component.MustNewType("netflow")
+)
+
+const (
+	defaultSockets   = 1
+	defaultWorkers   = 2
+	defaultQueueSize = 1_000_000
+)
+
+// NewFactory creates a factory for netflow receiver.
+func NewFactory() receiver.Factory {
+	return receiver.NewFactory(
+		typeStr,
+		createDefaultConfig,
+		receiver.WithLogs(createLogsReceiver, component.StabilityLevelAlpha))
+}
+
+func createDefaultConfig() component.Config {
+	return &Config{
+		Listeners: []ListenerConfig{
+			{
+				Scheme:    "sflow",
+				Port:      6343,
+				Sockets:   defaultSockets,
+				Workers:   defaultWorkers,
+				QueueSize: defaultQueueSize,
+			},
+			{
+				Scheme:    "netflow",
+				Port:      2055,
+				Sockets:   defaultSockets,
+				Workers:   defaultWorkers,
+				QueueSize: defaultQueueSize,
+			},
+		},
+	}
+}
+
+func createLogsReceiver(_ context.Context, params receiver.CreateSettings, cfg component.Config, consumer consumer.Logs) (receiver.Logs, error) {
+	logger := params.Logger
+	conf := cfg.(*Config)
+
+	nr := &netflowReceiver{
+		logger:      logger,
+		logConsumer: consumer,
+		config:      conf,
+	}
+
+	return nr, nil
+}