feat(ci-insights): Send only 1 span for reruns (#280)

The goal of this change is to send only 1 span wrapping all reruns. To
keep the required context, we are also sending `cicd.test.rerun_count`
and `cicd.test.flaky` attributes now.

References: MRGFY-6112

Author: remyduthu

pytest_mergify/__init__.py

@@ -211,56 +211,70 @@ def pytest_runtest_protocol(
         # calculate the number of reruns.
         with self.tracer.start_as_current_span(
             item.nodeid, attributes=attributes, context=context
-        ):
-            _pytest.runner.runtestprotocol(item=item, nextitem=nextitem, log=True)
+        ) as current_span:
+            distinct_outcomes = set()
 
-        if not self.mergify_ci.flaky_detector:
-            return True
+            for report in _pytest.runner.runtestprotocol(
+                item=item, nextitem=nextitem, log=True
+            ):
+                distinct_outcomes.add(report.outcome)
 
-        timeout_seconds = pytest_timeout._get_item_settings(item).timeout
+            if not self.mergify_ci.flaky_detector:
+                return True
 
-        for _ in range(
-            self.mergify_ci.flaky_detector.get_rerun_count_for_test(
-                test=item.nodeid,
-                timeout=datetime.timedelta(seconds=timeout_seconds)
-                if timeout_seconds
-                else None,
-            )
-        ):
-            with self.tracer.start_as_current_span(
-                item.nodeid, attributes=attributes, context=context
+            timeout_seconds = pytest_timeout._get_item_settings(item).timeout
+
+            rerun_count = 0
+            for _ in range(
+                self.mergify_ci.flaky_detector.get_rerun_count_for_test(
+                    test=item.nodeid,
+                    timeout=datetime.timedelta(seconds=timeout_seconds)
+                    if timeout_seconds
+                    else None,
+                )
             ):
-                self._reruntestprotocol(item, nextitem)
+                for report in self._reruntestprotocol(item, nextitem):
+                    distinct_outcomes.add(report.outcome)
 
-            if self.mergify_ci.flaky_detector.is_deadline_exceeded():
-                return True
+                rerun_count += 1
+
+                if self.mergify_ci.flaky_detector.is_deadline_exceeded():
+                    break
+
+            if "failed" in distinct_outcomes and "passed" in distinct_outcomes:
+                current_span.set_attribute("cicd.test.flaky", True)
+
+            current_span.set_attribute("cicd.test.rerun_count", rerun_count)
 
         return True
 
     def _reruntestprotocol(
         self, item: _pytest.nodes.Item, nextitem: typing.Optional[_pytest.nodes.Item]
-    ) -> None:
+    ) -> typing.List[_pytest.reports.TestReport]:
         """
         Run the protocol for a rerun of a given test.
 
         In `new` mode, we log rerun failures to pytest's report to enforce a
         quality gate and prevent merging PRs with new flaky tests. In other
         modes (`unhealthy`), we skip logging to avoid blocking CI, but still
-        capture reruns in spans.
+        capture reruns in metrics.
         """
 
         if not self.mergify_ci.flaky_detector:
-            return
+            return []
 
         if self.mergify_ci.flaky_detector.mode == "new":
-            _pytest.runner.runtestprotocol(item=item, nextitem=nextitem, log=True)
-            return
+            return _pytest.runner.runtestprotocol(
+                item=item, nextitem=nextitem, log=True
+            )
 
         reports = _pytest.runner.runtestprotocol(
             item=item, nextitem=nextitem, log=False
         )
         for report in reports:
-            self.pytest_runtest_logreport(report)
+            self.mergify_ci.flaky_detector.try_fill_metrics_from_report(report)
+
+        return reports
 
     @pytest.hookimpl(tryfirst=True)
     def pytest_runtest_teardown(
@@ -340,8 +354,7 @@ def pytest_runtest_logreport(self, report: _pytest.reports.TestReport) -> None:
         if not self.mergify_ci.flaky_detector:
             return
 
-        detected = self.mergify_ci.flaky_detector.detect_from_report(report)
-        if not detected:
+        if not self.mergify_ci.flaky_detector.try_fill_metrics_from_report(report):
             return
 
         test_span.set_attributes({"cicd.test.flaky_detection": True})

pytest_mergify/flaky_detection.py

@@ -165,7 +165,7 @@ def _fetch_context(self) -> _FlakyDetectionContext:
 
         return result
 
-    def detect_from_report(self, report: _pytest.reports.TestReport) -> bool:
+    def try_fill_metrics_from_report(self, report: _pytest.reports.TestReport) -> bool:
         if report.outcome not in ["failed", "passed"]:
             return False
 

tests/conftest.py

@@ -17,7 +17,6 @@
 )
 
 import pytest_mergify
-import pytest_mergify.quarantine
 from pytest_mergify import utils
 
 pytest_plugins = ["pytester"]
@@ -118,7 +117,7 @@ def _run(
             result.assert_outcomes(passed=1)
         if isinstance(plugin.mergify_ci.exporter, InMemorySpanExporter):
             spans = plugin.mergify_ci.exporter.get_finished_spans()
-            spans_as_dict = _map_spans_to_dict(spans)
+            spans_as_dict = {span.name: span for span in spans}
             # Make sure we don't lose spans in the process
             assert len(spans_as_dict) == len(spans)
         else:
@@ -129,21 +128,6 @@ def _run(
     return _run
 
 
-def _map_spans_to_dict(
-    spans: typing.Tuple[ReadableSpan, ...],
-) -> typing.Dict[str, ReadableSpan]:
-    result: typing.Dict[str, ReadableSpan] = {}
-
-    for span in spans:
-        if span.name not in result:
-            result[span.name] = span
-            continue
-
-        result[f"{span.name}.{span.start_time}.{span.end_time}"] = span
-
-    return result
-
-
 class TestHTTPRequestHandler(http.server.BaseHTTPRequestHandler):
     # Class attribute for the response code, set by the fixture.
     response_code: int = 200

tests/test_ci_insights.py

@@ -7,7 +7,6 @@
 import _pytest.reports
 import pytest
 import responses
-from opentelemetry.sdk import trace
 
 import pytest_mergify
 from pytest_mergify import ci_insights
@@ -199,10 +198,11 @@ def test_corge():
     )
 
     assert spans is not None
+    assert len(spans) == 1 + 6  # 1 for the session and one per test.
 
-    # 1 span for the session and one per test.
-    assert len(spans) == 1 + sum(result.parseoutcomes().values())
-
+    flaky_tests = [
+        "test_flaky_detection_for_new_tests.py::test_bar",
+    ]
     new_tests = [
         "test_flaky_detection_for_new_tests.py::test_bar",
         "test_flaky_detection_for_new_tests.py::test_baz",
@@ -212,12 +212,13 @@ def test_corge():
         assert span is not None
         assert span.attributes is not None
 
-        is_new_test = span.name in new_tests
-        if not is_new_test:
-            continue
+        if span.name in flaky_tests:
+            assert span.attributes.get("cicd.test.flaky", False) is True
 
-        assert span.attributes.get("cicd.test.flaky_detection", False)
-        assert span.attributes.get("cicd.test.new", False)
+        if span.name in new_tests:
+            assert span.attributes.get("cicd.test.flaky_detection", False) is True
+            assert span.attributes.get("cicd.test.new", False) is True
+            assert span.attributes.get("cicd.test.rerun_count", 0) == 1000
 
 
 @responses.activate
@@ -283,10 +284,9 @@ def test_quux():
     )
 
     assert spans is not None
+    assert len(spans) == 5 + 1  # 1 for the session and one per test.
 
-    # 1 span for the session and one per test, including 1000 reruns for each unhealthy test.
-    assert len(spans) == 1 + 3005
-
+    flaky_tests = ["test_flaky_detection_for_unhealthy_tests.py::test_bar"]
     unhealthy_tests = [
         "test_flaky_detection_for_unhealthy_tests.py::test_bar",
         "test_flaky_detection_for_unhealthy_tests.py::test_baz",
@@ -296,12 +296,13 @@ def test_quux():
         assert span is not None
         assert span.attributes is not None
 
-        is_unhealthy_test = span.name in unhealthy_tests
-        if not is_unhealthy_test:
-            continue
+        if span.name in flaky_tests:
+            assert span.attributes.get("cicd.test.flaky", False) is True
 
-        assert span.attributes.get("cicd.test.flaky_detection", False)
-        assert not span.attributes.get("cicd.test.new", False)
+        if span.name in unhealthy_tests:
+            assert not span.attributes.get("cicd.test.new")
+            assert span.attributes.get("cicd.test.flaky_detection", False) is True
+            assert span.attributes.get("cicd.test.rerun_count", 0) == 1000
 
 
 @responses.activate
@@ -448,14 +449,16 @@ def test_bar():
     result.assert_outcomes(passed=1002)
 
     assert spans is not None
+    assert len(spans) == 1 + 2  # 1 for the session and one per test.
 
-    # Only 1 execution for the existing test (`test_foo`), 1001 executions for
-    # the new test (`test_bar`) and 1 additional span for the session.
-    assert _get_span_counts(spans) == {
-        "pytest session start": 1,
-        "test_flaky_detection_with_only_one_new_test_at_the_end.py::test_foo": 1,
-        "test_flaky_detection_with_only_one_new_test_at_the_end.py::test_bar": 1001,
-    }
+    span = spans.get(
+        "test_flaky_detection_with_only_one_new_test_at_the_end.py::test_bar"
+    )
+    assert span is not None
+    assert span.attributes is not None
+    assert span.attributes.get("cicd.test.flaky_detection", False) is True
+    assert span.attributes.get("cicd.test.new", False) is True
+    assert span.attributes.get("cicd.test.rerun_count", 0) == 1000
 
 
 @responses.activate
@@ -622,13 +625,3 @@ def test_bar():
     # Unknown test should have been filtered out after collection.
     assert len(plugin.mergify_ci.flaky_detector._context.existing_test_names) == 2
     assert len(plugin.mergify_ci.flaky_detector._context.unhealthy_test_names) == 1
-
-
-def _get_span_counts(
-    spans: typing.Dict[str, trace.ReadableSpan],
-) -> typing.Dict[str, int]:
-    result: typing.Dict[str, int] = {}
-    for span in spans.values():
-        result[span.name] = result.get(span.name, 0) + 1
-
-    return result

tests/test_flaky_detection.py

@@ -62,7 +62,7 @@ def test_flaky_detector_get_duration_before_deadline() -> None:
     assert detector._get_duration_before_deadline() == datetime.timedelta(seconds=10)
 
 
-def test_flaky_detector_detect_from_report() -> None:
+def test_flaky_detector_try_fill_metrics_from_report() -> None:
     def make_report(
         nodeid: str, when: typing.Literal["setup", "call", "teardown"], duration: float
     ) -> _pytest.reports.TestReport:
@@ -79,13 +79,25 @@ def make_report(
     detector = InitializedFlakyDetector()
     detector._context = _make_flaky_detection_context(max_test_name_length=100)
 
-    detector.detect_from_report(make_report(nodeid="foo", when="setup", duration=1))
-    detector.detect_from_report(make_report(nodeid="foo", when="call", duration=2))
-    detector.detect_from_report(make_report(nodeid="foo", when="teardown", duration=3))
+    detector.try_fill_metrics_from_report(
+        make_report(nodeid="foo", when="setup", duration=1)
+    )
+    detector.try_fill_metrics_from_report(
+        make_report(nodeid="foo", when="call", duration=2)
+    )
+    detector.try_fill_metrics_from_report(
+        make_report(nodeid="foo", when="teardown", duration=3)
+    )
 
-    detector.detect_from_report(make_report(nodeid="foo", when="setup", duration=4))
-    detector.detect_from_report(make_report(nodeid="foo", when="call", duration=5))
-    detector.detect_from_report(make_report(nodeid="foo", when="teardown", duration=6))
+    detector.try_fill_metrics_from_report(
+        make_report(nodeid="foo", when="setup", duration=4)
+    )
+    detector.try_fill_metrics_from_report(
+        make_report(nodeid="foo", when="call", duration=5)
+    )
+    detector.try_fill_metrics_from_report(
+        make_report(nodeid="foo", when="teardown", duration=6)
+    )
 
     metrics = detector._test_metrics.get("foo")
     assert metrics is not None