[receiver/netflowreceiver] add send_raw option for sending unparsed logs (open-telemetry#38832)

#### Description
- Add a `send_raw` option to the config. 
- When set to true, the NetFlow log messages will be sent as an unparsed
string in the log body.
- When send_raw is false or is not configured, logs are parsed into
attributes as before.
- This works for all kinds of NetFlow log data, such as v5, v9, etc.

#### Link to tracking issue
Closes:
open-telemetry#38920

#### Testing
- New test in producer_test.go
- Update test in config_test.go
- Testing sending logs to Google SecOps:
- Simulate NetFlow logs:
https://hub.docker.com/r/networkstatic/nflow-generator
    - Set up configuration.yaml sending Netflow to Google SecOps
    - Confirm output of unparsed logs in Google SecOps

#### Documentation
- Update README.md with documentation of the `send_raw` field

Author: justinianvoss22

.chloggen/netflowreceiver-add-send-raw-option.yaml

@@ -0,0 +1,25 @@
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# 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: Add `send_raw` option to send logs as a raw string in the log body instead of parsed into attributes.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [38920]
+
+# (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:
+
+# 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/README.md

@@ -38,6 +38,12 @@ receivers:
       port: 6343
       sockets: 16
       workers: 32
+  netflow/raw:
+    - scheme: netflow
+      port: 2055
+      sockets: 16
+      workers: 32
+      send_raw: true
 
 processors:
   batch:
@@ -73,6 +79,12 @@ You would then configure your network devices to send netflow, sflow, or ipfix d
 | sockets | The number of sockets to use | 1 | 1 |
 | workers | The number of workers used to decode incoming flow messages | 2 | 2 |
 | queue_size | The size of the incoming netflow packets queue, it will always be at least 1000. | 5000 | 1000 |
+| send_raw   | Whether to send raw flow messages instead of parsing them                        | `true`, `false`    | `false`   |
+
+When `send_raw` is set to `true`, the receiver will:
+
+- Skip parsing the netflow/sflow messages
+- Send the raw message as the log body
 
 ## Data format
 

receiver/netflowreceiver/config.go

@@ -30,6 +30,9 @@ type Config struct {
 	// 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"`
+
+	// SendRaw determines whether to send raw flow messages instead of parsing them
+	SendRaw bool `mapstructure:"send_raw"`
 }
 
 // Validate checks if the receiver configuration is valid

receiver/netflowreceiver/config_test.go

@@ -60,6 +60,17 @@ func TestLoadConfig(t *testing.T) {
 				QueueSize: 1000,
 			},
 		},
+		{
+			id: component.NewIDWithName(metadata.Type, "raw_logs"),
+			expected: &Config{
+				Scheme:    "netflow",
+				Port:      2055,
+				Sockets:   1,
+				Workers:   1,
+				QueueSize: 1000,
+				SendRaw:   true,
+			},
+		},
 	}
 
 	for _, tt := range tests {

receiver/netflowreceiver/producer.go

@@ -5,6 +5,7 @@ package netflowreceiver // import "github.com/open-telemetry/opentelemetry-colle
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/netsampler/goflow2/v2/producer"
 	"go.opentelemetry.io/collector/consumer"
@@ -19,6 +20,7 @@ type OtelLogsProducerWrapper struct {
 	wrapped     producer.ProducerInterface
 	logConsumer consumer.Logs
 	logger      *zap.Logger
+	sendRaw     bool
 }
 
 // Produce converts the message into a list log records and sends them to log consumer
@@ -47,9 +49,14 @@ func (o *OtelLogsProducerWrapper) Produce(msg any, args *producer.ProduceArgs) (
 	// A single netflow packet can contain multiple flow messages
 	for _, msg := range flowMessageSet {
 		logRecord := logRecords.AppendEmpty()
-		parseErr := addMessageAttributes(msg, &logRecord)
-		if parseErr != nil {
-			continue
+		if o.sendRaw {
+			logRecord.Body().SetStr(fmt.Sprintf("%+v", msg))
+		} else {
+			// Parse the message and add the attributes to the log record
+			err = addMessageAttributes(msg, &logRecord)
+			if err != nil {
+				o.logger.Error("error adding message attributes", zap.Error(err))
+			}
 		}
 	}
 
@@ -73,10 +80,11 @@ func (o *OtelLogsProducerWrapper) Commit(flowMessageSet []producer.ProducerMessa
 	o.wrapped.Commit(flowMessageSet)
 }
 
-func newOtelLogsProducer(wrapped producer.ProducerInterface, logConsumer consumer.Logs, logger *zap.Logger) producer.ProducerInterface {
+func newOtelLogsProducer(wrapped producer.ProducerInterface, logConsumer consumer.Logs, logger *zap.Logger, sendRaw bool) producer.ProducerInterface {
 	return &OtelLogsProducerWrapper{
 		wrapped:     wrapped,
 		logConsumer: logConsumer,
 		logger:      logger,
+		sendRaw:     sendRaw,
 	}
 }

receiver/netflowreceiver/producer_test.go

@@ -4,6 +4,7 @@
 package netflowreceiver
 
 import (
+	"fmt"
 	"net/netip"
 	"testing"
 
@@ -56,7 +57,7 @@ func TestProduce(t *testing.T) {
 	protoProducer, err := protoproducer.CreateProtoProducer(cfgm, protoproducer.CreateSamplingSystem)
 	require.NoError(t, err)
 
-	otelLogsProducer := newOtelLogsProducer(protoProducer, consumertest.NewNop(), zap.NewNop())
+	otelLogsProducer := newOtelLogsProducer(protoProducer, consumertest.NewNop(), zap.NewNop(), false)
 	messages, err := otelLogsProducer.Produce(message, &producer.ProduceArgs{})
 	require.NoError(t, err)
 	require.NotNil(t, messages)
@@ -70,6 +71,85 @@ func TestProduce(t *testing.T) {
 	assert.Equal(t, uint32(838987416), pm.SequenceNum)
 }
 
+func TestProduceRaw(t *testing.T) {
+	// list of netflow.DataFlowSet
+	message := &netflow.NFv9Packet{
+		Version:        9,
+		Count:          1,
+		SystemUptime:   0xb3bff683,
+		UnixSeconds:    0x618aa3a8,
+		SequenceNumber: 838987416,
+		SourceId:       256,
+		FlowSets: []any{
+			netflow.DataFlowSet{
+				FlowSetHeader: netflow.FlowSetHeader{
+					Id:     260,
+					Length: 1372,
+				},
+				Records: []netflow.DataRecord{
+					{
+						Values: []netflow.DataField{
+							{
+								PenProvided: false,
+								Type:        2,
+								Pen:         0,
+								Value:       []uint8{0x00, 0x00, 0x00, 0x01},
+							},
+						},
+					},
+					{
+						Values: []netflow.DataField{
+							{
+								PenProvided: false,
+								Type:        2,
+								Pen:         0,
+								Value:       []uint8{0x00, 0x00, 0x00, 0x02},
+							},
+						},
+					},
+					{
+						Values: []netflow.DataField{
+							{
+								PenProvided: false,
+								Type:        2,
+								Pen:         0,
+								Value:       []uint8{0x00, 0x00, 0x00, 0x03},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	cfgProducer := &protoproducer.ProducerConfig{}
+	cfgm, err := cfgProducer.Compile()
+	require.NoError(t, err)
+
+	protoProducer, err := protoproducer.CreateProtoProducer(cfgm, protoproducer.CreateSamplingSystem)
+	require.NoError(t, err)
+
+	sink := &consumertest.LogsSink{}
+	otelLogsProducer := newOtelLogsProducer(protoProducer, sink, zap.NewNop(), true)
+
+	messages, err := otelLogsProducer.Produce(message, &producer.ProduceArgs{})
+	require.NoError(t, err)
+	require.NotNil(t, messages)
+	assert.Len(t, messages, 3)
+
+	logs := sink.AllLogs()
+	require.Len(t, logs, 1)
+	records := logs[0].ResourceLogs().At(0).ScopeLogs().At(0).LogRecords()
+	require.Equal(t, 3, records.Len()) // Should have one record per flow record
+
+	// Each record should be a raw string representation of the ProducerMessage
+	for i := 0; i < 3; i++ {
+		record := records.At(i)
+		msg := messages[i]
+		assert.Equal(t, fmt.Sprintf("%+v", msg), record.Body().Str())
+	}
+}
+
 // This PanicProducer replaces the ProtoProducer, to simulate it producing a panic
 type PanicProducer struct{}
 
@@ -90,7 +170,7 @@ func TestProducerPanic(t *testing.T) {
 	mockConsumer := consumertest.NewNop()
 
 	// Wrap a PanicProducer (instead of ProtoProducer) in the OtelLogsProducerWrapper
-	wrapper := newOtelLogsProducer(&PanicProducer{}, mockConsumer, logger)
+	wrapper := newOtelLogsProducer(&PanicProducer{}, mockConsumer, logger, false)
 
 	// Call Produce which should recover from panic
 	messages, err := wrapper.Produce(nil, &producer.ProduceArgs{

receiver/netflowreceiver/receiver.go

@@ -108,7 +108,7 @@ func (nr *netflowReceiver) buildDecodeFunc() (utils.DecoderFunc, error) {
 
 	// the otel log producer converts those messages into OpenTelemetry logs
 	// it is a wrapper around the protobuf producer
-	otelLogsProducer := newOtelLogsProducer(protoProducer, nr.logConsumer, nr.logger)
+	otelLogsProducer := newOtelLogsProducer(protoProducer, nr.logConsumer, nr.logger, nr.config.SendRaw)
 
 	cfgPipe := &utils.PipeConfig{
 		Producer: otelLogsProducer,

receiver/netflowreceiver/testdata/config.yaml

@@ -40,3 +40,11 @@ netflow/sflow:
   sockets: 1
   workers: 1
   queue_size: 0
+
+netflow/raw_logs:
+  scheme: netflow
+  port: 2055
+  sockets: 1
+  workers: 1
+  queue_size: 0
+  send_raw: true