Enable and fix Journalbeat integration tests (elastic#20140)

(cherry picked from commit 2025662)

journalbeat/Makefile

@@ -1,6 +1,6 @@
 BEAT_NAME=journalbeat
 BEAT_TITLE=Journalbeat
-SYSTEM_TESTS=false
+SYSTEM_TESTS=true
 TEST_ENVIRONMENT=false
 ES_BEATS?=..
 

journalbeat/input/input.go

@@ -79,7 +79,7 @@ func New(
 		state := states[cfg.CheckpointID]
 		r, err := reader.NewLocal(cfg, done, state, logger)
 		if err != nil {
-			return nil, fmt.Errorf("error creating reader for local journal: %v", err)
+			return nil, fmt.Errorf("error creating reader for local journal: %+v", err)
 		}
 		readers = append(readers, r)
 	}
@@ -99,7 +99,7 @@ func New(
 		state := states[cfg.CheckpointID]
 		r, err := reader.New(cfg, done, state, logger)
 		if err != nil {
-			return nil, fmt.Errorf("error creating reader for journal: %v", err)
+			return nil, fmt.Errorf("error creating reader for journal: %+v", err)
 		}
 		readers = append(readers, r)
 	}

journalbeat/tests/system/input/test.registry

@@ -1,6 +1,6 @@
 update_time: 2018-09-11T10:06:50.895829905Z
 journal_entries:
 - path: /home/n/go/src/github.com/elastic/beats/journalbeat/tests/system/input/test.journal
-  cursor: s=7d22fd7aa0c7482d88c303f47d5f32dc;i=2fcb;b=902dc834f07d4f41ade064f6b2ef8b4f;m=1bf0ff5c6d;t=55913a25fe765;x=c7e6480eec30822b
-  realtime_timestamp: 1505315746998117
-  monotonic_timestamp: 120007384173
+  cursor: s=018329e08e3a45a0ae03694421c4f553;i=2015d;b=fa3c2e3080dc4cd5be5cb5a43e140d51;m=29102136a4;t=5ab0792b1dc62;x=84a1467480b8f1af
+  realtime_timestamp: 1595423897803874
+  monotonic_timestamp: 176364271268

journalbeat/tests/system/test_base.py

@@ -44,19 +44,14 @@ def test_start_with_journal_directory(self):
         )
         journalbeat_proc = self.start_beat()
 
-        required_log_snippets = [
-            # journalbeat can be started
-            "journalbeat is running",
-            # journalbeat can seek to the position defined in the cursor
-            "Tailing the journal file",
-        ]
-        for snippet in required_log_snippets:
-            self.wait_until(lambda: self.log_contains(snippet),
-                            name="Line in '{}' Journalbeat log".format(snippet))
+        self.wait_until(lambda: self.log_contains("journalbeat is running"))
 
         exit_code = journalbeat_proc.kill_and_wait()
         assert exit_code == 0
 
+        # journalbeat is tailing an inactive journal
+        assert self.output_is_empty()
+
     @unittest.skipUnless(sys.platform.startswith("linux"), "Journald only on Linux")
     def test_start_with_selected_journal_file(self):
         """
@@ -73,17 +68,7 @@ def test_start_with_selected_journal_file(self):
         )
         journalbeat_proc = self.start_beat()
 
-        required_log_snippets = [
-            # journalbeat can be started
-            "journalbeat is running",
-            # journalbeat can seek to the position defined in the cursor
-            "Reading from the beginning of the journal file",
-            # message can be read from test journal
-            "\"message\": \"thinkpad_acpi: unhandled HKEY event 0x60b0\"",
-        ]
-        for snippet in required_log_snippets:
-            self.wait_until(lambda: self.log_contains(snippet),
-                            name="Line in '{}' Journalbeat log".format(snippet))
+        self.wait_until(lambda: self.output_has(lines=23))
 
         exit_code = journalbeat_proc.kill_and_wait()
         assert exit_code == 0
@@ -105,21 +90,14 @@ def test_start_with_selected_journal_file_with_cursor_fallback(self):
         )
         journalbeat_proc = self.start_beat()
 
-        required_log_snippets = [
-            # journalbeat can be started
-            "journalbeat is running",
-            # journalbeat can seek to the position defined in cursor_seek_fallback.
-            "Seeking method set to cursor, but no state is saved for reader. Starting to read from the end",
-            # message can be read from test journal
-            "\"message\": \"thinkpad_acpi: please report the conditions when this event happened to [email protected]\"",
-        ]
-        for snippet in required_log_snippets:
-            self.wait_until(lambda: self.log_contains(snippet),
-                            name="Line in '{}' Journalbeat log".format(snippet))
+        self.wait_until(lambda: self.log_contains("journalbeat is running"))
 
         exit_code = journalbeat_proc.kill_and_wait()
         assert exit_code == 0
 
+        # journalbeat is tailing an inactive journal with no cursor data
+        assert self.output_is_empty()
+
     @unittest.skipUnless(sys.platform.startswith("linux"), "Journald only on Linux")
     def test_read_events_with_existing_registry(self):
         """
@@ -142,19 +120,7 @@ def test_read_events_with_existing_registry(self):
         )
         journalbeat_proc = self.start_beat()
 
-        required_log_snippets = [
-            # journalbeat can be started
-            "journalbeat is running",
-            # journalbeat can seek to the position defined in the cursor
-            "Seeked to position defined in cursor",
-            # message can be read from test journal
-            "please report the conditions when this event happened to",
-            # only one event is read and published
-            'journalbeat successfully published events\t{"event.count": 1}',
-        ]
-        for snippet in required_log_snippets:
-            self.wait_until(lambda: self.log_contains(snippet),
-                            name="Line in '{}' Journalbeat log".format(snippet))
+        self.wait_until(lambda: self.output_has(lines=9))
 
         exit_code = journalbeat_proc.kill_and_wait()
         assert exit_code == 0
@@ -172,27 +138,13 @@ def test_read_events_with_include_matches(self):
                 ],
                 "seek": "head",
                 "include_matches": [
-                    "syslog.priority=5",
+                    "syslog.priority=6",
                 ]
             }],
         )
         journalbeat_proc = self.start_beat()
 
-        required_log_snippets = [
-            # journalbeat can be started
-            "journalbeat is running",
-            # journalbeat can seek to the position defined in the cursor
-            "Added matcher expression",
-            # message can be read from test journal
-            "unhandled HKEY event 0x60b0",
-            "please report the conditions when this event happened to",
-            "unhandled HKEY event 0x60b1",
-            # Four events with priority 5 is publised
-            'journalbeat successfully published events\t{"event.count": 4}',
-        ]
-        for snippet in required_log_snippets:
-            self.wait_until(lambda: self.log_contains(snippet),
-                            name="Line in '{}' Journalbeat log".format(snippet))
+        self.wait_until(lambda: self.output_has(lines=6))
 
         exit_code = journalbeat_proc.kill_and_wait()
         assert exit_code == 0

libbeat/tests/system/beat/beat.py

@@ -488,6 +488,21 @@ def output_has(self, lines, output_file=None):
         except IOError:
             return False
 
+    def output_is_empty(self, output_file=None):
+        """
+        Returns true if the output is empty.
+        """
+
+        # Init defaults
+        if output_file is None:
+            output_file = "output/" + self.beat_name
+
+        try:
+            with open(os.path.join(self.working_dir, output_file, ), "r", encoding="utf_8") as f:
+                return len([1 for line in f]) == 0
+        except IOError:
+            return True
+
     def output_has_message(self, message, output_file=None):
         """
         Returns true if the output has the given message field.