feat(netflow source): Add NetFlow source implementation

.github/actions/spelling/allow.txt

@@ -554,3 +554,5 @@ zst
 zstandard
 ZTE
 Zync
+IPFIX
+netflow

Cargo.toml

@@ -614,6 +614,7 @@ sources-logs = [
   "sources-file_descriptor",
   "sources-redis",
   "sources-socket",
+  "sources-netflow",
   "sources-splunk_hec",
   "sources-stdin",
   "sources-syslog",
@@ -688,6 +689,7 @@ sources-prometheus-pushgateway = ["sinks-prometheus", "sources-utils-http", "vec
 sources-pulsar = ["dep:apache-avro", "dep:pulsar"]
 sources-redis = ["dep:redis"]
 sources-socket = ["sources-utils-net", "tokio-util/net"]
+sources-netflow = ["sources-utils-net-udp", "tokio-util/net", "dep:base64"]
 sources-splunk_hec = ["dep:roaring"]
 sources-statsd = ["sources-utils-net", "tokio-util/net"]
 sources-stdin = ["tokio-util/io"]

changelog.d/netflow_source.feature.md

@@ -0,0 +1,3 @@
+Add NetFlow source implementation supporting NetFlow v5, NetFlow v9, IPFIX, and sFlow protocols. The implementation includes sophisticated template management, enterprise field support, template buffering for missing templates, and comprehensive error handling. Supports all major flow protocols with configurable enterprise field parsing and template caching.
+
+authors: modev2301

config/examples/netflow.yaml

@@ -0,0 +1,67 @@
+# NetFlow Source Example
+# ------------------------------------------------------------------------------
+# This example demonstrates how to collect NetFlow, IPFIX, and sFlow data
+# from network devices and export them to various destinations.
+#
+# Docs: https://vector.dev/docs/reference/configuration/sources/netflow
+
+data_dir: "/var/lib/vector"
+
+# Collect NetFlow data from network devices
+# Example: NetFlow v5, v9, IPFIX, and sFlow packets from routers/switches
+# Docs: https://vector.dev/docs/reference/configuration/sources/netflow
+sources:
+  netflow_data:
+    type: "netflow"
+    address: "0.0.0.0:2055"
+    protocols: ["netflow_v5", "netflow_v9", "ipfix", "sflow"]
+    max_packet_size: 65535
+    max_templates: 1000
+    template_timeout: 1800
+    parse_enterprise_fields: true
+    parse_options_templates: true
+    parse_variable_length_fields: true
+    buffer_missing_templates: true
+    max_buffered_records: 1000
+
+# Parse and enrich the flow data
+# Docs: https://vector.dev/docs/reference/configuration/transforms/remap
+transforms:
+  flow_parser:
+    inputs: ["netflow_data"]
+    type: "remap"
+    source: |
+      # Add timestamp if missing
+      if !exists(.timestamp) {
+        .timestamp = now()
+      }
+
+      # Add source information
+      .source = "netflow"
+
+      # Parse IP addresses for better visualization
+      if exists(.src_addr) {
+        .src_ip = parse_ip!(.src_addr)
+      }
+      if exists(.dst_addr) {
+        .dst_ip = parse_ip!(.dst_addr)
+      }
+
+# Send to Elasticsearch for analysis
+# Docs: https://vector.dev/docs/reference/configuration/sinks/elasticsearch
+sinks:
+  elasticsearch:
+    inputs: ["flow_parser"]
+    type: "elasticsearch"
+    endpoint: "http://localhost:9200"
+    index: "netflow-%Y-%m-%d"
+    encoding:
+      codec: "json"
+
+  # Also send to console for debugging
+  # Docs: https://vector.dev/docs/reference/configuration/sinks/console
+  console:
+    inputs: ["flow_parser"]
+    type: "console"
+    encoding:
+      codec: "json"

src/sources/mod.rs

@@ -82,6 +82,8 @@ pub mod pulsar;
 pub mod redis;
 #[cfg(feature = "sources-socket")]
 pub mod socket;
+#[cfg(feature = "sources-netflow")]
+pub mod netflow;
 #[cfg(feature = "sources-splunk_hec")]
 pub mod splunk_hec;
 #[cfg(feature = "sources-static_metrics")]