Make inputsource generic taking bufio.SplitFunc as input

CHANGELOG.asciidoc

@@ -56,6 +56,7 @@ https://github.com/elastic/beats/compare/v6.4.0...master[Check the HEAD diff]
 
 *Filebeat*
 
+- Make inputsource generic taking bufio.SplitFunc as input {pull}7746[7746]
 - Add custom unpack to log hints config to avoid env resolution {pull}7710[7710]
 
 *Heartbeat*

filebeat/docs/inputs/input-syslog.asciidoc

@@ -33,11 +33,11 @@ Example configurations:
 The `syslog` input supports protocol specific configuration options plus the
 <<{beatname_lc}-input-{type}-common-options>> described later.
 
-Protocol `udp`:
+===== Protocol `udp`:
 
 include::../inputs/input-common-udp-options.asciidoc[]
 
-Protocol `tcp`:
+===== Protocol `tcp`:
 
 include::../inputs/input-common-tcp-options.asciidoc[]
 

filebeat/input/syslog/config.go

@@ -41,10 +41,17 @@ var defaultConfig = config{
 	},
 }
 
-var defaultTCP = tcp.Config{
-	LineDelimiter:  "\n",
-	Timeout:        time.Minute * 5,
-	MaxMessageSize: 20 * humanize.MiByte,
+type syslogTCP struct {
+	tcp.Config    `config:",inline"`
+	LineDelimiter string `config:"line_delimiter" validate:"nonzero"`
+}
+
+var defaultTCP = syslogTCP{
+	Config: tcp.Config{
+		Timeout:        time.Minute * 5,
+		MaxMessageSize: 20 * humanize.MiByte,
+	},
+	LineDelimiter: "\n",
 }
 
 var defaultUDP = udp.Config{
@@ -64,7 +71,13 @@ func factory(
 		if err := cfg.Unpack(&config); err != nil {
 			return nil, err
 		}
-		return tcp.New(&config, cb)
+
+		splitFunc := tcp.SplitFunc([]byte(config.LineDelimiter))
+		if splitFunc == nil {
+			return nil, fmt.Errorf("error creating splitFunc from delimiter %s", config.LineDelimiter)
+		}
+
+		return tcp.New(&config.Config, splitFunc, cb)
 	case udp.Name:
 		config := defaultUDP
 		if err := cfg.Unpack(&config); err != nil {

filebeat/input/tcp/config.go

@@ -29,15 +29,17 @@ import (
 type config struct {
 	tcp.Config                `config:",inline"`
 	harvester.ForwarderConfig `config:",inline"`
+
+	LineDelimiter string `config:"line_delimiter" validate:"nonzero"`
 }
 
 var defaultConfig = config{
 	ForwarderConfig: harvester.ForwarderConfig{
 		Type: "tcp",
 	},
 	Config: tcp.Config{
-		LineDelimiter:  "\n",
 		Timeout:        time.Minute * 5,
 		MaxMessageSize: 20 * humanize.MiByte,
 	},
+	LineDelimiter: "\n",
 }

filebeat/input/tcp/input.go

@@ -18,6 +18,7 @@
 package tcp
 
 import (
+	"fmt"
 	"sync"
 	"time"
 
@@ -76,7 +77,12 @@ func NewInput(
 		forwarder.Send(event)
 	}
 
-	server, err := tcp.New(&config.Config, cb)
+	splitFunc := tcp.SplitFunc([]byte(config.LineDelimiter))
+	if splitFunc == nil {
+		return nil, fmt.Errorf("unable to create splitFunc for delimiter %s", config.LineDelimiter)
+	}
+
+	server, err := tcp.New(&config.Config, splitFunc, cb)
 	if err != nil {
 		return nil, err
 	}

filebeat/inputsource/tcp/config.go

@@ -33,7 +33,6 @@ type size uint64
 // Config exposes the tcp configuration.
 type Config struct {
 	Host           string                  `config:"host"`
-	LineDelimiter  string                  `config:"line_delimiter" validate:"nonzero"`
 	Timeout        time.Duration           `config:"timeout" validate:"nonzero,positive"`
 	MaxMessageSize cfgtype.ByteSize        `config:"max_message_size" validate:"nonzero,positive"`
 	TLS            *tlscommon.ServerConfig `config:"ssl"`

filebeat/inputsource/tcp/server.go

@@ -48,25 +48,24 @@ type Server struct {
 // New creates a new tcp server
 func New(
 	config *Config,
+	splitFunc bufio.SplitFunc,
 	callback inputsource.NetworkFunc,
 ) (*Server, error) {
-
-	if len(config.LineDelimiter) == 0 {
-		return nil, fmt.Errorf("empty line delimiter")
-	}
-
 	tlsConfig, err := tlscommon.LoadTLSServerConfig(config.TLS)
 	if err != nil {
 		return nil, err
 	}
 
-	sf := splitFunc([]byte(config.LineDelimiter))
+	if splitFunc == nil {
+		return nil, fmt.Errorf("SplitFunc can't be empty")
+	}
+
 	return &Server{
 		config:    config,
 		callback:  callback,
 		clients:   make(map[*client]struct{}, 0),
 		done:      make(chan struct{}),
-		splitFunc: sf,
+		splitFunc: splitFunc,
 		log:       logp.NewLogger("tcp").With("address", config.Host),
 		tlsConfig: tlsConfig,
 	}, nil
@@ -190,7 +189,8 @@ func (s *Server) clientsCount() int {
 	return len(s.clients)
 }
 
-func splitFunc(lineDelimiter []byte) bufio.SplitFunc {
+// SplitFunc allows to create a `bufio.SplitFunc` based on a delimiter provided.
+func SplitFunc(lineDelimiter []byte) bufio.SplitFunc {
 	ld := []byte(lineDelimiter)
 	if bytes.Equal(ld, []byte("\n")) {
 		// This will work for most usecases and will also strip \r if present.

filebeat/inputsource/tcp/server_test.go

@@ -18,6 +18,7 @@
 package tcp
 
 import (
+	"bufio"
 	"fmt"
 	"math/rand"
 	"net"
@@ -33,7 +34,6 @@ import (
 )
 
 var defaultConfig = Config{
-	LineDelimiter:  "\n",
 	Timeout:        time.Minute * 5,
 	MaxMessageSize: 20 * humanize.MiByte,
 }
@@ -44,11 +44,7 @@ type info struct {
 }
 
 func TestErrorOnEmptyLineDelimiter(t *testing.T) {
-	cfg := map[string]interface{}{
-		"line_delimiter": "",
-	}
-
-	c, _ := common.NewConfigFrom(cfg)
+	c := common.NewConfig()
 	config := defaultConfig
 	err := c.Unpack(&config)
 	assert.Error(t, err)
@@ -61,85 +57,83 @@ func TestReceiveEventsAndMetadata(t *testing.T) {
 	tests := []struct {
 		name             string
 		cfg              map[string]interface{}
+		splitFunc        bufio.SplitFunc
 		expectedMessages []string
 		messageSent      string
 	}{
 		{
 			name:             "NewLine",
 			cfg:              map[string]interface{}{},
+			splitFunc:        SplitFunc([]byte("\n")),
 			expectedMessages: expectedMessages,
 			messageSent:      strings.Join(expectedMessages, "\n"),
 		},
 		{
 			name:             "NewLineWithCR",
 			cfg:              map[string]interface{}{},
+			splitFunc:        SplitFunc([]byte("\r\n")),
 			expectedMessages: expectedMessages,
 			messageSent:      strings.Join(expectedMessages, "\r\n"),
 		},
 		{
-			name: "CustomDelimiter",
-			cfg: map[string]interface{}{
-				"line_delimiter": ";",
-			},
+			name:             "CustomDelimiter",
+			cfg:              map[string]interface{}{},
+			splitFunc:        SplitFunc([]byte(";")),
 			expectedMessages: expectedMessages,
 			messageSent:      strings.Join(expectedMessages, ";"),
 		},
 		{
-			name: "MultipleCharsCustomDelimiter",
-			cfg: map[string]interface{}{
-				"line_delimiter": "<END>",
-			},
+			name:             "MultipleCharsCustomDelimiter",
+			cfg:              map[string]interface{}{},
+			splitFunc:        SplitFunc([]byte("<END>")),
 			expectedMessages: expectedMessages,
 			messageSent:      strings.Join(expectedMessages, "<END>"),
 		},
 		{
-			name: "SingleCharCustomDelimiterMessageWithoutBoundaries",
-			cfg: map[string]interface{}{
-				"line_delimiter": ";",
-			},
+			name:             "SingleCharCustomDelimiterMessageWithoutBoundaries",
+			cfg:              map[string]interface{}{},
+			splitFunc:        SplitFunc([]byte(";")),
 			expectedMessages: []string{"hello"},
 			messageSent:      "hello",
 		},
 		{
-			name: "MultipleCharCustomDelimiterMessageWithoutBoundaries",
-			cfg: map[string]interface{}{
-				"line_delimiter": "<END>",
-			},
+			name:             "MultipleCharCustomDelimiterMessageWithoutBoundaries",
+			cfg:              map[string]interface{}{},
+			splitFunc:        SplitFunc([]byte("<END>")),
 			expectedMessages: []string{"hello"},
 			messageSent:      "hello",
 		},
 		{
-			name: "NewLineMessageWithoutBoundaries",
-			cfg: map[string]interface{}{
-				"line_delimiter": "\n",
-			},
+			name:             "NewLineMessageWithoutBoundaries",
+			cfg:              map[string]interface{}{},
+			splitFunc:        SplitFunc([]byte("\n")),
 			expectedMessages: []string{"hello"},
 			messageSent:      "hello",
 		},
 		{
-			name: "NewLineLargeMessagePayload",
-			cfg: map[string]interface{}{
-				"line_delimiter": "\n",
-			},
+			name:             "NewLineLargeMessagePayload",
+			cfg:              map[string]interface{}{},
+			splitFunc:        SplitFunc([]byte("\n")),
 			expectedMessages: largeMessages,
 			messageSent:      strings.Join(largeMessages, "\n"),
 		},
 		{
-			name: "CustomLargeMessagePayload",
-			cfg: map[string]interface{}{
-				"line_delimiter": ";",
-			},
+			name:             "CustomLargeMessagePayload",
+			cfg:              map[string]interface{}{},
+			splitFunc:        SplitFunc([]byte(";")),
 			expectedMessages: largeMessages,
 			messageSent:      strings.Join(largeMessages, ";"),
 		},
 		{
 			name:             "MaxReadBufferReached",
 			cfg:              map[string]interface{}{},
+			splitFunc:        SplitFunc([]byte("\n")),
 			expectedMessages: []string{},
 			messageSent:      randomString(900000),
 		},
 		{
-			name: "MaxReadBufferReachedUserConfigured",
+			name:      "MaxReadBufferReachedUserConfigured",
+			splitFunc: SplitFunc([]byte("\n")),
 			cfg: map[string]interface{}{
 				"max_read_message": 50000,
 			},
@@ -162,7 +156,7 @@ func TestReceiveEventsAndMetadata(t *testing.T) {
 			if !assert.NoError(t, err) {
 				return
 			}
-			server, err := New(&config, to)
+			server, err := New(&config, test.splitFunc, to)
 			if !assert.NoError(t, err) {
 				return
 			}
@@ -212,7 +206,7 @@ func TestReceiveNewEventsConcurrently(t *testing.T) {
 	if !assert.NoError(t, err) {
 		return
 	}
-	server, err := New(&config, to)
+	server, err := New(&config, bufio.ScanLines, to)
 	if !assert.NoError(t, err) {
 		return
 	}