Do not close filestream harvester if an unexpected error is returned …

…when close.on_state_change.* is enabled (#26411) (#26476)

## What does this PR do?

This PR returns early if `close.on_state_change.removed` is enabled and the opened file no longer exists. Otherwise, it logs an error message and keeps the reader running.

## Why is it important?

Previously, a message has been logged on error level and the reader has been stopped if the `Stat` call returned an error. However, it was not correct because if `close.on_state_change.renamed` was enabled the reader would have been closed if the file had been removed. Now the reader is not stopped.

(cherry picked from commit 1106449)

Co-authored-by: Noémi Ványi <kvch@users.noreply.github.com>

CHANGELOG.next.asciidoc

@@ -218,6 +218,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Fix bug in aws-s3 input where the end of gzipped log files might have been discarded. {pull}26260[26260]
 - Fix bug in `httpjson` that prevented `first_event` getting updated. {pull}26407[26407]
 - Fix bug in the Syslog input that misparsed rfc5424 days starting with 0. {pull}26419[26419]
+- Do not close filestream harvester if an unexpected error is returned when close.on_state_change.* is enabled. {pull}26411[26411] 
 
 *Filebeat*
 

filebeat/input/filestream/filestream.go

@@ -179,22 +179,29 @@ func (f *logFile) shouldBeClosed() bool {
 
 	info, statErr := f.file.Stat()
 	if statErr != nil {
+		// return early if the file does not exist anymore and the reader should be closed
+		if f.closeRemoved && errors.Is(statErr, os.ErrNotExist) {
+			f.log.Debugf("close.on_state_change.removed is enabled and file %s has been removed", f.file.Name())
+			return true
+		}
+
+		// If an unexpected error happens we keep the reader open hoping once everything will go back to normal.
 		f.log.Errorf("Unexpected error reading from %s; error: %s", f.file.Name(), statErr)
-		return true
+		return false
 	}
 
 	if f.closeRenamed {
 		// Check if the file can still be found under the same path
 		if !isSameFile(f.file.Name(), info) {
-			f.log.Debugf("close_renamed is enabled and file %s has been renamed", f.file.Name())
+			f.log.Debugf("close.on_state_change.renamed is enabled and file %s has been renamed", f.file.Name())
 			return true
 		}
 	}
 
 	if f.closeRemoved {
 		// Check if the file name exists. See https://github.com/elastic/filebeat/issues/93
 		if file.IsRemoved(f.file) {
-			f.log.Debugf("close_removed is enabled and file %s has been removed", f.file.Name())
+			f.log.Debugf("close.on_state_change.removed is enabled and file %s has been removed", f.file.Name())
 			return true
 		}
 	}