Update go-syslog dependency to accept any encoding in message (#2127)

go.mod

@@ -31,7 +31,7 @@ require (
 	github.com/grpc-ecosystem/grpc-opentracing v0.0.0-20180507213350-8e809c8a8645
 	github.com/hashicorp/golang-lru v0.5.3
 	github.com/hpcloud/tail v1.0.0
-	github.com/influxdata/go-syslog/v2 v2.0.1
+	github.com/influxdata/go-syslog/v3 v3.0.1-0.20200510134747-836dce2cf6da
 	github.com/jmespath/go-jmespath v0.0.0-20180206201540-c2b33e8439af
 	github.com/json-iterator/go v1.1.9
 	github.com/klauspost/compress v1.9.4

go.sum

@@ -527,8 +527,8 @@ github.com/hodgesds/perf-utils v0.0.8/go.mod h1:F6TfvsbtrF88i++hou29dTXlI2sfsJv+
 github.com/hudl/fargo v1.3.0/go.mod h1:y3CKSmjA+wD2gak7sUSXTAoopbhU08POFhmITJgmKTg=
 github.com/imdario/mergo v0.3.5/go.mod h1:2EnlNZ0deacrJVfApfmtdGgDfMuh/nq6Ok1EcJh5FfA=
 github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
-github.com/influxdata/go-syslog/v2 v2.0.1 h1:l44S4l4Q8MhGQcoOxJpbo+QQYxJqp0vdgIVHh4+DO0s=
-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/influxdb v1.7.7/go.mod h1:qZna6X/4elxqT3yI9iZYdZrWWdeFOOprn86kgg4+IzY=
 github.com/influxdata/influxdb1-client v0.0.0-20191209144304-8bf82d3c094d/go.mod h1:qj24IKcXYK6Iy9ceXlo3Tc+vtHo9lIhSX5JddghvEPo=
 github.com/jackc/fake v0.0.0-20150926172116-812a484cc733/go.mod h1:WrMFNQdiFJ80sQsxDoMokWK1W5TQtxBFNpzWTD84ibQ=

pkg/promtail/targets/syslogparser/syslogparser.go

@@ -5,9 +5,9 @@ import (
 	"fmt"
 	"io"
 
-	"github.com/influxdata/go-syslog/v2"
-	"github.com/influxdata/go-syslog/v2/nontransparent"
-	"github.com/influxdata/go-syslog/v2/octetcounting"
+	"github.com/influxdata/go-syslog/v3"
+	"github.com/influxdata/go-syslog/v3/nontransparent"
+	"github.com/influxdata/go-syslog/v3/octetcounting"
 )
 
 // ParseStream parses a rfc5424 syslog stream from the given Reader, calling

pkg/promtail/targets/syslogparser/syslogparser_test.go

@@ -5,7 +5,8 @@ import (
 	"strings"
 	"testing"
 
-	"github.com/influxdata/go-syslog/v2"
+	"github.com/influxdata/go-syslog/v3"
+	"github.com/influxdata/go-syslog/v3/rfc5424"
 	"github.com/stretchr/testify/require"
 
 	"github.com/grafana/loki/pkg/promtail/targets/syslogparser"
@@ -24,9 +25,9 @@ func TestParseStream_OctetCounting(t *testing.T) {
 
 	require.Equal(t, 2, len(results))
 	require.NoError(t, results[0].Error)
-	require.Equal(t, "First", *results[0].Message.Message())
+	require.Equal(t, "First", *results[0].Message.(*rfc5424.SyslogMessage).Message)
 	require.NoError(t, results[1].Error)
-	require.Equal(t, "Second", *results[1].Message.Message())
+	require.Equal(t, "Second", *results[1].Message.(*rfc5424.SyslogMessage).Message)
 }
 
 func TestParseStream_NewlineSeparated(t *testing.T) {
@@ -42,9 +43,9 @@ func TestParseStream_NewlineSeparated(t *testing.T) {
 
 	require.Equal(t, 2, len(results))
 	require.NoError(t, results[0].Error)
-	require.Equal(t, "First", *results[0].Message.Message())
+	require.Equal(t, "First", *results[0].Message.(*rfc5424.SyslogMessage).Message)
 	require.NoError(t, results[1].Error)
-	require.Equal(t, "Second", *results[1].Message.Message())
+	require.Equal(t, "Second", *results[1].Message.(*rfc5424.SyslogMessage).Message)
 }
 
 func TestParseStream_InvalidStream(t *testing.T) {

pkg/promtail/targets/syslogtarget.go

@@ -12,7 +12,8 @@ import (
 	"github.com/cortexproject/cortex/pkg/util"
 	"github.com/go-kit/kit/log"
 	"github.com/go-kit/kit/log/level"
-	"github.com/influxdata/go-syslog/v2"
+	"github.com/influxdata/go-syslog/v3"
+	"github.com/influxdata/go-syslog/v3/rfc5424"
 	"github.com/mwitkow/go-conntrack"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
@@ -176,32 +177,34 @@ func (t *SyslogTarget) handleMessageError(err error) {
 }
 
 func (t *SyslogTarget) handleMessage(connLabels labels.Labels, msg syslog.Message) {
-	if msg.Message() == nil {
+	rfc5424Msg := msg.(*rfc5424.SyslogMessage)
+
+	if rfc5424Msg.Message == nil {
 		return
 	}
 
 	lb := labels.NewBuilder(connLabels)
-	if v := msg.SeverityLevel(); v != nil {
+	if v := rfc5424Msg.SeverityLevel(); v != nil {
 		lb.Set("__syslog_message_severity", *v)
 	}
-	if v := msg.FacilityLevel(); v != nil {
+	if v := rfc5424Msg.FacilityLevel(); v != nil {
 		lb.Set("__syslog_message_facility", *v)
 	}
-	if v := msg.Hostname(); v != nil {
+	if v := rfc5424Msg.Hostname; v != nil {
 		lb.Set("__syslog_message_hostname", *v)
 	}
-	if v := msg.Appname(); v != nil {
+	if v := rfc5424Msg.Appname; v != nil {
 		lb.Set("__syslog_message_app_name", *v)
 	}
-	if v := msg.ProcID(); v != nil {
+	if v := rfc5424Msg.ProcID; v != nil {
 		lb.Set("__syslog_message_proc_id", *v)
 	}
-	if v := msg.MsgID(); v != nil {
+	if v := rfc5424Msg.MsgID; v != nil {
 		lb.Set("__syslog_message_msg_id", *v)
 	}
 
-	if t.config.LabelStructuredData && msg.StructuredData() != nil {
-		for id, params := range *msg.StructuredData() {
+	if t.config.LabelStructuredData && rfc5424Msg.StructuredData != nil {
+		for id, params := range *rfc5424Msg.StructuredData {
 			id = strings.Replace(id, "@", "_", -1)
 			for name, value := range params {
 				key := "__syslog_message_sd_" + id + "_" + name
@@ -220,7 +223,7 @@ func (t *SyslogTarget) handleMessage(connLabels labels.Labels, msg syslog.Messag
 		filtered[model.LabelName(lbl.Name)] = model.LabelValue(lbl.Value)
 	}
 
-	t.messages <- message{filtered, *msg.Message()}
+	t.messages <- message{filtered, *rfc5424Msg.Message}
 }
 
 func (t *SyslogTarget) messageSender() {

pkg/promtail/targets/syslogtarget_test.go

@@ -8,6 +8,7 @@ import (
 	"sync"
 	"testing"
 	"time"
+	"unicode/utf8"
 
 	"github.com/go-kit/kit/log"
 	"github.com/go-kit/kit/log/level"
@@ -185,6 +186,43 @@ func TestSyslogTarget_InvalidData(t *testing.T) {
 	require.EqualError(t, err, "EOF")
 }
 
+func TestSyslogTarget_NonUTF8Message(t *testing.T) {
+	w := log.NewSyncWriter(os.Stderr)
+	logger := log.NewLogfmtLogger(w)
+	client := &TestLabeledClient{log: logger}
+
+	tgt, err := NewSyslogTarget(logger, client, relabelConfig(t), &scrape.SyslogTargetConfig{
+		ListenAddress: "127.0.0.1:0",
+	})
+	require.NoError(t, err)
+	defer func() {
+		require.NoError(t, tgt.Stop())
+	}()
+
+	addr := tgt.ListenAddress().String()
+	c, err := net.Dial("tcp", addr)
+	require.NoError(t, err)
+
+	msg1 := "Some non utf8 \xF8\xF7\xE3\xE4 characters"
+	require.False(t, utf8.ValidString(msg1), "msg must no be valid utf8")
+	msg2 := "\xF8 other \xF7\xE3\xE4 characters \xE3"
+	require.False(t, utf8.ValidString(msg2), "msg must no be valid utf8")
+
+	err = writeMessagesToStream(c, []string{
+		"<165>1 - - - - - - " + msg1,
+		"<123>1 - - - - - - " + msg2,
+	}, true)
+	require.NoError(t, err)
+	require.NoError(t, c.Close())
+
+	require.Eventuallyf(t, func() bool {
+		return len(client.Messages()) == 2
+	}, time.Second, time.Millisecond, "Expected to receive 2 messages, got %d.", len(client.Messages()))
+
+	require.Equal(t, msg1, client.Messages()[0].Message)
+	require.Equal(t, msg2, client.Messages()[1].Message)
+}
+
 func TestSyslogTarget_IdleTimeout(t *testing.T) {
 	w := log.NewSyncWriter(os.Stderr)
 	logger := log.NewLogfmtLogger(w)