fix(unit_tests): use DbLogReader

unit_tests/test_cluster.py

@@ -21,12 +21,15 @@
 import tempfile
 import unittest
 from datetime import datetime
+from functools import cached_property
 from weakref import proxy as weakproxy
 
 from invoke import Result
 
 from sdcm.cluster import BaseNode, BaseCluster, BaseMonitorSet
+from sdcm.db_log_reader import DbLogReader
 from sdcm.sct_events import Severity
+from sdcm.sct_events.database import SYSTEM_ERROR_EVENTS_PATTERNS
 from sdcm.sct_events.group_common_events import ignore_upgrade_schema_errors
 from sdcm.sct_events.filters import DbEventsFilter
 from sdcm.sct_events.system import InstanceStatusEvent
@@ -35,8 +38,6 @@
 from unit_tests.dummy_remote import DummyRemote
 from unit_tests.lib.events_utils import EventsUtilsMixin
 
-# pylint: disable=protected-access
-
 
 class DummyNode(BaseNode):  # pylint: disable=abstract-method
     _system_log = None
@@ -85,14 +86,41 @@ def __init__(self, nodes):
         self.nodes = nodes
 
 
+class DummyDbLogReader(DbLogReader):
+    def get_scylla_debuginfo_file(self):
+        return "scylla_debug_info_file"
+
+
 class TestBaseNode(unittest.TestCase, EventsUtilsMixin):
     @classmethod
     def setUpClass(cls):
         cls.setup_events_processes(events_device=True, events_main_device=False, registry_patcher=True)
-        cls.node = DummyNode(name='test_node', parent_cluster=None,
-                             base_logdir=cls.temp_dir, ssh_login_info=dict(key_file='~/.ssh/scylla-test'))
-        cls.node.init()
-        cls.node.remoter = DummyRemote()
+
+    @cached_property
+    def node(self):
+        dummy_node = DummyNode(
+            name='test_node',
+            parent_cluster=None,
+            base_logdir=self.temp_dir,
+            ssh_login_info=dict(key_file='~/.ssh/scylla-test'),
+        )
+        dummy_node.init()
+        dummy_node.remoter = DummyRemote()
+        return dummy_node
+
+    @cached_property
+    def _db_log_reader(self):
+        return DummyDbLogReader(
+            system_log=self.node.system_log,
+            remoter=self.node.remoter,
+            node_name=str(self),
+            system_event_patterns=SYSTEM_ERROR_EVENTS_PATTERNS,
+            decoding_queue=None,
+            log_lines=False
+        )
+
+    def _read_and_publish_events(self):
+        self._db_log_reader._read_and_publish_events()  # pylint: disable=protected-access
 
     @classmethod
     def tearDownClass(cls):
@@ -113,7 +141,7 @@ def test_search_system_log_specific_log(self):
     def test_search_system_interlace_reactor_stall(self):
         self.node.system_log = os.path.join(os.path.dirname(__file__), 'test_data', 'system_interlace_stall.log')
 
-        self.node._read_system_log_and_publish_events(start_from_beginning=True)
+        self._read_and_publish_events()
 
         with self.get_raw_events_log().open() as events_file:
             events = [json.loads(line) for line in events_file]
@@ -129,7 +157,7 @@ def test_search_system_interlace_reactor_stall(self):
 
     def test_search_kernel_callstack(self):
         self.node.system_log = os.path.join(os.path.dirname(__file__), 'test_data', 'kernel_callstack.log')
-        self.node._read_system_log_and_publish_events(start_from_beginning=True)
+        self._read_and_publish_events()
         with self.get_raw_events_log().open() as events_file:
             events = [json.loads(line) for line in events_file]
 
@@ -145,7 +173,7 @@ def test_search_kernel_callstack(self):
     def test_search_cdc_invalid_request(self):
         self.node.system_log = os.path.join(os.path.dirname(__file__), 'test_data', 'system_cdc_invalid_request.log')
         with ignore_upgrade_schema_errors():
-            self.node._read_system_log_and_publish_events(start_from_beginning=True)
+            self._read_and_publish_events()
 
         time.sleep(0.1)
         with self.get_events_logger().events_logs_by_severity[Severity.ERROR].open() as events_file:
@@ -155,7 +183,7 @@ def test_search_cdc_invalid_request(self):
     def test_search_power_off(self):
         self.node.system_log = os.path.join(os.path.dirname(__file__), 'test_data', 'power_off.log')
         with DbEventsFilter(db_event=InstanceStatusEvent.POWER_OFF, node=self.node):
-            self.node._read_system_log_and_publish_events(start_from_beginning=True)
+            self._read_and_publish_events()
 
         InstanceStatusEvent.POWER_OFF().add_info(
             node="A", line_number=22,
@@ -172,7 +200,7 @@ def test_search_system_suppressed_messages(self):
         self.node.system_log = os.path.join(os.path.dirname(
             __file__), 'test_data', 'system_suppressed_messages.log')
 
-        self.node._read_system_log_and_publish_events(start_from_beginning=True)
+        self._read_and_publish_events()
 
         with self.get_raw_events_log().open() as events_file:
             events = [json.loads(line) for line in events_file]
@@ -186,7 +214,7 @@ def test_search_system_suppressed_messages(self):
     def test_search_one_line_backtraces(self):
         self.node.system_log = os.path.join(os.path.dirname(__file__), 'test_data', 'system_one_line_backtrace.log')
 
-        self.node._read_system_log_and_publish_events(start_from_beginning=True)
+        self._read_and_publish_events()
 
         with self.get_raw_events_log().open() as events_file:
             events = [json.loads(line) for line in events_file]

unit_tests/test_decode_backtrace.py

@@ -15,8 +15,11 @@
 import json
 import queue
 import unittest
+from functools import cached_property
 
 from sdcm.cluster import TestConfig
+from sdcm.db_log_reader import DbLogReader
+from sdcm.sct_events.database import SYSTEM_ERROR_EVENTS_PATTERNS
 
 from unit_tests.dummy_remote import DummyRemote
 from unit_tests.test_cluster import DummyNode
@@ -28,6 +31,8 @@ class DecodeDummyNode(DummyNode):  # pylint: disable=abstract-method
     def copy_scylla_debug_info(self, node, debug_file):
         return "scylla_debug_info_file"
 
+
+class DummyDbLogReader(DbLogReader):
     def get_scylla_debuginfo_file(self):
         return "scylla_debug_info_file"
 
@@ -36,28 +41,73 @@ class TestDecodeBactraces(unittest.TestCase, EventsUtilsMixin):
     @classmethod
     def setUpClass(cls):
         cls.setup_events_processes(events_device=True, events_main_device=False, registry_patcher=True)
-        cls.node = DecodeDummyNode(name='test_node', parent_cluster=None,
-                                   base_logdir=cls.temp_dir, ssh_login_info=dict(key_file='~/.ssh/scylla-test'))
-        cls.node.remoter = DummyRemote()
-
-        cls.monitor_node = DecodeDummyNode(name='test_monitor_node', parent_cluster=None,
-                                           base_logdir=cls.temp_dir, ssh_login_info=dict(key_file='~/.ssh/scylla-test'))
-        cls.monitor_node.remoter = DummyRemote()
-        cls.test_config = TestConfig()
 
     @classmethod
     def tearDownClass(cls):
         cls.teardown_events_processes()
 
+    @cached_property
+    def test_config(self):
+        result = TestConfig()
+        result.set_decoding_queue()
+        return result
+
+    @cached_property
+    def node(self):
+        dummy_node = DecodeDummyNode(
+            name='test_node',
+            parent_cluster=None,
+            base_logdir=self.temp_dir,
+            ssh_login_info=dict(key_file='~/.ssh/scylla-test'),
+        )
+        dummy_node.remoter = DummyRemote()
+        return dummy_node
+
+    @cached_property
+    def monitor_node(self):
+        dummy_monitor = DecodeDummyNode(
+            name='test_monitor_node',
+            parent_cluster=None,
+            base_logdir=self.temp_dir,
+            ssh_login_info=dict(key_file='~/.ssh/scylla-test'),
+        )
+        dummy_monitor.remoter = DummyRemote()
+        return dummy_monitor
+
+    @cached_property
+    def _db_log_reader(self):
+        return DummyDbLogReader(
+            system_log=self.node.system_log,
+            node_name=str(self),
+            remoter=self.node.remoter,
+            decoding_queue=self.test_config.DECODING_QUEUE,
+            system_event_patterns=SYSTEM_ERROR_EVENTS_PATTERNS,
+            log_lines=True,
+        )
+
+    @cached_property
+    def _db_log_reader_no_decoding(self):
+        return DummyDbLogReader(
+            system_log=self.node.system_log,
+            node_name=str(self),
+            remoter=self.node.remoter,
+            decoding_queue=None,
+            system_event_patterns=SYSTEM_ERROR_EVENTS_PATTERNS,
+            log_lines=False,
+        )
+
+    def _read_and_publish_events(self):
+        self._db_log_reader._read_and_publish_events()  # pylint: disable=protected-access
+
+    def _read_and_publish_events_no_decoding(self):
+        self._db_log_reader_no_decoding._read_and_publish_events()  # pylint: disable=protected-access
+
     def setUp(self):
         self.node.system_log = os.path.join(os.path.dirname(__file__), 'test_data', 'system.log')
 
     def test_01_reactor_stall_is_not_decoded_if_disabled(self):
-        self.test_config.DECODING_QUEUE = queue.Queue()
-        self.test_config.BACKTRACE_DECODING = False
-
         self.monitor_node.start_decode_on_monitor_node_thread()
-        self.node._read_system_log_and_publish_events()  # pylint: disable=protected-access
+        self._read_and_publish_events_no_decoding()  # pylint: disable=protected-access
         self.monitor_node.termination_event.set()
         self.monitor_node.stop_task_threads()
         self.monitor_node.wait_till_tasks_threads_are_stopped()
@@ -77,8 +127,7 @@ def test_02_reactor_stalls_is_decoded_if_enabled(self):
         self.test_config.DECODING_QUEUE = queue.Queue()
 
         self.monitor_node.start_decode_on_monitor_node_thread()
-        self.node._read_system_log_and_publish_events()  # pylint: disable=protected-access
-
+        self._read_and_publish_events()
         self.monitor_node.termination_event.set()
         self.monitor_node.stop_task_threads()
         self.monitor_node.wait_till_tasks_threads_are_stopped()
@@ -101,7 +150,7 @@ def test_03_decode_interlace_reactor_stall(self):  # pylint: disable=invalid-nam
         self.monitor_node.start_decode_on_monitor_node_thread()
         self.node.system_log = os.path.join(os.path.dirname(__file__), 'test_data', 'system_interlace_stall.log')
 
-        self.node._read_system_log_and_publish_events()  # pylint: disable=protected-access
+        self._read_and_publish_events()  # pylint: disable=protected-access
 
         self.monitor_node.termination_event.set()
         self.monitor_node.stop_task_threads()
@@ -125,7 +174,7 @@ def test_04_decode_backtraces_core(self):
         self.monitor_node.start_decode_on_monitor_node_thread()
         self.node.system_log = os.path.join(os.path.dirname(__file__), 'test_data', 'system_core.log')
 
-        self.node._read_system_log_and_publish_events()  # pylint: disable=protected-access
+        self._read_and_publish_events()
 
         self.monitor_node.termination_event.set()
         self.monitor_node.stop_task_threads()

unit_tests/test_sct_events_events_device.py

@@ -21,6 +21,7 @@
 from sdcm.sct_events.health import ClusterHealthValidatorEvent
 from sdcm.sct_events.events_device import EventsDevice, start_events_main_device, get_events_main_device
 from sdcm.sct_events.events_processes import EventsProcessesRegistry
+from sdcm.wait import wait_for
 
 
 class TestEventsDevice(unittest.TestCase):
@@ -80,6 +81,7 @@ def test_start_get_events_main_device(self):
         self.assertIsNone(get_events_main_device(_registry=self.events_processes_registry))
         start_events_main_device(_registry=self.events_processes_registry)
         events_device = get_events_main_device(_registry=self.events_processes_registry)
+        wait_for(func=events_device.is_alive, timeout=5)
         try:
             self.assertIsInstance(events_device, EventsDevice)
             self.assertEqual(events_device.events_counter, 0)