Handle data returned with io.EOF in LineReader (#26260)

andrewkroh · web-flow · commit 6896e7798551 · 2021-06-14T13:07:12.000-04:00
The libbeat LineReader implementation did not handle the case where the underlying io.Reader
it was reading from returns bytes and io.EOF. It was discarding the data in this case.

As per the io.Reader contract:

    a Reader returning a non-zero number of bytes at the end of the input stream may return either err == EOF or err == nil.

This occurs often with the gzip.Reader. It returns a large chunk of data at the end of the file and io.EOF at the same time.
diff --git a/CHANGELOG.next.asciidoc b/CHANGELOG.next.asciidoc
@@ -282,6 +282,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Fix CredentialsJSON unpacking for `gcp-pubsub` and `httpjson` inputs. {pull}23277[23277]
 - Fix issue with m365_defender, when parsing incidents that has no alerts attached: {pull}25421[25421]
 - Fix default config template values for paths on oracle module: {pull}26276[26276]
+- Fix bug in aws-s3 input where the end of gzipped log files might have been discarded. {pull}26260[26260]
 
 *Filebeat*
 
diff --git a/libbeat/reader/readfile/line.go b/libbeat/reader/readfile/line.go
@@ -138,6 +138,11 @@ func (r *LineReader) advance() error {
 		// Try to read more bytes into buffer
 		n, err := r.reader.Read(buf)
 
+		if err == io.EOF && n > 0 {
+			// Continue processing the returned bytes. The next call will yield EOF with 0 bytes.
+			err = nil
+		}
+
 		// Appends buffer also in case of err
 		r.inBuffer.Append(buf[:n])
 		if err != nil {
diff --git a/libbeat/reader/readfile/line_test.go b/libbeat/reader/readfile/line_test.go
@@ -219,19 +219,25 @@ func testReadLineLengths(t *testing.T, lineLengths []int) {
 		lines = append(lines, inputLine)
 	}
 
-	testReadLines(t, lines)
+	testReadLines(t, lines, false)
 }
 
-func testReadLines(t *testing.T, inputLines [][]byte) {
+func testReadLines(t *testing.T, inputLines [][]byte, eofOnLastRead bool) {
 	var inputStream []byte
 	for _, line := range inputLines {
 		inputStream = append(inputStream, line...)
 	}
 
 	// initialize reader
 	buffer := bytes.NewBuffer(inputStream)
-	codec, _ := encoding.Plain(buffer)
-	reader, err := NewLineReader(ioutil.NopCloser(buffer), Config{codec, buffer.Len(), LineFeed, unlimited})
+
+	var r io.Reader = buffer
+	if eofOnLastRead {
+		r = &eofWithNonZeroNumberOfBytesReader{buf: buffer}
+	}
+
+	codec, _ := encoding.Plain(r)
+	reader, err := NewLineReader(ioutil.NopCloser(r), Config{codec, buffer.Len(), LineFeed, unlimited})
 	if err != nil {
 		t.Fatalf("Error initializing reader: %v", err)
 	}
@@ -255,7 +261,7 @@ func testReadLines(t *testing.T, inputLines [][]byte) {
 }
 
 func testReadLine(t *testing.T, line []byte) {
-	testReadLines(t, [][]byte{line})
+	testReadLines(t, [][]byte{line}, false)
 }
 
 func randomInt(r *rand.Rand, min, max int) int {
@@ -425,3 +431,31 @@ func TestBufferSize(t *testing.T) {
 		require.Equal(t, string(b[:n]), lines[i])
 	}
 }
+
+// eofWithNonZeroNumberOfBytesReader is an io.Reader implementation that at the
+// end of the stream returns a non-zero number of bytes with io.EOF. This is
+// allowed under the io.Reader interface contract and must be handled by the
+// line reader.
+type eofWithNonZeroNumberOfBytesReader struct {
+	buf *bytes.Buffer
+}
+
+func (r *eofWithNonZeroNumberOfBytesReader) Read(d []byte) (int, error) {
+	n, err := r.buf.Read(d)
+	if err != nil {
+		return n, err
+	}
+
+	// As per the io.Reader contract:
+	//   "a Reader returning a non-zero number of bytes at the end of the input
+	//   stream may return either err == EOF or err == nil."
+	if r.buf.Len() == 0 {
+		return n, io.EOF
+	}
+	return n, nil
+}
+
+// Verify handling of the io.Reader returning n > 0 with io.EOF.
+func TestReadWithNonZeroNumberOfBytesAndEOF(t *testing.T) {
+	testReadLines(t, [][]byte{[]byte("Hello world!\n")}, true)
+}
