support max_message_length configuration for syslog parser

docs/sources/clients/promtail/configuration.md

@@ -771,6 +771,9 @@ labels:
 # When false, or if no timestamp is present on the syslog message, Promtail will assign the current timestamp to the log when it was processed.
 # Default is false
 use_incoming_timestamp: <bool>
+
+# Sets the maximum limit to the length of syslog messages
+max_message_length: <int>
 ```
 
 #### Available Labels

go.mod

@@ -33,7 +33,7 @@ require (
 	github.com/hashicorp/golang-lru v0.5.4
 	github.com/hpcloud/tail v1.0.0
 	github.com/imdario/mergo v0.3.9
-	github.com/influxdata/go-syslog/v3 v3.0.1-0.20200510134747-836dce2cf6da
+	github.com/influxdata/go-syslog/v3 v3.0.1-0.20201128200927-a1889d947b48
 	github.com/influxdata/telegraf v1.16.3
 	github.com/jmespath/go-jmespath v0.4.0
 	github.com/joncrlsn/dque v2.2.1-0.20200515025108-956d14155fa2+incompatible

go.sum

@@ -919,8 +919,8 @@ github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANyt
 github.com/influxdata/flux v0.65.0/go.mod h1:BwN2XG2lMszOoquQaFdPET8FRQfrXiZsWmcMO9rkaVY=
 github.com/influxdata/flux v0.65.1/go.mod h1:J754/zds0vvpfwuq7Gc2wRdVwEodfpCFM7mYlOw2LqY=
 github.com/influxdata/go-syslog/v2 v2.0.1/go.mod h1:hjvie1UTaD5E1fTnDmxaCw8RRDrT4Ve+XHr5O2dKSCo=
-github.com/influxdata/go-syslog/v3 v3.0.1-0.20200510134747-836dce2cf6da h1:yEuttQd/3jcdlUYYyDPub5y/JVCYR0UPuxH4xclZi/c=
-github.com/influxdata/go-syslog/v3 v3.0.1-0.20200510134747-836dce2cf6da/go.mod h1:aXdIdfn2OcGnMhOTojXmwZqXKgC3MU5riiNvzwwG9OY=
+github.com/influxdata/go-syslog/v3 v3.0.1-0.20201128200927-a1889d947b48 h1:0WbZ+ZVg74wbyQoRx1TD4D1Xoz8MsXJSTwdP9F7RMeQ=
+github.com/influxdata/go-syslog/v3 v3.0.1-0.20201128200927-a1889d947b48/go.mod h1:aXdIdfn2OcGnMhOTojXmwZqXKgC3MU5riiNvzwwG9OY=
 github.com/influxdata/influxdb v1.7.7/go.mod h1:qZna6X/4elxqT3yI9iZYdZrWWdeFOOprn86kgg4+IzY=
 github.com/influxdata/influxdb v1.8.0/go.mod h1:SIzcnsjaHRFpmlxpJ4S3NT64qtEKYweNTUMb/vh0OMQ=
 github.com/influxdata/influxdb v1.8.1/go.mod h1:SIzcnsjaHRFpmlxpJ4S3NT64qtEKYweNTUMb/vh0OMQ=

pkg/promtail/scrapeconfig/scrapeconfig.go

@@ -161,9 +161,12 @@ type SyslogTargetConfig struct {
 	// Labels optionally holds labels to associate with each record read from syslog.
 	Labels model.LabelSet `yaml:"labels"`
 
-	// UseIncomingTimestamp sets the timestamp to the incoming syslog mesages
+	// UseIncomingTimestamp sets the timestamp to the incoming syslog messages
 	// timestamp if it's set.
 	UseIncomingTimestamp bool `yaml:"use_incoming_timestamp"`
+
+	// MaxMessageLength sets the maximum limit to the length of syslog messages
+	MaxMessageLength int `yaml:"max_message_length"`
 }
 
 // WindowsEventsTargetConfig describes a scrape config that listen for windows event logs.

pkg/promtail/targets/syslog/syslogparser/syslogparser.go

@@ -14,7 +14,7 @@ import (
 // the callback function with the parsed messages. The parser automatically
 // detects octet counting.
 // The function returns on EOF or unrecoverable errors.
-func ParseStream(r io.Reader, callback func(res *syslog.Result)) error {
+func ParseStream(r io.Reader, callback func(res *syslog.Result), maxMessageLength int) error {
 	buf := bufio.NewReader(r)
 
 	firstByte, err := buf.Peek(1)
@@ -24,9 +24,9 @@ func ParseStream(r io.Reader, callback func(res *syslog.Result)) error {
 
 	b := firstByte[0]
 	if b == '<' {
-		nontransparent.NewParser(syslog.WithListener(callback)).Parse(buf)
+		nontransparent.NewParser(syslog.WithListener(callback), syslog.WithMaxMessageLength(maxMessageLength)).Parse(buf)
 	} else if b >= '0' && b <= '9' {
-		octetcounting.NewParser(syslog.WithListener(callback)).Parse(buf)
+		octetcounting.NewParser(syslog.WithListener(callback), syslog.WithMaxMessageLength(maxMessageLength)).Parse(buf)
 	} else {
 		return fmt.Errorf("invalid or unsupported framing. first byte: '%s'", firstByte)
 	}

pkg/promtail/targets/syslog/syslogparser/syslogparser_test.go

@@ -12,6 +12,10 @@ import (
 	"github.com/grafana/loki/pkg/promtail/targets/syslog/syslogparser"
 )
 
+var (
+	defaultMaxMessageLength = 8192
+)
+
 func TestParseStream_OctetCounting(t *testing.T) {
 	r := strings.NewReader("23 <13>1 - - - - - - First24 <13>1 - - - - - - Second")
 
@@ -20,7 +24,7 @@ func TestParseStream_OctetCounting(t *testing.T) {
 		results = append(results, res)
 	}
 
-	err := syslogparser.ParseStream(r, cb)
+	err := syslogparser.ParseStream(r, cb, defaultMaxMessageLength)
 	require.NoError(t, err)
 
 	require.Equal(t, 2, len(results))
@@ -30,6 +34,21 @@ func TestParseStream_OctetCounting(t *testing.T) {
 	require.Equal(t, "Second", *results[1].Message.(*rfc5424.SyslogMessage).Message)
 }
 
+func TestParseStream_OctetCounting_LongMessage(t *testing.T) {
+	r := strings.NewReader("8198 <13>1 - - - - - - First")
+
+	results := make([]*syslog.Result, 0)
+	cb := func(res *syslog.Result) {
+		results = append(results, res)
+	}
+
+	err := syslogparser.ParseStream(r, cb, defaultMaxMessageLength)
+	require.NoError(t, err)
+
+	require.Equal(t, 1, len(results))
+	require.EqualError(t, results[0].Error, "message too long to parse. was size 8198, max length 8192")
+}
+
 func TestParseStream_NewlineSeparated(t *testing.T) {
 	r := strings.NewReader("<13>1 - - - - - - First\n<13>1 - - - - - - Second\n")
 
@@ -38,7 +57,7 @@ func TestParseStream_NewlineSeparated(t *testing.T) {
 		results = append(results, res)
 	}
 
-	err := syslogparser.ParseStream(r, cb)
+	err := syslogparser.ParseStream(r, cb, defaultMaxMessageLength)
 	require.NoError(t, err)
 
 	require.Equal(t, 2, len(results))
@@ -51,13 +70,13 @@ func TestParseStream_NewlineSeparated(t *testing.T) {
 func TestParseStream_InvalidStream(t *testing.T) {
 	r := strings.NewReader("invalid")
 
-	err := syslogparser.ParseStream(r, func(res *syslog.Result) {})
+	err := syslogparser.ParseStream(r, func(res *syslog.Result) {}, defaultMaxMessageLength)
 	require.EqualError(t, err, "invalid or unsupported framing. first byte: 'i'")
 }
 
 func TestParseStream_EmptyStream(t *testing.T) {
 	r := strings.NewReader("")
 
-	err := syslogparser.ParseStream(r, func(res *syslog.Result) {})
+	err := syslogparser.ParseStream(r, func(res *syslog.Result) {}, defaultMaxMessageLength)
 	require.Equal(t, err, io.EOF)
 }

pkg/promtail/targets/syslog/syslogtarget.go

@@ -27,7 +27,8 @@ import (
 )
 
 var (
-	defaultIdleTimeout = 120 * time.Second
+	defaultIdleTimeout      = 120 * time.Second
+	defaultMaxMessageLength = 8192
 )
 
 // SyslogTarget listens to syslog messages.
@@ -153,7 +154,7 @@ func (t *SyslogTarget) handleConnection(cn net.Conn) {
 			return
 		}
 		t.handleMessage(connLabels.Copy(), msg.Message)
-	})
+	}, t.maxMessageLength())
 
 	if err != nil {
 		level.Warn(t.logger).Log("msg", "error initializing syslog stream", "err", err)
@@ -310,12 +311,19 @@ func (t *SyslogTarget) ListenAddress() net.Addr {
 }
 
 func (t *SyslogTarget) idleTimeout() time.Duration {
-	if tm := t.config.IdleTimeout; tm != 0 {
-		return tm
+	if t.config.IdleTimeout != 0 {
+		return t.config.IdleTimeout
 	}
 	return defaultIdleTimeout
 }
 
+func (t *SyslogTarget) maxMessageLength() int {
+	if t.config.MaxMessageLength != 0 {
+		return t.config.MaxMessageLength
+	}
+	return defaultMaxMessageLength
+}
+
 type idleTimeoutConn struct {
 	net.Conn
 	idleTimeout time.Duration

vendor/modules.txt

@@ -555,7 +555,7 @@ github.com/hpcloud/tail/winfile
 # github.com/imdario/mergo v0.3.9
 ## explicit
 github.com/imdario/mergo
-# github.com/influxdata/go-syslog/v3 v3.0.1-0.20200510134747-836dce2cf6da
+# github.com/influxdata/go-syslog/v3 v3.0.1-0.20201128200927-a1889d947b48
 ## explicit
 github.com/influxdata/go-syslog/v3
 github.com/influxdata/go-syslog/v3/common