refactor(ci_visibility): efd refactor

ddtrace/internal/ci_visibility/_api_client.py

@@ -12,6 +12,7 @@
 from ddtrace.ext.test_visibility import ITR_SKIPPING_LEVEL
 from ddtrace.ext.test_visibility._item_ids import TestModuleId
 from ddtrace.ext.test_visibility._item_ids import TestSuiteId
+from ddtrace.internal.ci_visibility.api._efd_settings import EarlyFlakeDetectionSettings
 from ddtrace.internal.ci_visibility.constants import AGENTLESS_API_KEY_HEADER_NAME
 from ddtrace.internal.ci_visibility.constants import AGENTLESS_DEFAULT_SITE
 from ddtrace.internal.ci_visibility.constants import EVP_PROXY_AGENT_BASE_PATH
@@ -83,16 +84,6 @@ class TestVisibilitySkippableItemsError(Exception):
     pass
 
 
-@dataclasses.dataclass(frozen=True)
-class EarlyFlakeDetectionSettings:
-    enabled: bool = False
-    slow_test_retries_5s: int = 10
-    slow_test_retries_10s: int = 5
-    slow_test_retries_30s: int = 3
-    slow_test_retries_5m: int = 2
-    faulty_session_threshold: int = 30
-
-
 @dataclasses.dataclass(frozen=True)
 class TestManagementSettings:
     enabled: bool = False

ddtrace/internal/ci_visibility/api/_efd_settings.py

@@ -0,0 +1,65 @@
+"""Settings for Early Flake Detection.
+
+This module contains settings classes for the Early Flake Detection feature.
+"""
+
+import dataclasses
+from typing import List
+from typing import Tuple
+
+from ddtrace.internal.logger import get_logger
+
+
+log = get_logger(__name__)
+
+
+@dataclasses.dataclass(frozen=True)
+class EarlyFlakeDetectionSettings:
+    """Settings for Early Flake Detection.
+
+    This class encapsulates the settings for the Early Flake Detection feature, which
+    includes configuration for retry counts based on test duration and thresholds for
+    determining faulty sessions.
+
+    Attributes:
+        enabled: Whether Early Flake Detection is enabled
+        slow_test_retries_5s: Maximum retries for tests that take less than 5 seconds
+        slow_test_retries_10s: Maximum retries for tests that take less than 10 seconds
+        slow_test_retries_30s: Maximum retries for tests that take less than 30 seconds
+        slow_test_retries_5m: Maximum retries for tests that take less than 5 minutes
+        faulty_session_threshold: Percentage threshold for determining if a session is faulty
+    """
+
+    enabled: bool = False
+    slow_test_retries_5s: int = 10
+    slow_test_retries_10s: int = 5
+    slow_test_retries_30s: int = 3
+    slow_test_retries_5m: int = 2
+    faulty_session_threshold: int = 30
+
+    def get_threshold_definitions(self) -> List[Tuple[float, int]]:
+        """Get the duration thresholds and their associated retry limits.
+
+        Returns:
+            A list of tuples (duration_threshold, retry_limit) sorted by duration
+        """
+        return [
+            (5.0, self.slow_test_retries_5s),  # ≤ 5 seconds tests
+            (10.0, self.slow_test_retries_10s),  # ≤ 10 seconds tests
+            (30.0, self.slow_test_retries_30s),  # ≤ 30 seconds tests
+            (300.0, self.slow_test_retries_5m),  # ≤ 5 minutes tests
+        ]
+
+    def get_max_retries_for_duration(self, duration_seconds: float) -> int:
+        """Get the maximum number of retries allowed for a test with the given duration.
+
+        Args:
+            duration_seconds: The duration of the test in seconds
+
+        Returns:
+            The maximum number of retries allowed for this test
+        """
+        for threshold, retries in self.get_threshold_definitions():
+            if duration_seconds <= threshold:
+                return retries
+        return 0  # No retries for tests longer than 5 minutes

ddtrace/internal/ci_visibility/api/_session.py

@@ -9,18 +9,16 @@
 from ddtrace.internal.ci_visibility.api._base import TestVisibilityParentItem
 from ddtrace.internal.ci_visibility.api._base import TestVisibilitySessionSettings
 from ddtrace.internal.ci_visibility.api._module import TestVisibilityModule
+from ddtrace.internal.ci_visibility.api._test_early_flake_detection import EarlyFlakeDetectionSessionHandler
 from ddtrace.internal.ci_visibility.constants import SESSION_ID
 from ddtrace.internal.ci_visibility.constants import SESSION_TYPE
 from ddtrace.internal.ci_visibility.constants import SUITE
 from ddtrace.internal.ci_visibility.constants import TEST
-from ddtrace.internal.ci_visibility.constants import TEST_EFD_ABORT_REASON
-from ddtrace.internal.ci_visibility.constants import TEST_EFD_ENABLED
 from ddtrace.internal.ci_visibility.constants import TEST_MANAGEMENT_ENABLED
 from ddtrace.internal.ci_visibility.telemetry.constants import EVENT_TYPES
 from ddtrace.internal.ci_visibility.telemetry.events import record_event_created
 from ddtrace.internal.ci_visibility.telemetry.events import record_event_finished
 from ddtrace.internal.logger import get_logger
-from ddtrace.internal.test_visibility._efd_mixins import EFDTestStatus
 
 
 log = get_logger(__name__)
@@ -47,6 +45,10 @@ def __init__(
         )
         self._test_command = self._session_settings.test_command
 
+        # Create EFD handler for session-level operations
+        self._efd_handler = EarlyFlakeDetectionSessionHandler.create(self, session_settings)
+
+        # Keep these for backward compatibility
         self._efd_abort_reason: Optional[str] = None
         self._efd_is_faulty_session: Optional[bool] = None
         self._efd_has_efd_failed_tests: bool = False
@@ -64,13 +66,14 @@ def get_session_settings(self) -> TestVisibilitySessionSettings:
         return self._session_settings
 
     def _set_efd_tags(self):
-        if self._session_settings.efd_settings.enabled:
-            self.set_tag(TEST_EFD_ENABLED, True)
-        if self._efd_abort_reason is not None:
-            # Allow any set abort reason to override faulty session abort reason
-            self.set_tag(TEST_EFD_ABORT_REASON, self._efd_abort_reason)
-        elif self.efd_is_faulty_session():
-            self.set_tag(TEST_EFD_ABORT_REASON, "faulty")
+        # Delegate to the EFD handler
+        self._efd_handler.set_tags()
+
+        # Update backward compatibility fields if needed
+        if self._efd_handler._abort_reason is not None:
+            self._efd_abort_reason = self._efd_handler._abort_reason
+        if self._efd_handler._is_faulty_session is not None:
+            self._efd_is_faulty_session = self._efd_handler._is_faulty_session
 
     def _set_test_management_tags(self):
         self.set_tag(TEST_MANAGEMENT_ENABLED, True)
@@ -117,51 +120,25 @@ def get_session(self):
     # EFD (Early Flake Detection) functionality
     #
     def efd_is_enabled(self):
-        return self._session_settings.efd_settings.enabled
+        """Check if EFD is enabled for this session."""
+        return self._efd_handler.is_enabled()
 
     def set_efd_abort_reason(self, abort_reason: str):
-        self._efd_abort_reason = abort_reason
+        """Set the reason for aborting EFD for this session."""
+        self._efd_handler.set_abort_reason(abort_reason)
+        self._efd_abort_reason = abort_reason  # For backward compatibility
 
     def efd_is_faulty_session(self):
-        """A session is considered "EFD faulty" if the percentage of tests considered new is greater than the
-        given threshold, and the total number of news tests exceeds the threshold.
-
-        NOTE: this behavior is cached on the assumption that this method will only be called once
-        """
-        if self._efd_is_faulty_session is not None:
-            return self._efd_is_faulty_session
-
-        if self._session_settings.efd_settings.enabled is False:
-            return False
-
-        total_tests_count = 0
-        new_tests_count = 0
-        for _module in self._children.values():
-            for _suite in _module._children.values():
-                for _test in _suite._children.values():
-                    total_tests_count += 1
-                    if _test.is_new():
-                        new_tests_count += 1
-
-        if new_tests_count <= self._session_settings.efd_settings.faulty_session_threshold:
-            return False
-
-        new_tests_pct = 100 * (new_tests_count / total_tests_count)
-
-        self._efd_is_faulty_session = new_tests_pct > self._session_settings.efd_settings.faulty_session_threshold
-
-        return self._efd_is_faulty_session
+        """Check if this is a faulty session for EFD purposes."""
+        result = self._efd_handler.is_faulty_session()
+        self._efd_is_faulty_session = result  # For backward compatibility
+        return result
 
     def efd_has_failed_tests(self):
-        if (not self._session_settings.efd_settings.enabled) or self.efd_is_faulty_session():
-            return False
-
-        for _module in self._children.values():
-            for _suite in _module._children.values():
-                for _test in _suite._children.values():
-                    if _test.efd_has_retries() and _test.efd_get_final_status() == EFDTestStatus.ALL_FAIL:
-                        return True
-        return False
+        """Check if the session has tests that failed in all EFD retries."""
+        result = self._efd_handler.has_failed_tests()
+        self._efd_has_efd_failed_tests = result  # For backward compatibility
+        return result
 
     #
     # ATR (Auto Test Retries , AKA Flaky Test Retries) functionality