chore(ci-insights): Add debug logs to the flaky detector (#348)

The goal of this change is to add some structure debug logs to the flaky
detector. We often need these logs to debug some budget/reruns issues.
The logs are only displayed if the `PYTEST_MERGIFY_DEBUG` flag is set.

Depends-On: #347

Author: remyduthu

pytest_mergify/flaky_detection.py

@@ -155,6 +155,10 @@ class FlakyDetector:
     https://github.com/pytest-dev/pytest-rerunfailures/blob/master/src/pytest_rerunfailures.py#L503-L542
     """
 
+    _debug_logs: typing.List[utils.StructuredLog] = dataclasses.field(
+        init=False, default_factory=list
+    )
+
     def __post_init__(self) -> None:
         self._context = self._fetch_context()
 
@@ -257,11 +261,24 @@ def is_rerunning_test(self, test: str) -> bool:
     def is_last_rerun_for_test(self, test: str) -> bool:
         metrics = self._test_metrics[test]
 
-        return (
+        will_exceed_deadline = metrics.will_exceed_deadline()
+        will_exceed_rerun_count = (
             metrics.rerun_count >= self._context.max_test_execution_count
-            or metrics.will_exceed_deadline()
         )
 
+        self._debug_logs.append(
+            utils.StructuredLog.make(
+                message="Check for last rerun",
+                test=test,
+                deadline=metrics.deadline.isoformat() if metrics.deadline else None,
+                rerun_count=metrics.rerun_count,
+                will_exceed_deadline=will_exceed_deadline,
+                will_exceed_rerun_count=will_exceed_rerun_count,
+            )
+        )
+
+        return will_exceed_deadline or will_exceed_rerun_count
+
     def make_report(self) -> str:
         result = "🐛 Flaky detection"
         if self._over_length_tests:
@@ -334,16 +351,34 @@ def make_report(self) -> str:
                 f"Reference: https://github.com/pytest-dev/pytest-timeout?tab=readme-ov-file#avoiding-timeouts-in-fixtures"
             )
 
+        if os.environ.get("PYTEST_MERGIFY_DEBUG") and self._debug_logs:
+            result += f"{os.linesep}🔎 Debug Logs"
+            for log in self._debug_logs:
+                result += f"{os.linesep}{log.to_json()}"
+
         return result
 
     def set_test_deadline(
         self, test: str, timeout: typing.Optional[datetime.timedelta] = None
     ) -> None:
         metrics = self._test_metrics[test]
 
+        remaining_budget = self._get_remaining_budget_duration()
+        remaining_tests = self._count_remaining_tests()
+
         # Distribute remaining budget equally across remaining tests.
         metrics.deadline = datetime.datetime.now(datetime.timezone.utc) + (
-            self._get_remaining_budget_duration() / self._count_remaining_tests()
+            remaining_budget / remaining_tests
+        )
+        self._debug_logs.append(
+            utils.StructuredLog.make(
+                message="Deadline set",
+                test=test,
+                available_budget=str(self._available_budget_duration),
+                remaining_budget=str(remaining_budget),
+                all_tests=len(self._tests_to_process),
+                remaining_tests=remaining_tests,
+            )
         )
 
         if not timeout:
@@ -356,6 +391,15 @@ def set_test_deadline(
         if not metrics.deadline or timeout_deadline < metrics.deadline:
             metrics.deadline = timeout_deadline
             metrics.prevented_timeout = True
+            self._debug_logs.append(
+                utils.StructuredLog.make(
+                    message="Deadline updated to prevent timeout",
+                    test=test,
+                    timeout=str(timeout),
+                    safe_timeout=str(safe_timeout),
+                    deadline=metrics.deadline,
+                )
+            )
 
     def suspend_item_finalizers(self, item: _pytest.nodes.Item) -> None:
         """

pytest_mergify/utils.py

@@ -1,3 +1,6 @@
+import dataclasses
+import datetime
+import json
 import os
 import re
 import subprocess
@@ -15,6 +18,30 @@
 }
 
 
+@dataclasses.dataclass
+class StructuredLog:
+    message: str
+    timestamp: datetime.datetime
+    attributes: typing.Dict[str, typing.Any]
+
+    @classmethod
+    def make(cls, message: str, **kwargs: typing.Any) -> "StructuredLog":
+        return cls(
+            message=message,
+            timestamp=datetime.datetime.now(datetime.timezone.utc),
+            attributes=kwargs,
+        )
+
+    def to_json(self) -> str:
+        return json.dumps(
+            {
+                "timestamp": self.timestamp.isoformat(),
+                "message": self.message,
+                **self.attributes,
+            }
+        )
+
+
 def is_in_ci() -> bool:
     return strtobool(os.environ.get("CI", "false")) or strtobool(
         os.environ.get("PYTEST_MERGIFY_ENABLE", "false")