Fixes metricbeat system test flakyness with better expectations (elas…

…tic#8052)

This patch changes the expectations from looking at output and pending stop operations to looking at the number of modules currently running. Publishing to a pipeline is async, so a module can receive a stop send a message, then actually stop. This should make things more robust.

metricbeat/tests/system/test_reload.py

@@ -44,6 +44,8 @@ def test_reload(self):
 
     @unittest.skipUnless(re.match("(?i)win|linux|darwin|freebsd|openbsd", sys.platform), "os")
     def test_start_stop(self):
+        def reload_line(
+            num_runners): return "Starting reload procedure, current runners: %d" % num_runners
         """
         Test if module is properly started and stopped
         """
@@ -57,6 +59,11 @@ def test_start_stop(self):
         config_path = self.working_dir + "/configs/system.yml"
         proc = self.start_beat()
 
+        # Ensure no modules are loaded
+        self.wait_until(
+            lambda: self.log_contains(reload_line(0)),
+            max_timeout=10)
+
         systemConfig = """
 - module: system
   metricsets: ["cpu"]
@@ -66,28 +73,21 @@ def test_start_stop(self):
         with open(config_path, 'w') as f:
             f.write(systemConfig)
 
-        # Wait until offset for new line is updated
+        # Ensure the module was successfully loaded
         self.wait_until(
-            lambda: self.log_contains("Starting runner: system [metricsets=1]"),
+            lambda: self.log_contains(reload_line(1)),
             max_timeout=10)
 
-        self.wait_until(lambda: self.output_lines() > 0)
-
         # Remove config again
         os.remove(config_path)
 
-        # Wait until offset for new line is updated
+        # Ensure the module was successfully unloaded
         self.wait_until(
-            lambda: self.log_contains("Stopping runner: system [metricsets=1]"),
+            lambda: self.log_contains(reload_line(0)),
             max_timeout=10)
 
-        lines = self.output_lines()
-
         time.sleep(1)
 
-        # Make sure no new lines were added since stopping
-        assert lines == self.output_lines()
-
         proc.check_kill_and_wait()
 
     @unittest.skipUnless(re.match("(?i)win|linux|darwin|freebsd|openbsd", sys.platform), "os")