[CI][Green-Ray][4] Compute and store unique crash pattern from logs

release/ray_release/anyscale_util.py

@@ -7,7 +7,7 @@
     from anyscale.sdk.anyscale_client.sdk import AnyscaleSDK
 
 
-LAST_LOGS_LENGTH = 10
+LAST_LOGS_LENGTH = 30
 
 
 def find_cloud_by_name(

release/ray_release/job_manager/anyscale_job_manager.py

@@ -332,7 +332,10 @@ def _get_logs():
                     )
                     print("", flush=True)
             output = buf.getvalue().strip()
-            if "### Starting ###" not in output:
+            # Many of Ray components have their separated logs (e.g. dashboard,
+            # gcs_server, etc.), so the interesting errors are not always in the
+            # job logs. If the job has no logs, check other ray logs for error patterns.
+            if not output:
                 output = self._get_ray_error_logs()
             assert output, "No logs fetched"
             return "\n".join(output.splitlines()[-LAST_LOGS_LENGTH * 3 :])

release/ray_release/log_aggregator.py

@@ -0,0 +1,100 @@
+import re
+from typing import List
+
+TRACEBACK_PATTERN = "Traceback (most recent call last)"
+
+
+class LogAggregator:
+    def __init__(self, log: str):
+        self.log = log
+
+    def compute_crash_pattern(self) -> str:
+        stack_trace = LogAggregator._compute_stack_trace(self.log.splitlines())
+        # truncate short enough to store in databases, but long enough to keep the
+        # pattern unique
+        return LogAggregator._compute_signature(stack_trace)[:4000]
+
+    @staticmethod
+    def _compute_signature(stack_trace: List[str]) -> str:
+        """
+        Compute signature pattern from stack trace, by remove factors such as date,
+        time, temp directory, line numbers, etc. This help to aggregate similar logs
+        into same bug patterns
+        """
+        massaged_trace = []
+        for line in stack_trace:
+            line = re.sub(r"\d", "", line.strip())
+            if line == "Traceback (most recent call last):":
+                continue
+            file_line = re.search(r'File "(.*)", (.*)', line)
+            if file_line:
+                # append the file's base name and caller information; the result string
+                # is not something meaningful to human, we just need something that
+                # uniquely represent the stack trace
+                line = f'{file_line.group(1).split("/")[-1]}{file_line.group(2)}'
+            massaged_trace.append(line)
+        return "".join(massaged_trace)
+
+    @staticmethod
+    def _compute_stack_trace(logs: List[str]) -> List[str]:
+        """
+        Extract stack trace pattern from the logs. Stack trace pattern often matches
+        the following:
+        ERROR ...
+        Traceback (most recent call last):
+            File "...", line ..., in ...
+            ...
+        Exception: exception error
+        """
+        error_stacktrace = []
+        stacktrace = []
+        i = 0
+        while i < len(logs):
+            stack = []
+            trace = error_stacktrace
+            # Search for lines that are either
+            # ... ERROR ...
+            # or
+            # ... ERROR ...
+            # Traceback (most recent call last):
+            if "ERROR" in logs[i]:
+                stack.append(logs[i])
+                next = i + 1
+                if i + 1 < len(logs) and TRACEBACK_PATTERN in logs[i + 1]:
+                    stack.append(logs[i + 1])
+                    next = i + 2
+            # Or if the line with ERROR does not exist, just search for the line with
+            # Traceback (most recent call last):
+            elif TRACEBACK_PATTERN in logs[i]:
+                stack.append(logs[i])
+                trace = stacktrace
+                next = i + 1
+            # Or else, skip this line and continue
+            else:
+                i = i + 1
+                continue
+            # If the line that contains ERROR, Traceback, etc. is found, scan the logs
+            # until the line no longer has indentation. This is because stack trace
+            # is always indented, and stops when the line is no longer indented
+            while next < len(logs):
+                if logs[next].startswith((" ", "\t")):
+                    stack.append(logs[next])
+                    next = next + 1
+                else:
+                    break
+            # Finished capturing the entire stack trace
+            if next < len(logs):
+                stack.append(logs[next])
+            if stack:
+                trace.append(stack)
+            i = next + 1
+
+        # Favor stack trace that contains the ERROR keyword
+        if error_stacktrace:
+            return error_stacktrace[-1]
+
+        # Otherwise any stack trace is fine
+        if stacktrace:
+            return stacktrace[-1]
+
+        return []

release/ray_release/reporter/db.py

@@ -7,6 +7,7 @@
 from ray_release.result import Result
 from ray_release.config import Test
 from ray_release.logger import logger
+from ray_release.log_aggregator import LogAggregator
 
 
 class DBReporter(Reporter):
@@ -40,6 +41,9 @@ def report_result(self, test: Test, result: Result):
             "return_code": result.return_code,
             "smoke_test": result.smoke_test,
             "extra_tags": result.extra_tags or {},
+            "crash_pattern": LogAggregator(
+                result.last_logs or ""
+            ).compute_crash_pattern(),
         }
 
         logger.debug(f"Result json: {json.dumps(result_json)}")

release/ray_release/tests/test_log_aggregator.py

@@ -0,0 +1,68 @@
+from ray_release.log_aggregator import LogAggregator
+
+
+def test_compute_stack_pattern():
+    assert (
+        LogAggregator(
+            "\n".join(
+                [
+                    "haha",
+                    "Traceback (most recent call last):",
+                    '    File "/tmp/something", line 584',
+                    "Exception: yaya45",
+                    "hehe",
+                ]
+            )
+        ).compute_crash_pattern()
+        == "somethingline Exception: yaya"
+    )
+
+
+def test_compute_signature():
+    assert (
+        LogAggregator._compute_signature(
+            [
+                "Traceback (most recent call last):",
+                '   File "/tmp/something", line 584',
+                "Exception: yaya45",
+            ]
+        )
+        == "somethingline Exception: yaya"
+    )
+
+
+def test_compute_stack_trace():
+    trace = [
+        "Traceback (most recent call last):",
+        '   File "/tmp/something", line 584, in run_release_test',
+        "       raise pipeline_exception",
+        "ray_release.exception.JobNoLogsError: Could not obtain logs for the job.",
+    ]
+    error_trace = [
+        "[2023-01-01] ERROR: something is wrong",
+        "Traceback (most recent call last):",
+        '   File "/tmp/something", line 584, in run_release_test',
+        "       raise pipeline_exception",
+        "ray_release.exception.JobStartupTimeout: Cluster did not start.",
+    ]
+    error_trace_short = [
+        "[2023-01-01] ERROR: something is wrong"
+        '   File "/tmp/something", line 584, in run_release_test',
+        "       raise pipeline_exception",
+        "ray_release.exception.JobStartupTimeout: Cluster did not start.",
+    ]
+    assert LogAggregator._compute_stack_trace(["haha"] + trace + ["hehe"]) == trace
+    assert (
+        LogAggregator._compute_stack_trace(["haha"] + error_trace + ["hehe"])
+        == error_trace
+    )
+    assert (
+        LogAggregator._compute_stack_trace(["haha"] + error_trace_short + ["hehe"])
+        == error_trace_short
+    )
+    assert (
+        LogAggregator._compute_stack_trace(
+            ["haha"] + trace + ["w00t"] + error_trace + ["hehe"]
+        )
+        == error_trace
+    )