src/frostfs_testlib/hosting/docker_host.py

@@ -212,6 +212,30 @@ class DockerHost(Host):
             with open(file_path, "wb") as file:
                 file.write(logs)
 
+    def get_filtered_logs(
+        self,
+        filter_regex: str,
+        since: Optional[datetime] = None,
+        until: Optional[datetime] = None,
+        unit: Optional[str] = None,
+    ) -> str:
+        client = self._get_docker_client()
+        filtered_logs = ""
+        for service_config in self._config.services:
+            container_name = self._get_service_attributes(service_config.name).container_name
+            try:
+                filtered_logs = client.logs(container_name, since=since, until=until)
+            except HTTPError as exc:
+                logger.info(f"Got exception while dumping logs of '{container_name}': {exc}")
+                continue
+
+            matches = re.findall(filter_regex, filtered_logs, re.IGNORECASE + re.MULTILINE)
+            found = list(matches)
+            if found:
+                filtered_logs += f"{container_name}:\n{os.linesep.join(found)}"
+
+        return filtered_logs
+
     def is_message_in_logs(
         self,
         message_regex: str,

src/frostfs_testlib/hosting/interfaces.py

@@ -115,7 +115,6 @@ class Host(ABC):
             service_name: Name of the service to restart.
         """
 
-
     @abstractmethod
     def get_data_directory(self, service_name: str) -> str:
         """
@@ -126,7 +125,6 @@ class Host(ABC):
             service_name: Name of storage node service.
         """
 
-
     @abstractmethod
     def wait_success_suspend_process(self, process_name: str) -> None:
         """Search for a service ID by its name and stop the process
@@ -251,6 +249,27 @@ class Host(ABC):
             filter_regex: regex to filter output
         """
 
+    @abstractmethod
+    def get_filtered_logs(
+        self,
+        filter_regex: str,
+        since: Optional[datetime] = None,
+        until: Optional[datetime] = None,
+        unit: Optional[str] = None,
+    ) -> str:
+        """Get logs from host filtered by regex.
+
+        Args:
+            filter_regex: regex filter for logs.
+            since: If set, limits the time from which logs should be collected. Must be in UTC.
+            until: If set, limits the time until which logs should be collected. Must be in UTC.
+            unit: required unit.
+
+        Returns:
+            Found entries as str if any found.
+            Empty string otherwise.
+        """
+
     @abstractmethod
     def is_message_in_logs(
         self,
@@ -270,10 +289,11 @@ class Host(ABC):
             True if message found in logs in the given time frame.
             False otherwise.
         """
-        
 
     @abstractmethod
-    def wait_for_service_to_be_in_state(self, systemd_service_name: str, expected_state: str, timeout: int) -> None:
+    def wait_for_service_to_be_in_state(
+        self, systemd_service_name: str, expected_state: str, timeout: int
+    ) -> None:
         """
             Waites for service to be in specified state.