import os import shutil from datetime import datetime import allure import pytest from frostfs_testlib.testing.cluster_test_base import ClusterTestBase class TestLogs(ClusterTestBase): @pytest.mark.logs_after_session def test_logs_after_session(self, temp_directory: str, session_start_time: datetime): """ This test automatically added to any test run to check logs from cluster for critical errors. """ end_time = datetime.utcnow() logs_dir = os.path.join(temp_directory, "logs") os.makedirs(logs_dir) # Using \b here because 'oom' and 'panic' can sometimes be found in OID or CID issues_regex = r"\bpanic\b|\boom\b|too many|insufficient funds" hosts_with_problems = [] for host in self.cluster.hosts: with allure.step(f"Check logs on {host.config.address}"): if host.is_message_in_logs(issues_regex, session_start_time, end_time): hosts_with_problems.append(host.config.address) host.dump_logs( logs_dir, since=session_start_time, until=end_time, filter_regex=issues_regex, ) if hosts_with_problems: self._attach_logs(logs_dir) assert ( not hosts_with_problems ), f"The following hosts contains contain critical errors in system logs: {', '.join(hosts_with_problems)}" def _attach_logs(self, logs_dir: str) -> None: # Zip all files and attach to Allure because it is more convenient to download a single # zip with all logs rather than mess with individual logs files per service or node logs_zip_file_path = shutil.make_archive(logs_dir, "zip", logs_dir) allure.attach.file(logs_zip_file_path, name="logs.zip", extension="zip")