Modify clean_removed handling

CHANGELOG.asciidoc

@@ -25,6 +25,7 @@ https://github.com/elastic/beats/compare/v5.1.1...master[Check the HEAD diff]
 - Always use absolute path for event and registry. This can lead to issues when relative paths were used before. {pull}3328[3328]
 - Remove code to convert states from 1.x. {pull}3767[3767]
 - Remove deprecated config options force_close_files and close_older. {pull}3768[3768]
+- Change clean_removed behaviour to also remove states for files which cannot be found anymore under the same name. {pull}3827[3827]
 
 *Heartbeat*
 

filebeat/docs/reference/configuration/filebeat-options.asciidoc

@@ -273,14 +273,12 @@ NOTE: Every time a file is renamed, the file state is updated and the counter fo
 [[clean-removed]]
 ===== clean_removed
 
-When this option is enabled, Filebeat cleans files from the registry if they cannot be found on disk anymore. This setting does not apply to renamed files or files that were moved to another directory that is still visible to Filebeat. This option is enabled by default.
-
+When this option is enabled, Filebeat cleans files from the registry if they cannot be found on disk anymore under the last known name. This means also files which were renamed after the harvester was finished will be removed. This option is enabled by default.
 
 If a shared drive disappears for a short period and appears again, all files will be read again from the beginning because the states were removed from the registry file. In such cases, we recommend that you disable the `clean_removed` option.
 
 You must disable this option if you also disable `close_removed`.
 
-
 [[scan-frequency]]
 ===== scan_frequency
 

filebeat/prospector/prospector_log.go

@@ -94,24 +94,42 @@ func (l *Log) Run() {
 	if l.config.CleanRemoved {
 		for _, state := range l.Prospector.states.GetStates() {
 			// os.Stat will return an error in case the file does not exist
-			_, err := os.Stat(state.Source)
+			stat, err := os.Stat(state.Source)
 			if err != nil {
-				// Only clean up files where state is Finished
-				if state.Finished {
-					state.TTL = 0
-					err := l.Prospector.updateState(input.NewEvent(state))
-					if err != nil {
-						logp.Err("File cleanup state update error: %s", err)
-					}
+				if os.IsNotExist(err) {
+					l.removeState(state)
 					logp.Debug("prospector", "Remove state for file as file removed: %s", state.Source)
 				} else {
-					logp.Debug("prospector", "State for file not removed because not finished: %s", state.Source)
+					logp.Err("Prospector state for %s was not removed: %s", state.Source, err)
+				}
+			} else {
+				// Check if existing source on disk and state are the same. Remove if not the case.
+				newState := file.NewState(stat, state.Source)
+				if !newState.FileStateOS.IsSame(state.FileStateOS) {
+					l.removeState(state)
+					logp.Debug("prospector", "Remove state for file as file removed or renamed: %s", state.Source)
 				}
 			}
 		}
 	}
 }
 
+func (l *Log) removeState(state file.State) {
+
+	// Only clean up files where state is Finished
+	if !state.Finished {
+		logp.Debug("prospector", "State for file not removed because harvester not finished: %s", state.Source)
+		return
+	}
+
+	state.TTL = 0
+	err := l.Prospector.updateState(input.NewEvent(state))
+	if err != nil {
+		logp.Err("File cleanup state update error: %s", err)
+	}
+
+}
+
 // getFiles returns all files which have to be harvested
 // All globs are expanded and then directory and excluded files are removed
 func (l *Log) getFiles() map[string]os.FileInfo {

filebeat/tests/system/test_harvester.py

@@ -20,6 +20,7 @@ def test_close_renamed(self):
         self.render_config_template(
             path=os.path.abspath(self.working_dir) + "/log/test.log",
             close_renamed="true",
+            clean_removed="false",
             scan_frequency="0.1s"
         )
         os.mkdir(self.working_dir + "/log/")
@@ -620,7 +621,8 @@ def test_symlink_removed(self):
         self.render_config_template(
             path=os.path.abspath(self.working_dir) + "/log/symlink.log",
             symlinks="true",
-            clean_removed="false"
+            clean_removed="false",
+            close_removed="false",
         )
 
         os.mkdir(self.working_dir + "/log/")

filebeat/tests/system/test_registrar.py

@@ -396,7 +396,8 @@ def test_rotating_file_with_restart(self):
         self.render_config_template(
             path=os.path.abspath(self.working_dir) + "/log/input*",
             scan_frequency="1s",
-            close_inactive="1s"
+            close_inactive="1s",
+            clean_removed="false"
         )
 
         if os.name == "nt":