fix(sct lib): raft topology errors during rolling upgrade ignoration …

…added

sdcm/sct_events/events_device.py

@@ -119,6 +119,8 @@ def run(self):
                         pass
 
     def publish_event(self, event, timeout=PUBLISH_EVENT_TIMEOUT) -> None:
+
+        LOGGER.debug(f"LOGGING THE PROBLEMATIC EVENT {event.__dict__}")
         with verbose_suppress("%s: failed to write %s to %s", self, event, self.raw_events_log):
             with self._raw_events_lock, open(self.raw_events_log, "ab+", buffering=0) as log_file:
                 log_file.write(event.to_json().encode("utf-8") + b"\n")

sdcm/sct_events/filters.py

@@ -21,18 +21,48 @@
 
 
 class DbEventsFilter(BaseFilter):
+    """
+    A filter for database events, allowing filtering by event type, log line patterns,
+    specific nodes, and additional expiration time.
+
+    Parameters:
+        line (Optional[Union[str, re.Pattern]]):
+            A string or a regular expression pattern to filter log lines.
+            If you want to use a regular expression, it must be compiled using `re.compile()`
+            before passing it to this parameter.
+    """
+
     def __init__(self,
                  db_event: Union[LogEventProtocol, Type[LogEventProtocol]],
-                 line: Optional[str] = None,
+                 line: Optional[Union[str, re.Pattern]] = None,
                  node: Optional = None,
                  extra_time_to_expiration: Optional[int] = 0):
         super().__init__()
+        if node:
+            import logging
+            LOGGER = logging.getLogger(__name__)
+            LOGGER.debug(f"Node type: {type(node)}")
+            LOGGER.debug(f"Node __dict__: {getattr(node, '__dict__', 'No __dict__ attribute')}")
 
         self.filter_type = db_event.type
         self.filter_line = line
-        self.filter_node = str(node.name if hasattr(node, "name") else node) if node else None
-
+        self.filter_node = str(node.name) if hasattr(node, "name") else None
         self.extra_time_to_expiration = extra_time_to_expiration
+        self.regex = None  # Initialize regex to None
+        self.regex_flags = 0  # Initialize regex_flags to default value
+        if isinstance(line, re.Pattern):
+            self.regex = line.pattern
+            self.regex_flags = line.flags
+        elif isinstance(line, str):
+            self.regex = re.escape(line)
+            self.regex_flags = re.MULTILINE | re.DOTALL
+
+    @cached_property
+    def _regex(self):
+        try:
+            return self.regex and re.compile(self.regex, self.regex_flags)
+        except Exception as exc:
+            raise ValueError(f'Compilation of the regexp "{self.regex}" failed with error: {exc}') from None
 
     def eval_filter(self, event: LogEventProtocol) -> bool:
         if not isinstance(event, LogEventProtocol):
@@ -43,12 +73,14 @@ def eval_filter(self, event: LogEventProtocol) -> bool:
 
         result = bool(self.filter_type) and self.filter_type == event.type
 
-        if self.filter_line:
-            result &= self.filter_line in (getattr(event, "line", "") or "")
+        if self._regex:
+            event_line = (getattr(event, "line", "") or "")
+            result &= (self._regex.search(event_line) is not None)
 
         if self.filter_node:
             result &= self.filter_node in (getattr(event, "node", "") or "").split()
 
+
         return result
 
     def cancel_filter(self) -> None:
@@ -61,8 +93,8 @@ def msgfmt(self) -> str:
         output = ['{0.base}']
         if self.filter_type:
             output.append('type={0.filter_type}')
-        if self.filter_line:
-            output.append('line={0.filter_line}')
+        if self._regex:
+            output.append(f'line={self._regex.pattern}')
         if self.filter_node:
             output.append('node={0.filter_node}')
         return '(' + (' '.join(output)) + ')'

sdcm/sct_events/group_common_events.py

@@ -10,7 +10,7 @@
 # See LICENSE for more details.
 #
 # Copyright (c) 2020 ScyllaDB
-
+import re
 from contextlib import contextmanager, ExitStack, ContextDecorator
 from functools import wraps
 from typing import ContextManager, Callable, Sequence
@@ -105,6 +105,12 @@ def ignore_topology_change_coordinator_errors():
                      " (raft topology: exec_global_command(barrier) failed with seastar::rpc::closed_error"
                      " (connection is closed))",
             ))
+            stack.enter_context(DbEventsFilter(
+                db_event=DatabaseLogEvent.RUNTIME_ERROR,
+                line=re.compile(r".*raft_topology - drain rpc failed, proceed to fence "
+                                r"old writes:.*connection is closed")
+,
+            ))
         yield