Fix state loading in case a state falls under ignore_older after restart

CHANGELOG.asciidoc

@@ -36,6 +36,7 @@ https://github.com/elastic/beats/compare/v5.0.0-rc1...5.0[Check the HEAD diff]
 *Topbeat*
 
 *Filebeat*
+- Fix registry cleanup issue when files falling under ignore_older after restart. {issue}2818[2818]
 
 *Winlogbeat*
 

filebeat/prospector/prospector_log.go

@@ -41,13 +41,19 @@ func (p *ProspectorLog) Init() {
 	fileStates := p.Prospector.states.GetStates()
 
 	// Make sure all states are set as finished
-	for key, state := range fileStates {
+	for _, state := range fileStates {
 		state.Finished = true
-		fileStates[key] = state
+		// Set all states again to infinity TTL to make sure only removed if config still same
+		// clean_inactive / clean_removed could have been changed between restarts
+		state.TTL = -1
+
+		// Update prospector states and send new states to registry
+		err := p.Prospector.updateState(input.NewEvent(state))
+		if err != nil {
+			logp.Err("Problem putting initial state: %+v", err)
+		}
 	}
 
-	// Overwrite prospector states
-	p.Prospector.states.SetStates(fileStates)
 	p.lastClean = time.Now()
 
 	logp.Info("Previous states loaded: %v", p.Prospector.states.Count())
@@ -74,8 +80,7 @@ func (p *ProspectorLog) Run() {
 				// Only clean up files where state is Finished
 				if state.Finished {
 					state.TTL = 0
-					event := input.NewEvent(state)
-					err := p.Prospector.updateState(event)
+					err := p.Prospector.updateState(input.NewEvent(state))
 					if err != nil {
 						logp.Err("File cleanup state update error: %s", err)
 					}
@@ -234,8 +239,7 @@ func (p *ProspectorLog) harvestExistingFile(newState file.State, oldState file.S
 			logp.Debug("prospector", "Updating state for renamed file: %s -> %s, Current offset: %v", oldState.Source, newState.Source, oldState.Offset)
 			// Update state because of file rotation
 			oldState.Source = newState.Source
-			event := input.NewEvent(oldState)
-			err := p.Prospector.updateState(event)
+			err := p.Prospector.updateState(input.NewEvent(oldState))
 			if err != nil {
 				logp.Err("File rotation state update error: %s", err)
 			}

filebeat/tests/system/test_registrar.py

@@ -640,6 +640,8 @@ def test_migration_non_windows(self):
 
         self.render_config_template(
             path=os.path.abspath(self.working_dir) + "/log/input*",
+            clean_removed="false",
+            clean_inactive="0",
         )
 
         filebeat = self.start_beat()
@@ -936,3 +938,55 @@ def test_clean_removed_with_clean_inactive(self):
             assert data[0]["offset"] == len("make sure registry is written\n" + "2\n") + 2
         else:
             assert data[0]["offset"] == len("make sure registry is written\n" + "2\n")
+
+    def test_restart_state(self):
+        """
+        Make sure that states are rewritten correctly on restart and cleaned
+        """
+
+        self.render_config_template(
+            path=os.path.abspath(self.working_dir) + "/log/*",
+            close_inactive="1s",
+            ignore_older="3s",
+            clean_inactive="5s",
+        )
+        os.mkdir(self.working_dir + "/log/")
+
+        testfile1 = self.working_dir + "/log/test1.log"
+        testfile2 = self.working_dir + "/log/test2.log"
+        testfile3 = self.working_dir + "/log/test3.log"
+        testfile4 = self.working_dir + "/log/test4.log"
+
+        with open(testfile1, 'w') as file:
+            file.write("Hello World\n")
+        with open(testfile2, 'w') as file:
+            file.write("Hello World\n")
+        with open(testfile3, 'w') as file:
+            file.write("Hello World\n")
+
+        filebeat = self.start_beat()
+
+        # Make sure states written appears one more time
+        self.wait_until(
+            lambda: self.log_contains("Ignore file because ignore_older"),
+            max_timeout=10)
+
+        filebeat.check_kill_and_wait()
+
+        filebeat = self.start_beat(output="filebeat2.log")
+
+        # Write additional file
+        with open(testfile4, 'w') as file:
+            file.write("Hello World\n")
+
+        # Make sure all 4 states are persisted
+        self.wait_until(
+            lambda: self.log_contains("Before: 4, After: 4", logfile="filebeat2.log"),
+            max_timeout=10)
+
+        # Wait until registry file is cleaned
+        self.wait_until(
+            lambda: self.log_contains("Before: 0, After: 0", logfile="filebeat2.log"),
+            max_timeout=10)
+
+        filebeat.check_kill_and_wait()