Fix issue when clean_removed and clean_inactive were used together th…

…at states were not directly removed from the registry (#2750) (#2754)

The clean_inactive option was updating the TTL after clean_removed was applied. This lead to the issue that during every run clean_removed updated the state and clean_inactive reset the TTL. In the end removed files were never purged from the registry if both options were used together.

* Test added to verify fix
* Updated incorrect log message

Closes #2646

CHANGELOG.asciidoc

@@ -40,6 +40,7 @@ https://github.com/elastic/beats/compare/v5.0.0-rc1...5.0[Check the HEAD diff]
 *Topbeat*
 
 *Filebeat*
+- Fix issue when clean_removed and clean_inactive were used together that states were not directly removed from the registry.
 
 *Winlogbeat*
 

filebeat/prospector/config.go

@@ -39,11 +39,11 @@ func (config *prospectorConfig) Validate() error {
 	}
 
 	if config.CleanInactive != 0 && config.IgnoreOlder == 0 {
-		return fmt.Errorf("ignore_older must be enabled when clean_older is used.")
+		return fmt.Errorf("ignore_older must be enabled when clean_inactive is used.")
 	}
 
 	if config.CleanInactive != 0 && config.CleanInactive <= config.IgnoreOlder+config.ScanFrequency {
-		return fmt.Errorf("clean_older must be > ignore_older + scan_frequency to make sure only files which are not monitored anymore are removed.")
+		return fmt.Errorf("clean_inactive must be > ignore_older + scan_frequency to make sure only files which are not monitored anymore are removed.")
 	}
 
 	return nil

filebeat/prospector/prospector.go

@@ -144,8 +144,8 @@ func (p *Prospector) Run() {
 // All state updates done by the prospector itself are synchronous to make sure not states are overwritten
 func (p *Prospector) updateState(event *input.Event) error {
 
-	// Add ttl if cleanOlder is enabled
-	if p.config.CleanInactive > 0 {
+	// Add ttl if cleanOlder is enabled and TTL is not already 0
+	if p.config.CleanInactive > 0 && event.State.TTL != 0 {
 		event.State.TTL = p.config.CleanInactive
 	}
 

filebeat/tests/system/test_registrar.py

@@ -869,3 +869,70 @@ def test_clean_removed(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_clean_removed_with_clean_inactive(self):
+        """
+        Checks that files which were removed, the state is removed
+        """
+        self.render_config_template(
+            path=os.path.abspath(self.working_dir) + "/log/input*",
+            scan_frequency="0.1s",
+            clean_removed=True,
+            clean_inactive="20s",
+            ignore_older="15s",
+            close_removed=True
+        )
+
+        os.mkdir(self.working_dir + "/log/")
+        testfile1 = self.working_dir + "/log/input1"
+        testfile2 = self.working_dir + "/log/input2"
+
+        with open(testfile1, 'w') as f:
+            f.write("file to be removed\n")
+
+        with open(testfile2, 'w') as f:
+            f.write("2\n")
+
+        filebeat = self.start_beat()
+
+        self.wait_until(
+            lambda: self.output_has(lines=2),
+            max_timeout=10)
+
+        # Wait until registry file is created
+        self.wait_until(
+            lambda: self.log_contains_count("Registry file updated") > 0,
+            max_timeout=15)
+
+        data = self.get_registry()
+        assert len(data) == 2
+
+        os.remove(testfile1)
+
+        # Wait until states are removed from prospectors
+        self.wait_until(
+            lambda: self.log_contains(
+                "Remove state for file as file removed"),
+            max_timeout=15)
+
+        # Add one more line to make sure registry is written
+        with open(testfile2, 'a') as f:
+            f.write("make sure registry is written\n")
+
+        self.wait_until(
+            lambda: self.output_has(lines=3),
+            max_timeout=10)
+
+        time.sleep(3)
+
+        filebeat.check_kill_and_wait()
+
+        # Check that the first to files were removed from the registry
+        data = self.get_registry()
+        assert len(data) == 1
+
+        # Make sure the last file in the registry is the correct one and has the correct offset
+        if os.name == "nt":
+            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")