Filebeat TCP Input (#6700)

Allow to receive new events via TCP, this will create a new event per
line and add some useful information about the connected client to the
evant. This input is marked as **experimental**.

This input expose the following settings:

- `line_delimiter`: The characters used to split incoming events, by
default it will split on `\n` and will also strip both `\n` and `\r`.
You can also use any characters like `;` and you can also used multiple
characters delimiter like `<END>`, the delimiter tokens will always be
removed from the string.

- `max_message_size`: This is a number of bytes that a client can buffer
in memory before finding a new message, if the limit is reached the
connection is killed and the event is logged. This is to prevent rogue
clients to DoS attack by consuming all the available memory. The default
values is 20971520.

- `timeout`: The server will close any client that reach this inactivity
timeout.

Ref: #5862

CHANGELOG.asciidoc

@@ -52,6 +52,7 @@ https://github.com/elastic/beats/compare/v6.0.0-beta2...master[Check the HEAD di
 - Remove the undefined `username` option from the Redis input and clarify the documentation. {pull}6662[6662]
 - Add validation for Stdin, when Filebeat is configured with Stdin and any other inputs, Filebeat
   will now refuses to start. {pull}6463[6463]
+- Addition of the TCP input {pull}6700[6700]
 
 *Heartbeat*
 

filebeat/_meta/common.reference.p2.yml

@@ -241,6 +241,23 @@ filebeat.inputs:
   # Maximum size of the message received over UDP
   #max_message_size: 10240
 
+#------------------------------ TCP prospector --------------------------------
+# Experimental: Config options for the TCP input
+#- type: tcp
+  #enabled: false
+
+  # The host and port to receive the new event
+  #host: "localhost:9000"
+
+  # Character used to split new message
+  #line_delimiter: "\n"
+
+  # Maximum size in bytes of the message received over TCP
+  #max_message_size: 20MiB
+
+  # The number of seconds of inactivity before a remote connection is closed.
+  #timeout: 300s
+
 #========================== Filebeat autodiscover ==============================
 
 # Autodiscover allows you to detect changes in the system and spawn new modules

filebeat/docs/filebeat-options.asciidoc

@@ -14,7 +14,7 @@ and configuring modules.
 To configure {beatname_uc} manually (instead of using
 <<{beatname_lc}-modules-overview,modules>>), you specify a list of inputs in the
 +{beatname_lc}.inputs+ section of the +{beatname_lc}.yml+. Inputs specify how
-{beatname_uc} locates and processes input data.  
+{beatname_uc} locates and processes input data.
 
 The list is a http://yaml.org/[YAML] array, so each input begins with
 a dash (`-`). You can specify multiple inputs, and you can specify the same
@@ -47,6 +47,7 @@ You can configure {beatname_uc} to use the following inputs:
 * <<{beatname_lc}-input-redis>>
 * <<{beatname_lc}-input-udp>>
 * <<{beatname_lc}-input-docker>>
+* <<{beatname_lc}-input-tcp>>
 
 
 
@@ -59,3 +60,5 @@ include::inputs/input-redis.asciidoc[]
 include::inputs/input-udp.asciidoc[]
 
 include::inputs/input-docker.asciidoc[]
+
+include::inputs/input-tcp.asciidoc[]

filebeat/docs/inputs/input-tcp.asciidoc

@@ -0,0 +1,55 @@
+:type: tcp
+
+[id="{beatname_lc}-input-{type}"]
+=== TCP input
+
+++++
+<titleabbrev>TCP</titleabbrev>
+++++
+
+Use the `TCP` input to read events over TCP.
+
+Example configuration:
+
+["source","yaml",subs="attributes"]
+----
+{beatname_lc}.inputs:
+- type: tcp
+  max_message_size: 10240
+  host: "localhost:9000"
+----
+
+
+==== Configuration options
+
+The `tcp` input supports the following configuration options plus the
+<<{beatname_lc}-input-{type}-common-options>> described later.
+
+[float]
+[id="{beatname_lc}-input-{type}-max-message-size"]
+==== `max_message_size`
+
+The maximum size of the message received over TCP. The default is `20MiB`.
+
+[float]
+[id="{beatname_lc}-input-{type}-host"]
+==== `host`
+
+The host and TCP port to listen on for event streams.
+
+[float]
+[id="{beatname_lc}-input-{type}-line-delimiter"]
+==== `line_delimiter`
+
+Specify the characters used to split the incoming events. The default is '\n'.
+
+[float]
+[id="{beatname_lc}-input-{type}-timeout"]
+==== `timeout`
+
+The number of seconds of inactivity before a remote connection is closed. The default is `300s`.
+
+[id="{beatname_lc}-input-{type}-common-options"]
+include::../inputs/input-common-options.asciidoc[]
+
+:type!:

filebeat/filebeat.reference.yml

@@ -536,6 +536,23 @@ filebeat.inputs:
   # Maximum size of the message received over UDP
   #max_message_size: 10240
 
+#------------------------------ TCP prospector --------------------------------
+# Experimental: Config options for the TCP input
+#- type: tcp
+  #enabled: false
+
+  # The host and port to receive the new event
+  #host: "localhost:9000"
+
+  # Character used to split new message
+  #line_delimiter: "\n"
+
+  # Maximum size in bytes of the message received over TCP
+  #max_message_size: 20MiB
+
+  # The number of seconds of inactivity before a remote connection is closed.
+  #timeout: 300s
+
 #========================== Filebeat autodiscover ==============================
 
 # Autodiscover allows you to detect changes in the system and spawn new modules

filebeat/include/list.go

@@ -12,5 +12,6 @@ import (
 	_ "github.com/elastic/beats/filebeat/input/log"
 	_ "github.com/elastic/beats/filebeat/input/redis"
 	_ "github.com/elastic/beats/filebeat/input/stdin"
+	_ "github.com/elastic/beats/filebeat/input/tcp"
 	_ "github.com/elastic/beats/filebeat/input/udp"
 )

filebeat/input/tcp/config.go

@@ -0,0 +1,26 @@
+package tcp
+
+import (
+	"time"
+
+	"github.com/dustin/go-humanize"
+
+	"github.com/elastic/beats/filebeat/harvester"
+	"github.com/elastic/beats/filebeat/inputsource/tcp"
+)
+
+type config struct {
+	tcp.Config                `config:",inline"`
+	harvester.ForwarderConfig `config:",inline"`
+}
+
+var defaultConfig = config{
+	ForwarderConfig: harvester.ForwarderConfig{
+		Type: "tcp",
+	},
+	Config: tcp.Config{
+		LineDelimiter:  "\n",
+		Timeout:        time.Minute * 5,
+		MaxMessageSize: 20 * humanize.MiByte,
+	},
+}

filebeat/input/tcp/input.go

@@ -0,0 +1,116 @@
+package tcp
+
+import (
+	"sync"
+	"time"
+
+	"github.com/elastic/beats/filebeat/channel"
+	"github.com/elastic/beats/filebeat/harvester"
+	"github.com/elastic/beats/filebeat/input"
+	"github.com/elastic/beats/filebeat/inputsource/tcp"
+	"github.com/elastic/beats/filebeat/util"
+	"github.com/elastic/beats/libbeat/beat"
+	"github.com/elastic/beats/libbeat/common"
+	"github.com/elastic/beats/libbeat/common/cfgwarn"
+	"github.com/elastic/beats/libbeat/logp"
+)
+
+func init() {
+	err := input.Register("tcp", NewInput)
+	if err != nil {
+		panic(err)
+	}
+}
+
+// Input for TCP connection
+type Input struct {
+	sync.Mutex
+	server  *tcp.Server
+	started bool
+	outlet  channel.Outleter
+	config  *config
+	log     *logp.Logger
+}
+
+// NewInput creates a new TCP input
+func NewInput(
+	cfg *common.Config,
+	outlet channel.Factory,
+	context input.Context,
+) (input.Input, error) {
+	cfgwarn.Experimental("TCP input type is used")
+
+	out, err := outlet(cfg, context.DynamicFields)
+	if err != nil {
+		return nil, err
+	}
+
+	forwarder := harvester.NewForwarder(out)
+
+	config := defaultConfig
+	err = cfg.Unpack(&config)
+	if err != nil {
+		return nil, err
+	}
+
+	cb := func(data []byte, metadata tcp.Metadata) {
+		event := createEvent(data, metadata)
+		forwarder.Send(event)
+	}
+
+	server, err := tcp.New(cb, &config.Config)
+	if err != nil {
+		return nil, err
+	}
+
+	return &Input{
+		server:  server,
+		started: false,
+		outlet:  out,
+		config:  &config,
+		log:     logp.NewLogger("tcp input").With(config.Config.Host),
+	}, nil
+}
+
+// Run start a TCP input
+func (p *Input) Run() {
+	p.Lock()
+	defer p.Unlock()
+
+	if !p.started {
+		p.log.Info("Starting TCP input")
+		err := p.server.Start()
+		if err != nil {
+			p.log.Errorw("Error starting the TCP server", "error", err)
+		}
+		p.started = true
+	}
+}
+
+// Stop stops TCP server
+func (p *Input) Stop() {
+	defer p.outlet.Close()
+	p.Lock()
+	defer p.Unlock()
+
+	p.log.Info("Stopping TCP input")
+	p.server.Stop()
+	p.started = false
+}
+
+// Wait stop the current server
+func (p *Input) Wait() {
+	p.Stop()
+}
+
+func createEvent(raw []byte, metadata tcp.Metadata) *util.Data {
+	data := util.NewData()
+	data.Event = beat.Event{
+		Timestamp: time.Now(),
+		Fields: common.MapStr{
+			"message": string(raw),
+			"source":  metadata.RemoteAddr.String(),
+		},
+	}
+	return data
+}

filebeat/input/tcp/input_test.go

@@ -0,0 +1,30 @@
+package tcp
+
+import (
+	"net"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/elastic/beats/filebeat/inputsource/tcp"
+)
+
+func TestCreateEvent(t *testing.T) {
+	hello := "hello world"
+	ip := "127.0.0.1"
+	parsedIP := net.ParseIP(ip)
+	addr := &net.IPAddr{IP: parsedIP, Zone: ""}
+
+	message := []byte(hello)
+	mt := tcp.Metadata{RemoteAddr: addr}
+
+	data := createEvent(message, mt)
+	event := data.GetEvent()
+
+	m, err := event.GetValue("message")
+	assert.NoError(t, err)
+	assert.Equal(t, string(message), m)
+
+	from, _ := event.GetValue("source")
+	assert.Equal(t, ip, from)
+}

filebeat/inputsource/tcp/client.go

@@ -0,0 +1,78 @@
+package tcp
+
+import (
+	"bufio"
+	"net"
+	"time"
+
+	"github.com/pkg/errors"
+
+	"github.com/elastic/beats/libbeat/logp"
+)
+
+// Client is a remote client.
+type client struct {
+	conn           net.Conn
+	log            *logp.Logger
+	callback       CallbackFunc
+	done           chan struct{}
+	metadata       Metadata
+	splitFunc      bufio.SplitFunc
+	maxReadMessage size
+	timeout        time.Duration
+}
+
+func newClient(
+	conn net.Conn,
+	log *logp.Logger,
+	callback CallbackFunc,
+	splitFunc bufio.SplitFunc,
+	maxReadMessage size,
+	timeout time.Duration,
+) *client {
+	client := &client{
+		conn:           conn,
+		log:            log.With("address", conn.RemoteAddr()),
+		callback:       callback,
+		done:           make(chan struct{}),
+		splitFunc:      splitFunc,
+		maxReadMessage: maxReadMessage,
+		timeout:        timeout,
+		metadata: Metadata{
+			RemoteAddr: conn.RemoteAddr(),
+		},
+	}
+	return client
+}
+
+func (c *client) handle() error {
+	r := NewResetableLimitedReader(NewDeadlineReader(c.conn, c.timeout), uint64(c.maxReadMessage))
+	buf := bufio.NewReader(r)
+	scanner := bufio.NewScanner(buf)
+	scanner.Split(c.splitFunc)
+
+	for scanner.Scan() {
+		err := scanner.Err()
+		if err != nil {
+			// we are forcing a close on the socket, lets ignore any error that could happen.
+			select {
+			case <-c.done:
+				break
+			default:
+			}
+			// This is a user defined limit and we should notify the user.
+			if IsMaxReadBufferErr(err) {
+				c.log.Errorw("client errors", "error", err)
+			}
+			return errors.Wrap(err, "tcp client error")
+		}
+		r.Reset()
+		c.callback(scanner.Bytes(), c.metadata)
+	}
+	return nil
+}
+
+func (c *client) close() {
+	close(c.done)
+	c.conn.Close()
+}