support max_message_length configuration for syslog parser (#3400)

* support max_message_length configuration for syslog parser

* address review comments

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

@@ -34,7 +34,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

@@ -937,8 +937,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/github.com/influxdata/go-syslog/v3/README.md

@@ -4,11 +4,13 @@
 
 > [Blazing fast](#Performances) Syslog parsers
 
+_By [@leodido](https://github.com/leodido)_.
+
 To wrap up, this package provides:
 
 - a [RFC5424-compliant parser and builder](/rfc5424)
 - a [RFC3164-compliant parser](/rfc3164) - ie., BSD-syslog messages
-- a parser which works on streams for syslog with [octet counting](https://tools.ietf.org/html/rfc5425#section-4.3) framing technique, see [octetcounting](/cotentcounting)
+- a parser which works on streams for syslog with [octet counting](https://tools.ietf.org/html/rfc5425#section-4.3) framing technique, see [octetcounting](/octetcounting)
 - a parser which works on streams for syslog with [non-transparent](https://tools.ietf.org/html/rfc6587#section-3.4.2) framing technique, see [nontransparent](/nontransparent)
 
 This library provides the pieces to parse Syslog messages transported following various RFCs.
@@ -199,7 +201,7 @@ To run the benchmark execute the following command.
 make bench
 ```
 
-On my machine<sup>[1](#mymachine)</sup> this are the results obtained paring RFC5424 syslog messages with best effort mode on.
+On my machine<sup>[1](#mymachine)</sup> these are the results obtained paring RFC5424 syslog messages with best effort mode on.
 
 ```
 [no]_empty_input__________________________________  4524100        274 ns/op      272 B/op        4 allocs/op
@@ -237,4 +239,4 @@ _TBD: comparison against other Go parsers_.
 
 ---
 
-* <a name="mymachine">[1]</a>: Intel Core i7-8850H CPU @ 2.60GHz
+* <a name="mymachine">[1]</a>: Intel Core i7-8850H CPU @ 2.60GHz

vendor/github.com/influxdata/go-syslog/v3/makefile

@@ -42,7 +42,7 @@ snake2camel:
 	}' $(file)
 
 .PHONY: bench
-bench: rfc5424/*_test.go rfc5424/machine.go
+bench: rfc5424/*_test.go rfc5424/machine.go octetcounting/performance_test.go
 	go test -bench=. -benchmem -benchtime=5s ./...
 
 .PHONY: tests

vendor/github.com/influxdata/go-syslog/v3/nontransparent/parser.go

@@ -218,6 +218,9 @@ func NewParser(options ...syslog.ParserOption) syslog.Parser {
 	return m
 }
 
+// WithMaxMessageLength does nothing for this parser
+func (m *machine) WithMaxMessageLength(length int) {}
+
 // HasBestEffort tells whether the receiving parser has best effort mode on or off.
 func (m *machine) HasBestEffort() bool {
 	return m.bestEffort

vendor/github.com/influxdata/go-syslog/v3/octetcounting/parser.go

@@ -12,19 +12,21 @@ import (
 //
 // Use NewParser function to instantiate one.
 type parser struct {
-	msglen     int64
-	s          Scanner
-	internal   syslog.Machine
-	last       Token
-	stepback   bool // Wheter to retrieve the last token or not
-	bestEffort bool // Best effort mode flag
-	emit       syslog.ParserListener
+	msglen           int64
+	maxMessageLength int
+	s                Scanner
+	internal         syslog.Machine
+	last             Token
+	stepback         bool // Wheter to retrieve the last token or not
+	bestEffort       bool // Best effort mode flag
+	emit             syslog.ParserListener
 }
 
 // NewParser returns a syslog.Parser suitable to parse syslog messages sent with transparent - ie. octet counting (RFC 5425) - framing.
 func NewParser(opts ...syslog.ParserOption) syslog.Parser {
 	p := &parser{
-		emit: func(*syslog.Result) { /* noop */ },
+		emit:             func(*syslog.Result) { /* noop */ },
+		maxMessageLength: 8192, // size as per RFC5425#section-4.3.1
 	}
 
 	for _, opt := range opts {
@@ -41,6 +43,10 @@ func NewParser(opts ...syslog.ParserOption) syslog.Parser {
 	return p
 }
 
+func (p *parser) WithMaxMessageLength(length int) {
+	p.maxMessageLength = length
+}
+
 // HasBestEffort tells whether the receiving parser has best effort mode on or off.
 func (p *parser) HasBestEffort() bool {
 	return p.bestEffort
@@ -64,7 +70,7 @@ func (p *parser) WithListener(f syslog.ParserListener) {
 //
 // It stops parsing when an error regarding RFC 5425 is found.
 func (p *parser) Parse(r io.Reader) {
-	p.s = *NewScanner(r)
+	p.s = *NewScanner(r, p.maxMessageLength)
 	p.run()
 }
 
@@ -80,6 +86,13 @@ func (p *parser) run() {
 			break
 		}
 
+		if int(p.s.msglen) > p.maxMessageLength {
+			p.emit(&syslog.Result{
+				Error: fmt.Errorf("message too long to parse. was size %d, max length %d", p.s.msglen, p.maxMessageLength),
+			})
+			break
+		}
+
 		// Next we MUST see a WS
 		if tok = p.scan(); tok.typ != WS {
 			p.emit(&syslog.Result{

vendor/github.com/influxdata/go-syslog/v3/octetcounting/scanner.go

@@ -7,9 +7,6 @@ import (
 	"strconv"
 )
 
-// size as per RFC5425#section-4.3.1
-var size = 8192
-
 // eof represents a marker byte for the end of the reader
 var eof = byte(0)
 
@@ -37,9 +34,9 @@ type Scanner struct {
 }
 
 // NewScanner returns a pointer to a new instance of Scanner.
-func NewScanner(r io.Reader) *Scanner {
+func NewScanner(r io.Reader, maxLength int) *Scanner {
 	return &Scanner{
-		r: bufio.NewReaderSize(r, size+5), // "8192 " has length 5
+		r: bufio.NewReaderSize(r, maxLength+20), // max uint64 is 19 characters + a space
 	}
 }
 
@@ -116,9 +113,6 @@ func (s *Scanner) scanMsgLen() Token {
 	msglen := buf.String()
 	s.msglen, _ = strconv.ParseUint(msglen, 10, 64)
 
-	// (todo) > return ILLEGAL if s.msglen > size (8192)
-	// (todo) > only when NOT in besteffort mode or always?
-
 	return Token{
 		typ: MSGLEN,
 		lit: buf.Bytes(),

vendor/github.com/influxdata/go-syslog/v3/options.go

@@ -17,3 +17,11 @@ func WithBestEffort() ParserOption {
 		return p
 	}
 }
+
+// WithMaxMessageLength sets the length of the buffer for octect parsing.
+func WithMaxMessageLength(length int) ParserOption {
+	return func(p Parser) Parser {
+		p.WithMaxMessageLength(length)
+		return p
+	}
+}

vendor/github.com/influxdata/go-syslog/v3/syslog.go

@@ -15,6 +15,11 @@ type BestEfforter interface {
 	HasBestEffort() bool
 }
 
+// MaxMessager sets the max message size the parser should be able to parse
+type MaxMessager interface {
+	WithMaxMessageLength(length int)
+}
+
 // Machine represent a FSM able to parse an entire syslog message and return it in an structured way.
 type Machine interface {
 	Parse(input []byte) (Message, error)
@@ -29,6 +34,7 @@ type Parser interface {
 	Parse(r io.Reader)
 	WithListener(ParserListener)
 	BestEfforter
+	MaxMessager
 }
 
 // ParserOption represent the type of option setters for Parser instances.

vendor/modules.txt

@@ -566,7 +566,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