Goflow input operator (#332)

* WIP, just playing around with netflow

* wip

* sflow

* convert byte keys

* pass context to Publish. Remove go func, not needed

* fix error message

* refactor Publish methods

* replace map keys in Parse() instead of copying the map, benchmark results are faster

* add tests

* remove early return

* remove duplicate from init. hardcode reuse to true

* initial netflow docs

* make tidy

* remove reuse from test

* re-combine all 3 goflow operators into goflow (previously netflow) package. add  param

* refactor WriteGoFlowMessage

* promote timestamp

* enhance parse testing

* integration test for netflow v5

* refactor publish

* use stanza's logging for goflow logging interface

* Refactor startup handling, stop stanza if Goflow fails at anytime

* Refactor goflow process handling. verify that the socket is available during Build(), implement automatic restart on goflow failure with a backoff

* refactor test, ensure file_output content is correct

* check sampler address, it is based on the source ip

* we need docker for tests

* setup int tests at a later time

* Added Goflow operator for receiving Netflow (v5, v9, ipfix) and Sflow

* cleanup commented fields

* rename netflow v9 --> netflow ipfix. ipfix is the latest version and is backwards compatible with v9

* use const values that already exist

* use netflow_ipfix as default mode

* use ListenAddress instead of address + port. Set mode default before checking if valid

Author: Joseph Sirianni

.gitignore

@@ -7,3 +7,4 @@ local/*
 artifacts/*
 **/.vscode/*
 gen/
+**/testdata/*.log

CHANGELOG.md

@@ -4,6 +4,11 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## Unreleased
+
+### Added
+- Added Goflow operator for receiving Netflow (v5, v9, ipfix) and Sflow [PR 332](https://github.com/observIQ/stanza/pull/332)
+
 ## 1.0.1 - 2021-06-16
 
 ### Fixed

Makefile

@@ -23,6 +23,14 @@ test: vet test-only
 test-only:
 	$(MAKE) for-all CMD="go test -race -coverprofile coverage.txt -coverpkg ./... ./..."
 
+.PHONY: test-integration
+test-integration:
+	mkdir -p artifacts
+	curl -fL https://github.com/observiq/stanza-plugins/releases/latest/download/stanza-plugins.tar.gz -o ./artifacts/stanza-plugins.tar.gz
+	docker build . -t stanza-integration:latest
+	$(MAKE) for-all CMD="go clean -testcache ./... ./..."
+	$(MAKE) for-all CMD="go test -tags integration ./... ./..."
+
 .PHONY: bench
 bench:
 	$(MAKE) for-all CMD="go test -run=NONE -bench '.*' ./... -benchmem"

cmd/stanza/init_common.go

@@ -8,6 +8,7 @@ import (
 	_ "github.com/observiq/stanza/operator/builtin/input/file"
 	_ "github.com/observiq/stanza/operator/builtin/input/forward"
 	_ "github.com/observiq/stanza/operator/builtin/input/generate"
+	_ "github.com/observiq/stanza/operator/builtin/input/goflow"
 	_ "github.com/observiq/stanza/operator/builtin/input/k8sevent"
 	_ "github.com/observiq/stanza/operator/builtin/input/stanza"
 	_ "github.com/observiq/stanza/operator/builtin/input/stdin"

docs/operators/goflow_input.md

@@ -0,0 +1,57 @@
+## `goflow_input` operator
+
+The `goflow_input` operator recieves Netflow v9 / IPFIX, Netflow v5, and Sflow messages from network devices. `goflow_input` implements [Goflow](https://github.com/cloudflare/goflow).
+
+The `timereceived` field is promoted as the entries Timestamp.
+
+### Configuration Fields
+
+| Field        | Default             | Description                                                                                   |
+| ---          | ---                 | ---                                                                                           |
+| `id`         | `goflow_input`      | A unique identifier for the operator                                                          |
+| `output`     | Next in pipeline    | The connected operator(s) that will receive all outbound entries                              |
+| `mode`       | `netflow_ipfix`     | The Goflow mode [`netflow_ipfix`, `netflow_v5`, `sflow`]                                      |
+| `address`    | `0.0.0.0`           | The ip address to bind to                                                                     |
+| `port`       | required            | The port to bind to                                                                           |
+| `workers`    | `1`                 | Number of worker processes spawned by the underlying [Goflow package](https://github.com/cloudflare/goflow)  |
+
+### Example Configuration
+
+Configuration:
+```yaml
+pipeline:
+- type: goflow_input
+  mode: netflow_v5
+  port: 2000
+- type: stdout
+```
+
+### Example Output
+
+```json
+{
+  "timestamp": "2021-06-15T11:59:26-04:00",
+  "severity": 0,
+  "record": {
+    "bytes": 936,
+    "dstaddr": "173.195.121.172",
+    "dstas": 14164,
+    "dstnet": 5,
+    "dstport": 17210,
+    "etype": 2048,
+    "nexthop": "66.88.34.2",
+    "packets": 100,
+    "proto": 6,
+    "sampleraddress": "172.17.0.2",
+    "sequencenum": 7,
+    "srcaddr": "241.104.80.243",
+    "srcas": 43137,
+    "srcnet": 11,
+    "srcport": 37247,
+    "timeflowend": 1623772766,
+    "timeflowstart": 1623772766,
+    "type": 2
+  }
+}
+
+```

go.mod

@@ -9,7 +9,9 @@ require (
 	github.com/aws/aws-sdk-go v1.38.31
 	github.com/bmatcuk/doublestar/v2 v2.0.4
 	github.com/cenkalti/backoff/v4 v4.1.1
+	github.com/cloudflare/goflow/v3 v3.4.2
 	github.com/elastic/go-elasticsearch/v7 v7.13.0
+	github.com/fatih/structs v1.1.0
 	github.com/golang/protobuf v1.5.2
 	github.com/hashicorp/go-uuid v1.0.2
 	github.com/jpillora/backoff v1.0.0
@@ -19,8 +21,10 @@ require (
 	github.com/observiq/ctimefmt v1.0.0
 	github.com/observiq/go-syslog/v3 v3.0.2
 	github.com/observiq/nanojack v0.0.0-20201106172433-343928847ebc
+	github.com/sirupsen/logrus v1.7.0
 	github.com/spf13/cobra v1.1.3
 	github.com/stretchr/testify v1.7.0
+	github.com/testcontainers/testcontainers-go v0.11.0
 	go.etcd.io/bbolt v1.3.5
 	go.opentelemetry.io/collector v0.13.0
 	go.uber.org/multierr v1.5.0