Merge pull request ClickHouse#82602 from ClickHouse/ci_post_run_test_cases_in_ft

CI: Run 00002_log_and_exception_messages_formatting in the end

Author: maxknv

ci/jobs/functional_tests.py

@@ -381,7 +381,7 @@ def start():
         results.append(
             Result.create_from(
                 name="Check errors",
-                results=CH.check_fatal_messeges_in_logs(),
+                results=CH.check_fatal_messages_in_logs(),
                 status=Result.Status.SUCCESS,
                 stopwatch=sw_,
             )

ci/jobs/scripts/clickhouse_proc.py

@@ -770,10 +770,16 @@ def _get_logs_archives_server(self):
         ).exists(), f"Log directory {self.log_dir} does not exist"
         return [f for f in glob.glob(f"{self.log_dir}/*.log")]
 
-    def check_fatal_messeges_in_logs(self):
+    def check_fatal_messages_in_logs(self):
         results = []
 
         # if command exit code is 1 - it's failed test case, script output will be stored into test case info
+        results.append(
+            Result.from_commands_run(
+                name="Exception in test runner",
+                command=f"! grep -A10 'Traceback (most recent call last)' {temp_dir}/job.log | head -n 100 | tee /dev/stderr | grep -q .",
+            )
+        )
         results.append(
             Result.from_commands_run(
                 name="Sanitizer assert (in stderr.log)",

tests/clickhouse-test

@@ -1422,7 +1422,7 @@ class TestCase:
         self.suite = suite
         self.case: str = case  # case file name
         self.args: Namespace = args
-        self.tags: Set[str] = suite.all_tags[case] if case in suite.all_tags else set()
+        self.tags: Set[str] = suite.all_tags[case]
         self.random_settings_limits = (
             suite.all_random_settings_limits[case]
             if case in suite.all_random_settings_limits
@@ -1550,7 +1550,7 @@ class TestCase:
 
         if tags and ("no-fasttest" in tags) and args.fast_tests_only:
             return FailureReason.FAST_ONLY
-        
+
         if tags and ("fasttest-only" in tags) and not args.fast_tests_only:
             return FailureReason.NOT_FAST_ONLY
 
@@ -1877,6 +1877,8 @@ class TestCase:
                 print("Cannot insert coverage data: ", str(e))
 
             # Check for dumped coverage files
+
+            # FIXME: This is a race condition
             file_pattern = "coverage.*"
             matching_files = glob.glob(file_pattern)
             for file_path in matching_files:
@@ -1892,8 +1894,13 @@ class TestCase:
                     )
                 except Exception as e:
                     print("Cannot insert coverage data: ", str(e))
+
                 # Remove the file even in case of exception to avoid accumulation and quadratic complexity.
-                os.remove(file_path)
+                try:
+                    os.remove(file_path)
+                except Exception as e:
+                    print("FIXME: Race! Cannot remove coverage file: ", str(e))
+            # FIXME: This is a race condition. END
 
             _ = clickhouse_execute(args, "SYSTEM FLUSH ASYNC INSERT QUEUE")
 
@@ -2443,8 +2450,7 @@ class TestSuite:
             ) = load_tags_and_random_settings_limits_from_file(
                 os.path.join(suite_dir, test_name)
             )  # noqa: ignore E203
-            if tags:
-                all_tags[test_name] = tags
+            all_tags[test_name] = tags or set()
             if random_settings_limits:
                 all_random_settings_limits[test_name] = random_settings_limits
         elapsed = (datetime.now() - start_time).total_seconds()
@@ -2499,13 +2505,20 @@ class TestSuite:
         self.all_tests = self.apply_test_runs(all_tests)
         self.all_tests.sort(key=self.tests_in_suite_key_func)
 
-        for test_name in self.all_tests:
-            if self.is_sequential_test(test_name):
+        post_run_check_tests = []
+
+        for test_name, tags in self.all_tags.items():
+            if "post-run-check" in tags:
+                # post_run_check_tests are tests that supposed to run after all normal tests finished
+                post_run_check_tests.append(test_name)
+                continue
+            if self.is_sequential_test(test_name, tags):
                 if not args.no_sequential:
                     self.sequential_tests.append(test_name)
             else:
                 if not args.no_parallel:
                     self.parallel_tests.append(test_name)
+        self.sequential_tests.extend(post_run_check_tests)
 
     def apply_test_runs(self, all_tests):
         test_runs = self.args.test_runs
@@ -2526,7 +2539,13 @@ class TestSuite:
             return False
         return "long" in self.all_tags[test_name]
 
-    def is_sequential_test(self, test_name):
+    def is_sequential_test(self, test_name, tags=None):
+        if tags is not None:
+            return (
+                ("no-parallel" in tags)
+                or ("sequential" in tags)
+                or ("stateful" in tags)
+            )
         if args.sequential:
             if any(s in test_name for s in args.sequential):
                 return True
@@ -2996,7 +3015,13 @@ def run_tests_process(*args_, **kwargs):
 
 
 def do_run_tests(
-    jobs, test_suite: TestSuite, args, exit_code, restarted_tests, server_died
+    jobs,
+    test_suite: TestSuite,
+    args,
+    exit_code,
+    restarted_tests,
+    server_died,
+    runner_process_killed,
 ):
     print(
         "Found",
@@ -3077,7 +3102,14 @@ def do_run_tests(
 
             for p in processes[:]:
                 if not p.is_alive():
+                    # Check if process was killed with exception
+                    if p.exitcode is not None and p.exitcode != 0:
+                        print(
+                            f"ERROR: Process {p.name} was killed with exit code {p.exitcode}"
+                        )
+                        runner_process_killed.set()
                     processes.remove(p)
+
     if test_suite.sequential_tests:
         run_tests_array(
             (
@@ -3336,6 +3368,7 @@ def try_get_skip_list(base_dir, name, remove_comment=True):
 def main(args):
     exit_code = multiprocessing.Value("i", 0)
     server_died = multiprocessing.Event()
+    runner_process_killed = multiprocessing.Event()
     multiprocessing_manager = multiprocessing.Manager()
     restarted_tests = multiprocessing_manager.list()
 
@@ -3468,12 +3501,21 @@ def main(args):
         test_suite.private_skip_list = private_skip_list
         test_suite.blacklist_check = blacklist_check
         total_tests_run += do_run_tests(
-            args.jobs, test_suite, args, exit_code, restarted_tests, server_died
+            args.jobs,
+            test_suite,
+            args,
+            exit_code,
+            restarted_tests,
+            server_died,
+            runner_process_killed,
         )
 
     if server_died.is_set():
         exit_code.value = 1
 
+    if runner_process_killed.is_set():
+        exit_code.value = 1
+
     if args.hung_check:
         # Some queries may execute in background for some time after test was finished. This is normal.
         print("Checking the hung queries: ", end="")

tests/queries/0_stateless/00002_log_and_exception_messages_formatting.sql

@@ -1,8 +1,8 @@
--- Tags: no-parallel, no-fasttest, no-ubsan, no-batch, no-flaky-check
--- no-parallel because we want to run this test when most of the other tests already passed
+-- Tags: no-fasttest, no-ubsan, no-batch, no-flaky-check, post-run-check
 
+-- post-run-check - to run it after all other tests:
 -- This is not a regular test. It is intended to run once after other tests to validate certain statistics about the whole test runs.
--- TODO: I advise to put in inside clickhouse-test instead.
+
 
 -- If this test fails, see the "Top patterns of log messages" diagnostics in the end of run.log