fix syslog parser data race problem (#32)

* fix syslog parser data race

.gitignore

@@ -7,3 +7,4 @@ local/*
 artifacts/*
 .vscode/*
 gen/
+.idea/*

operator/builtin/input/syslog/syslog_test.go

@@ -65,16 +65,15 @@ func SyslogInputTest(t *testing.T, cfg *SyslogInputConfig, tc syslog.Case) {
 	conn.Close()
 	require.NoError(t, err)
 
+	defer p.Stop()
 	select {
 	case e := <-fake.Received:
 		// close pipeline to avoid data race
-		p.Stop()
 		require.Equal(t, tc.ExpectedRecord, e.Record)
 		require.Equal(t, tc.ExpectedTimestamp, e.Timestamp)
 		require.Equal(t, tc.ExpectedSeverity, e.Severity)
 		require.Equal(t, tc.ExpectedSeverityText, e.SeverityText)
 	case <-time.After(time.Second):
-		p.Stop()
 		require.FailNow(t, "Timed out waiting for entry to be processed")
 	}
 }

operator/builtin/parser/syslog/syslog.go

@@ -103,10 +103,7 @@ type SyslogParser struct {
 
 // Process will parse an entry field as syslog.
 func (s *SyslogParser) Process(ctx context.Context, entry *entry.Entry) error {
-	if err := s.ParserOperator.ProcessWith(ctx, entry, s.parse); err != nil {
-		return err
-	}
-	return promoteSeverity(entry)
+	return s.ParserOperator.ProcessWithCallback(ctx, entry, s.parse, promoteSeverity)
 }
 
 // parse will parse a value as syslog.

operator/helper/parser.go

@@ -87,6 +87,10 @@ type ParserOperator struct {
 
 // ProcessWith will run ParseWith on the entry, then forward the entry on to the next operators.
 func (p *ParserOperator) ProcessWith(ctx context.Context, entry *entry.Entry, parse ParseFunction) error {
+	return p.ProcessWithCallback(ctx, entry, parse, nil)
+}
+
+func (p *ParserOperator) ProcessWithCallback(ctx context.Context, entry *entry.Entry, parse ParseFunction, cb func(*entry.Entry) error) error {
 	// Short circuit if the "if" condition does not match
 	skip, err := p.Skip(ctx, entry)
 	if err != nil {
@@ -100,6 +104,13 @@ func (p *ParserOperator) ProcessWith(ctx context.Context, entry *entry.Entry, pa
 	if err := p.ParseWith(ctx, entry, parse); err != nil {
 		return err
 	}
+	if cb != nil {
+		err = cb(entry)
+		if err != nil {
+			return err
+		}
+	}
+
 	p.Write(ctx, entry)
 	return nil
 }