fix issue with stream parser blocking when data is in buffer (#7631)

plugins/parsers/influx/machine.go

@@ -31678,6 +31678,16 @@ func (m *streamMachine) Next() error {
 			m.machine.data = expanded
 		}
 
+		err := m.machine.exec()
+		if err != nil {
+			return err
+		}
+
+		// If we have successfully parsed a full metric line break out
+		if m.machine.finishMetric {
+			break
+		}
+
 		n, err := m.reader.Read(m.machine.data[m.machine.pe:])
 		if n == 0 && err == io.EOF {
 			m.machine.eof = m.machine.pe
@@ -31692,16 +31702,6 @@ func (m *streamMachine) Next() error {
 
 		m.machine.pe += n
 
-		err = m.machine.exec()
-		if err != nil {
-			return err
-		}
-
-		// If we have successfully parsed a full metric line break out
-		if m.machine.finishMetric {
-			break
-		}
-
 	}
 
 	return nil

plugins/parsers/influx/machine.go.rl

@@ -506,6 +506,16 @@ func (m *streamMachine) Next() error {
 			m.machine.data = expanded
 		}
 
+		err := m.machine.exec()
+		if err != nil {
+			return err
+		}
+
+		// If we have successfully parsed a full metric line break out
+		if m.machine.finishMetric {
+			break
+		}
+
 		n, err := m.reader.Read(m.machine.data[m.machine.pe:])
 		if n == 0 && err == io.EOF {
 			m.machine.eof = m.machine.pe
@@ -520,16 +530,6 @@ func (m *streamMachine) Next() error {
 
 		m.machine.pe += n
 
-		err = m.machine.exec()
-		if err != nil {
-			return err
-		}
-
-		// If we have successfully parsed a full metric line break out
-		if m.machine.finishMetric {
-			break
-		}
-
 	}
 
 	return nil

plugins/parsers/influx/parser_test.go

@@ -3,6 +3,7 @@ package influx
 import (
 	"bytes"
 	"errors"
+	"io"
 	"strconv"
 	"strings"
 	"testing"
@@ -895,3 +896,19 @@ func TestStreamParserReaderError(t *testing.T) {
 	_, err = parser.Next()
 	require.Equal(t, err, EOF)
 }
+
+func TestStreamParserProducesAllAvailableMetrics(t *testing.T) {
+	r, w := io.Pipe()
+
+	parser := NewStreamParser(r)
+	parser.SetTimeFunc(DefaultTime)
+
+	go w.Write([]byte("metric value=1\nmetric2 value=1\n"))
+
+	_, err := parser.Next()
+	require.NoError(t, err)
+
+	// should not block on second read
+	_, err = parser.Next()
+	require.NoError(t, err)
+}