Forwarder (#214)

* Minimal working refactor

* Replace flush func with a more flexible clearer

* Move chunk size into buffer options

* Fix formatting

* Add docs for buffer configs

* WIP

* Add forward test and format

* Add TLS listener test

* Add details about cert generation

* Add test for stanza forward output

* Full build

* Update changelog

* Add docs

* Add check for required address field

* Compare timestamps by string to avoid locale issues

* Use time.Equal to compare potentially different locales

Author: camdencheek

CHANGELOG.md

@@ -4,6 +4,10 @@ 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
+- New operators `forward_output` and `forward_input` to easily send log entries between stanza instances.
+
 ## [0.13.2] - 2020-11-17
 ### Added
 - New parameter `if` to parser plugins to allow for easy conditional parsing without routers

cmd/stanza/go.mod

@@ -3,7 +3,7 @@ module github.com/observiq/stanza/cmd/stanza
 go 1.14
 
 require (
-	github.com/kardianos/service v1.1.0
+	github.com/kardianos/service v1.2.0
 	github.com/observiq/stanza v0.12.5
 	github.com/observiq/stanza/operator/builtin/input/k8sevent v0.1.0
 	github.com/observiq/stanza/operator/builtin/input/windows v0.1.1

cmd/stanza/go.sum

@@ -628,6 +628,8 @@ github.com/jung-kurt/gofpdf v1.0.3-0.20190309125859-24315acbbda5/go.mod h1:7Id9E
 github.com/jwilder/encoding v0.0.0-20170811194829-b4e1701a28ef/go.mod h1:Ct9fl0F6iIOGgxJ5npU/IUOhOhqlVrGjyIZc8/MagT0=
 github.com/kardianos/service v1.1.0 h1:QV2SiEeWK42P0aEmGcsAgjApw/lRxkwopvT+Gu6t1/0=
 github.com/kardianos/service v1.1.0/go.mod h1:RrJI2xn5vve/r32U5suTbeaSGoMU6GbNPoj36CVYcHc=
+github.com/kardianos/service v1.2.0 h1:bGuZ/epo3vrt8IPC7mnKQolqFeYJb7Cs8Rk4PSOBB/g=
+github.com/kardianos/service v1.2.0/go.mod h1:CIMRFEJVL+0DS1a3Nx06NaMn4Dz63Ng6O7dl0qH0zVM=
 github.com/karrick/godirwalk v1.8.0/go.mod h1:H5KPZjojv4lE+QYImBI8xVtrBRgYrIVsaRPx4tDPEn4=
 github.com/karrick/godirwalk v1.10.3/go.mod h1:RoGL9dQei4vP9ilrpETWE8CLOZ1kiN0LhBygSwrAsHA=
 github.com/kisielk/errcheck v1.1.0/go.mod h1:EZBBE59ingxPouuu3KfxchcWSUPOHkagtvWXihfKN4Q=
@@ -1295,6 +1297,8 @@ golang.org/x/sys v0.0.0-20200803210538-64077c9b5642/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20200821140526-fda516888d29/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200905004654-be1d3432aa8f h1:Fqb3ao1hUmOR3GkUOg/Y+BadLwykBIzs5q8Ez2SbHyc=
 golang.org/x/sys v0.0.0-20200905004654-be1d3432aa8f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20201015000850-e3ed0017c211 h1:9UQO31fZ+0aKQOFldThf7BKPMJTiBfWycGh/u3UoO88=
+golang.org/x/sys v0.0.0-20201015000850-e3ed0017c211/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/text v0.0.0-20160726164857-2910a502d2bf/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=

cmd/stanza/init_common.go

@@ -3,6 +3,7 @@ package main
 import (
 	// Load packages when importing input operators
 	_ "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/k8sevent"
 	_ "github.com/observiq/stanza/operator/builtin/input/stanza"
@@ -28,6 +29,7 @@ import (
 	_ "github.com/observiq/stanza/operator/builtin/output/drop"
 	_ "github.com/observiq/stanza/operator/builtin/output/elastic"
 	_ "github.com/observiq/stanza/operator/builtin/output/file"
+	_ "github.com/observiq/stanza/operator/builtin/output/forward"
 	_ "github.com/observiq/stanza/operator/builtin/output/googlecloud"
 	_ "github.com/observiq/stanza/operator/builtin/output/newrelic"
 	_ "github.com/observiq/stanza/operator/builtin/output/otlp"

docs/operators/forward_input.md

@@ -0,0 +1,40 @@
+## `foward_input` operator
+
+The `foward_input` operator receives logs from another Stanza instance running `forward_output`.
+
+### Configuration Fields
+
+| Field            | Default          | Description                                           |
+| ---              | ---              | ---                                                   |
+| `id`             | `forward_output` | A unique identifier for the operator                  |
+| `listen_address` | `:80`            | The IP address and port to listen on                  |
+| `tls`            |                  | A block for configuring the server to listen with TLS |
+
+#### TLS block configuration
+
+| Field       | Default | Description                          |
+| ---         | ---     | ---                                  |
+| `cert_file` |         | The location of the certificate file |
+| `key_file`  |         | The location of the key file         |
+
+
+### Example Configurations
+
+#### Simple configuration
+
+Configuration:
+```yaml
+- type: forward_input
+  listen_address: ":25535"
+```
+
+#### TLS configuration
+
+Configuration:
+```yaml
+- type: forward_input
+  listen_address: ":25535"
+  tls:
+    cert_file: /tmp/public.crt
+    key_file: /tmp/private.key
+```

docs/operators/forward_output.md

@@ -0,0 +1,23 @@
+## `forward_output` operator
+
+The `forward_output` operator sends logs to another Stanza instance running `forward_input`.
+
+### Configuration Fields
+
+| Field     | Default          | Description                                                                              |
+| ---       | ---              | ---                                                                                      |
+| `id`      | `forward_output` | A unique identifier for the operator                                                     |
+| `address`      | required | The address that the downstream Stanza instance is listening on |
+| `buffer`  |                  | A [buffer](/docs/types/buffer.md) block indicating how to buffer entries before flushing |
+| `flusher` |                  | A [flusher](/docs/types/flusher.md) block configuring flushing behavior                  |
+
+
+### Example Configurations
+
+#### Simple configuration
+
+Configuration:
+```yaml
+- type: forward_output
+  address: "http://downstream_server:25535"
+```

go.mod

@@ -6,21 +6,22 @@ require (
 	github.com/antonmedv/expr v1.8.2
 	github.com/cenkalti/backoff/v4 v4.0.2
 	github.com/json-iterator/go v1.1.10
+	github.com/kardianos/service v1.2.0 // indirect
 	github.com/kr/text v0.2.0 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.1 // indirect
 	github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e // indirect
 	github.com/observiq/ctimefmt v1.0.0
 	github.com/observiq/nanojack v0.0.0-20201106172433-343928847ebc
 	github.com/pkg/errors v0.9.1 // indirect
+	github.com/spf13/cobra v1.1.1 // indirect
 	github.com/stretchr/objx v0.1.1 // indirect
 	github.com/stretchr/testify v1.6.1
 	go.etcd.io/bbolt v1.3.4
 	go.uber.org/zap v1.15.0
 	golang.org/x/exp v0.0.0-20200224162631-6cc2880d07d6 // indirect
 	golang.org/x/lint v0.0.0-20200302205851-738671d3881b // indirect
 	golang.org/x/sync v0.0.0-20200625203802-6e8e738ad208
-	golang.org/x/sys v0.0.0-20200905004654-be1d3432aa8f // indirect
 	golang.org/x/text v0.3.3
 	golang.org/x/tools v0.0.0-20200904185747-39188db58858 // indirect
 	gonum.org/v1/gonum v0.6.2