Feature: TCP Input

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

@@ -240,6 +240,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

@@ -535,6 +535,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()
+}