Create module for NetFlow to add geoip (#10877)

This creates a module around the Filebeat netflow input in order to enrich
the events with source.geo.* and destination.geo.* fields by using the geoip
processor in Elasticsearch Ingest Node.

Usage:

```
filebeat.modules:
- module: netflow
  log:
    enabled: true
    var:
      netflow_host: 0.0.0.0
      netflow_port: 2055
```

CHANGELOG.next.asciidoc

@@ -243,6 +243,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Populate more ECS fields in the Suricata module. {pull}10006[10006]
 - Add ISO8601 timestamp support in syslog metricset. {issue}8716[8716] {pull}10736[10736]
 - Add more info to message logged when a duplicated symlink file is found {pull}10845[10845]
+- Add Netflow module to enrich flow events with geoip data. {pull}10877[10877]
 
 *Heartbeat*
 

filebeat/docs/fields.asciidoc

@@ -33,6 +33,7 @@ grouped in the following categories:
 * <<exported-fields-mysql>>
 * <<exported-fields-nats>>
 * <<exported-fields-netflow>>
+* <<exported-fields-netflow-module>>
 * <<exported-fields-nginx>>
 * <<exported-fields-osquery>>
 * <<exported-fields-postgresql>>
@@ -10621,6 +10622,12 @@ type: short
 
 --
 
+[[exported-fields-netflow-module]]
+== NetFlow fields
+
+Module for receiving NetFlow and IPFIX flow records over UDP. The module does not add fields beyond what the netflow input provides.
+
+
 [[exported-fields-nginx]]
 == Nginx fields
 

filebeat/docs/modules/netflow.asciidoc

@@ -0,0 +1,80 @@
+////
+This file is generated! See scripts/docs_collector.py
+////
+
+[[filebeat-module-netflow]]
+[role="xpack"]
+
+:modulename: netflow
+:has-dashboards: false
+
+== NetFlow module
+
+This is a module for receiving NetFlow and IPFIX flow records over UDP. This
+input supports NetFlow versions 1, 5, 6, 7, 8 and 9, as well as IPFIX. For
+NetFlow versions older than 9, fields are mapped automatically to NetFlow v9.
+
+This module wraps the <<filebeat-input-netflow,netflow input>> to enrich the
+flow records with geolocation information about the IP endpoints by using
+Elasticsearch Ingest Node.
+
+[float]
+=== Compatibility
+
+This module requires the {elasticsearch-plugins}/ingest-geoip.html[ingest-geoip]
+Elasticsearch plugins.
+
+include::../include/running-modules.asciidoc[]
+
+include::../include/configuring-intro.asciidoc[]
+
+:fileset_ex: log
+
+include::../include/config-option-intro.asciidoc[]
+
+[float]
+==== `log` fileset settings
+
+The fileset is by default configured to listen for UDP traffic on
+`localhost:2055`. For most uses cases you will want to set the `netflow_host`
+variable to allow the input bind to all interfaces so that it can receive
+traffic from network devices.
+
+["source","yaml",subs="attributes"]
+-----
+- module: netflow
+  log:
+    enabled: true
+    var:
+      netflow_host: 0.0.0.0
+      netflow_port: 2055
+-----
+
+`var.netflow_host`:: Address to find to. Defaults to `localhost`.
+
+`var.netflow_port`:: Port to listen on. Defaults to `2055`.
+
+`var.max_message_size`:: The maximum size of the message received over UDP.
+The default is `10KiB`.
+
+`var.expiration_timeout`:: The time before an idle session or unused template is
+expired. Only applicable to v9 and IPFIX protocols. A value of zero disables
+expiration.
+
+`var.queue_size`:: The maximum number of packets that can be queued for
+processing. Use this setting to avoid packet-loss when dealing with occasional
+bursts of traffic.
+
+:has-dashboards!:
+
+:fileset_ex!:
+
+:modulename!:
+
+
+[float]
+=== Fields
+
+For a description of each field in the module, see the
+<<exported-fields-netflow,exported fields>> section.
+

filebeat/docs/modules_list.asciidoc

@@ -16,6 +16,7 @@ This file is generated! See scripts/docs_collector.py
   * <<filebeat-module-mongodb>>
   * <<filebeat-module-mysql>>
   * <<filebeat-module-nats>>
+  * <<filebeat-module-netflow>>
   * <<filebeat-module-nginx>>
   * <<filebeat-module-osquery>>
   * <<filebeat-module-postgresql>>
@@ -43,6 +44,7 @@ include::modules/logstash.asciidoc[]
 include::modules/mongodb.asciidoc[]
 include::modules/mysql.asciidoc[]
 include::modules/nats.asciidoc[]
+include::modules/netflow.asciidoc[]
 include::modules/nginx.asciidoc[]
 include::modules/osquery.asciidoc[]
 include::modules/postgresql.asciidoc[]

x-pack/filebeat/filebeat.reference.yml

@@ -312,6 +312,14 @@ filebeat.modules:
     # Filebeat will choose the paths depending on your OS.
     #var.paths:
 
+#------------------------------- NetFlow Module -------------------------------
+- module: netflow
+  log:
+    enabled: true
+    var:
+      netflow_host: localhost
+      netflow_port: 2055
+
 #-------------------------------- Nginx Module --------------------------------
 #- module: nginx
   # Access logs

x-pack/filebeat/module/netflow/_meta/config.yml

@@ -0,0 +1,6 @@
+- module: netflow
+  log:
+    enabled: true
+    var:
+      netflow_host: localhost
+      netflow_port: 2055

x-pack/filebeat/module/netflow/_meta/docs.asciidoc

@@ -0,0 +1,67 @@
+[role="xpack"]
+
+:modulename: netflow
+:has-dashboards: false
+
+== NetFlow module
+
+This is a module for receiving NetFlow and IPFIX flow records over UDP. This
+input supports NetFlow versions 1, 5, 6, 7, 8 and 9, as well as IPFIX. For
+NetFlow versions older than 9, fields are mapped automatically to NetFlow v9.
+
+This module wraps the <<filebeat-input-netflow,netflow input>> to enrich the
+flow records with geolocation information about the IP endpoints by using
+Elasticsearch Ingest Node.
+
+[float]
+=== Compatibility
+
+This module requires the {elasticsearch-plugins}/ingest-geoip.html[ingest-geoip]
+Elasticsearch plugins.
+
+include::../include/running-modules.asciidoc[]
+
+include::../include/configuring-intro.asciidoc[]
+
+:fileset_ex: log
+
+include::../include/config-option-intro.asciidoc[]
+
+[float]
+==== `log` fileset settings
+
+The fileset is by default configured to listen for UDP traffic on
+`localhost:2055`. For most uses cases you will want to set the `netflow_host`
+variable to allow the input bind to all interfaces so that it can receive
+traffic from network devices.
+
+["source","yaml",subs="attributes"]
+-----
+- module: netflow
+  log:
+    enabled: true
+    var:
+      netflow_host: 0.0.0.0
+      netflow_port: 2055
+-----
+
+`var.netflow_host`:: Address to find to. Defaults to `localhost`.
+
+`var.netflow_port`:: Port to listen on. Defaults to `2055`.
+
+`var.max_message_size`:: The maximum size of the message received over UDP.
+The default is `10KiB`.
+
+`var.expiration_timeout`:: The time before an idle session or unused template is
+expired. Only applicable to v9 and IPFIX protocols. A value of zero disables
+expiration.
+
+`var.queue_size`:: The maximum number of packets that can be queued for
+processing. Use this setting to avoid packet-loss when dealing with occasional
+bursts of traffic.
+
+:has-dashboards!:
+
+:fileset_ex!:
+
+:modulename!:

x-pack/filebeat/module/netflow/_meta/fields.yml

@@ -0,0 +1,6 @@
+- key: netflow-module
+  title: NetFlow
+  description: >
+    Module for receiving NetFlow and IPFIX flow records over UDP. The module
+    does not add fields beyond what the netflow input provides.
+  fields:

x-pack/filebeat/module/netflow/log/config/netflow.yml

@@ -0,0 +1,6 @@
+type: netflow
+protocols: [v1, v5, v6, v7, v8, v9, ipfix]
+host: '{{.netflow_host}}:{{.netflow_port}}'
+max_message_size: '{{.max_message_size}}'
+expiration_timeout: '{{.expiration_timeout}}'
+queue_size: {{.queue_size}}

x-pack/filebeat/module/netflow/log/ingest/pipeline.json

@@ -0,0 +1,29 @@
+{
+    "description": "Pipeline for Filebeat NetFlow",
+    "processors": [
+        {
+            "geoip": {
+                "if": "ctx.source?.geo == null",
+                "field": "source.ip",
+                "target_field": "source.geo",
+                "ignore_missing": true
+            }
+        },
+        {
+            "geoip": {
+                "if": "ctx.destination?.geo == null",
+                "field": "destination.ip",
+                "target_field": "destination.geo",
+                "ignore_missing": true
+            }
+        }
+    ],
+    "on_failure": [
+        {
+            "set": {
+                "field": "error.message",
+                "value": "{{ _ingest.on_failure_message }}"
+            }
+        }
+    ]
+}

x-pack/filebeat/module/netflow/log/manifest.yml

@@ -0,0 +1,20 @@
+module_version: "1.0"
+
+var:
+  - name: netflow_host
+    default: localhost
+  - name: netflow_port
+    default: 2055
+  - name: max_message_size
+    default: 10KiB
+  - name: expiration_timeout
+    default: 30m
+  - name: queue_size
+    default: 8192
+
+ingest_pipeline: ingest/pipeline.json
+input: config/netflow.yml
+
+requires.processors:
+- name: geoip
+  plugin: ingest-geoip

x-pack/filebeat/modules.d/netflow.yml.disabled

@@ -0,0 +1,9 @@
+# Module: netflow
+# Docs: https://www.elastic.co/guide/en/beats/filebeat/master/filebeat-module-netflow.html
+
+- module: netflow
+  log:
+    enabled: true
+    var:
+      netflow_host: localhost
+      netflow_port: 2055