chore(ci_visibility): add support for bitmap-based coverage data

ddtrace/contrib/pytest/_plugin_v2.py

@@ -36,12 +36,13 @@
 from ddtrace.ext.ci_visibility.api import is_ci_visibility_enabled
 from ddtrace.internal.ci_visibility.constants import SKIPPED_BY_ITR_REASON
 from ddtrace.internal.ci_visibility.telemetry.coverage import COVERAGE_LIBRARY
+from ddtrace.internal.ci_visibility.telemetry.coverage import record_code_coverage_empty
 from ddtrace.internal.ci_visibility.telemetry.coverage import record_code_coverage_finished
 from ddtrace.internal.ci_visibility.telemetry.coverage import record_code_coverage_started
 from ddtrace.internal.ci_visibility.utils import take_over_logger_stream_handler
 from ddtrace.internal.coverage.code import ModuleCodeCollector
 from ddtrace.internal.coverage.installer import install as install_coverage
-from ddtrace.internal.coverage.util import collapse_ranges
+from ddtrace.internal.coverage.lines import CoverageLines
 from ddtrace.internal.logger import get_logger
 
 
@@ -96,25 +97,13 @@ def _handle_collected_coverage(test_id, coverage_collector) -> None:
 
     if not test_covered_lines:
         log.debug("No covered lines found for test %s", test_id)
+        record_code_coverage_empty()
         return
 
-    # TODO: switch representation to bytearrays as part of new ITR coverage strategy
-    # This code is temporary / PoC
-
-    coverage_data: t.Dict[Path, t.List[t.Tuple[int, int]]] = {}
+    coverage_data: t.Dict[Path, CoverageLines] = {}
 
     for path_str, covered_lines in test_covered_lines.items():
-        file_path = Path(path_str)
-        if not file_path.is_absolute():
-            file_path = file_path.resolve()
-
-        sorted_lines = sorted(covered_lines)
-
-        collapsed_ranges = collapse_ranges(sorted_lines)
-        file_segments = []
-        for file_segment in collapsed_ranges:
-            file_segments.append((file_segment[0], file_segment[1]))
-        coverage_data[file_path] = file_segments
+        coverage_data[Path(path_str).absolute()] = covered_lines
 
     CISuite.add_coverage_data(test_id.parent_id, coverage_data)
 

ddtrace/ext/ci_visibility/api.py

@@ -21,7 +21,6 @@
 from typing import List
 from typing import NamedTuple
 from typing import Optional
-from typing import Tuple
 from typing import Type
 from typing import Union
 
@@ -43,6 +42,7 @@
 from ddtrace.ext.test import Status as TestStatus
 from ddtrace.internal import core
 from ddtrace.internal.codeowners import Codeowners
+from ddtrace.internal.coverage.lines import CoverageLines
 from ddtrace.internal.logger import get_logger
 
 
@@ -421,11 +421,11 @@ def was_item_skipped_by_itr(item_id: Union[CISuiteId, CITestId]) -> bool:
 
     class AddCoverageArgs(NamedTuple):
         item_id: Union[_CIVisibilityChildItemIdBase, _CIVisibilityRootItemIdBase]
-        coverage_data: Dict[Path, List[Tuple[int, int]]]
+        coverage_data: Dict[Path, CoverageLines]
 
     @staticmethod
     @_catch_and_log_exceptions
-    def add_coverage_data(item_id: Union[CISuiteId, CITestId], coverage_data: Dict[Path, List[Tuple[int, int]]]):
+    def add_coverage_data(item_id: Union[CISuiteId, CITestId], coverage_data: Dict[Path, CoverageLines]):
         log.debug("Adding coverage data for item %s: %s", item_id, coverage_data)
         core.dispatch("ci_visibility.item.add_coverage_data", (CIITRMixin.AddCoverageArgs(item_id, coverage_data),))
 

ddtrace/internal/ci_visibility/api/ci_base.py

@@ -9,7 +9,6 @@
 from typing import Generic
 from typing import List
 from typing import Optional
-from typing import Tuple
 from typing import TypeVar
 from typing import Union
 
@@ -34,6 +33,7 @@
 from ddtrace.internal.ci_visibility.telemetry.itr import record_itr_skipped
 from ddtrace.internal.ci_visibility.telemetry.itr import record_itr_unskippable
 from ddtrace.internal.constants import COMPONENT
+from ddtrace.internal.coverage.lines import CoverageLines
 from ddtrace.internal.logger import get_logger
 
 
@@ -431,15 +431,15 @@ def get_parent_span(self) -> Optional[Span]:
         return None
 
     @abc.abstractmethod
-    def add_coverage_data(self, coverage_data: Dict[Path, List[Tuple[int, int]]]) -> None:
+    def add_coverage_data(self, coverage_data: Dict[Path, CoverageLines]) -> None:
         pass
 
     @_require_span
     def _add_coverage_data_tag(self) -> None:
         if self._span is None:
             return
         if self._coverage_data:
-            self._span.set_tag_str(
+            self._span.set_struct_tag(
                 COVERAGE_TAG_NAME, self._coverage_data.build_payload(self._session_settings.workspace_path)
             )
 

ddtrace/internal/ci_visibility/api/ci_coverage_data.py

@@ -1,9 +1,9 @@
 from collections import defaultdict
-import json
 from pathlib import Path
 from typing import Dict
 from typing import List
-from typing import Tuple
+
+from ddtrace.internal.coverage.lines import CoverageLines
 
 
 try:
@@ -15,60 +15,41 @@
 
 class CoverageFilePayload(TypedDict):
     filename: str
-    segments: List[Tuple[int, int, int, int, int]]
-
-
-class CICoverageSegment:
-    """Container for coverage segment data
-
-    Columns and counts are currently unused in the current data model in ddtrace.internal.ci_visibility.coverage and
-    .encoder , so they are not included in this class.
-    """
-
-    def __init__(self, start_line: int, end_line: int):
-        self.start_line: int = start_line
-        self.end_line: int = end_line
+    bitmap: bytes
 
 
 class CICoverageData:
     """Container for coverage data for an item (suite or test)"""
 
     def __init__(self) -> None:
-        self._coverage_data: Dict[Path, List[CICoverageSegment]] = defaultdict(list)
+        self._coverage_data: Dict[Path, CoverageLines] = defaultdict(CoverageLines)
 
     def __bool__(self):
         return bool(self._coverage_data)
 
-    def add_coverage_segment(self, file_path: Path, start_line, end_line):
-        """Add a coverage segment to the coverage data"""
-        self._coverage_data[file_path.absolute()].append(CICoverageSegment(start_line, end_line))
-
-    def add_coverage_segments(self, segments: Dict[Path, List[Tuple[int, int]]]):
+    def add_covered_files(self, covered_files: Dict[Path, CoverageLines]):
         """Add coverage segments to the coverage data"""
-        for file_path, segment_data in segments.items():
-            for segment in segment_data:
-                self.add_coverage_segment(file_path, segment[0], segment[1])
+        for file_path, covered_lines in covered_files.items():
+            self._coverage_data[file_path.absolute()].update(covered_lines)
 
     def _build_payload(self, root_dir: Path) -> List[CoverageFilePayload]:
         """Generate a CI Visibility coverage payload
 
         Tuples are used here since JSON serializes tuples as lists.
         """
         coverage_data = []
-        for file_path, segments in self._coverage_data.items():
+        for file_path, covered_lines in self._coverage_data.items():
             try:
                 # Report relative path unless the file path is not relative to root_dir
                 # Paths are assumed to be absolute based on having been converted at instantiation / add time.
                 relative_path = file_path.relative_to(root_dir)
             except ValueError:
                 relative_path = file_path
-            segments_data = []
-            for segment in segments:
-                segments_data.append((segment.start_line, 0, segment.end_line, 0, -1))
-            file_payload: CoverageFilePayload = {"filename": str(relative_path), "segments": segments_data}
+            path_str = f"/{str(relative_path)}"
+            file_payload: CoverageFilePayload = {"filename": path_str, "bitmap": covered_lines.to_bytes()}
             coverage_data.append(file_payload)
         return coverage_data
 
-    def build_payload(self, root_dir: Path) -> str:
+    def build_payload(self, root_dir: Path) -> Dict[str, List[CoverageFilePayload]]:
         """Generate a CI Visibility coverage payload in JSON format"""
-        return json.dumps({"files": self._build_payload(root_dir)})
+        return {"files": self._build_payload(root_dir)}

ddtrace/internal/ci_visibility/api/ci_module.py

@@ -1,7 +1,5 @@
 from typing import Dict
-from typing import List
 from typing import Optional
-from typing import Tuple
 
 from ddtrace.ext import test
 from ddtrace.ext.ci_visibility.api import CIModuleId
@@ -83,5 +81,5 @@ def _telemetry_record_event_finished(self):
             test_framework=self._session_settings.test_framework_metric_name,
         )
 
-    def add_coverage_data(self, coverage_data: Dict[Path, List[Tuple[int, int]]]):
+    def add_coverage_data(self, *args, **kwargs):
         raise NotImplementedError("Coverage data cannot be added to modules.")

ddtrace/internal/ci_visibility/api/ci_session.py

@@ -1,9 +1,6 @@
-from pathlib import Path
 from typing import Any
 from typing import Dict
-from typing import List
 from typing import Optional
-from typing import Tuple
 
 from ddtrace.ext import test
 from ddtrace.ext.ci_visibility.api import CIModuleId
@@ -77,5 +74,5 @@ def _telemetry_record_event_finished(self):
             is_unsupported_ci=self._session_settings.is_unsupported_ci,
         )
 
-    def add_coverage_data(self, coverage_data: Dict[Path, List[Tuple[int, int]]]) -> None:
-        raise NotImplementedError("Coverage data cannot be added to sessions.")
+    def add_coverage_data(self, *args, **kwargs):
+        raise NotImplementedError("Coverage data cannot be added to modules.")

ddtrace/internal/ci_visibility/api/ci_suite.py

@@ -2,7 +2,6 @@
 from typing import Dict
 from typing import List
 from typing import Optional
-from typing import Tuple
 
 from ddtrace.ext import test
 from ddtrace.ext.ci_visibility.api import CISourceFileInfo
@@ -20,6 +19,7 @@
 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.coverage.lines import CoverageLines
 from ddtrace.internal.logger import get_logger
 
 
@@ -44,6 +44,10 @@ def __init__(
 
         self._coverage_data: CICoverageData = CICoverageData()
 
+    def __repr__(self) -> str:
+        module_name = self.parent.name if self.parent is not None else "none"
+        return f"{self.__class__.__name__}(name={self.name}, module={module_name})"
+
     def finish(self, force: bool = False, override_status: Optional[CITestStatus] = None) -> None:
         super().finish(force=force, override_status=override_status)
 
@@ -88,5 +92,5 @@ def _telemetry_record_event_finished(self):
             test_framework=self._session_settings.test_framework_metric_name,
         )
 
-    def add_coverage_data(self, coverage_data: Dict[Path, List[Tuple[int, int]]]) -> None:
-        self._coverage_data.add_coverage_segments(coverage_data)
+    def add_coverage_data(self, coverage_data: Dict[Path, CoverageLines]) -> None:
+        self._coverage_data.add_covered_files(coverage_data)

ddtrace/internal/ci_visibility/api/ci_test.py

@@ -2,7 +2,6 @@
 from typing import Dict
 from typing import List
 from typing import Optional
-from typing import Tuple
 
 from ddtrace.ext import test
 from ddtrace.ext.ci_visibility.api import CIExcInfo
@@ -17,6 +16,7 @@
 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.coverage.lines import CoverageLines
 from ddtrace.internal.logger import get_logger
 
 
@@ -60,7 +60,11 @@ def __init__(
         self._is_benchmark = None
 
     def __repr__(self) -> str:
-        return f"{self.__class__.__name__}(name={self.name}, parameters={self._parameters})"
+        suite_name = self.parent.name if self.parent is not None else "none"
+        module_name = self.parent.parent.name if self.parent is not None and self.parent.parent is not None else "none"
+        return "{}(name={}, suite={}, module={}, parameters={})".format(
+            self.__class__.__name__, self.name, suite_name, module_name, self._parameters
+        )
 
     def _get_hierarchy_tags(self) -> Dict[str, str]:
         return {
@@ -139,5 +143,5 @@ def make_early_flake_retry_from_test(self, original_test_id: CITestId, retry_num
             ),
         )
 
-    def add_coverage_data(self, coverage_data: Dict[Path, List[Tuple[int, int]]]) -> None:
-        self._coverage_data.add_coverage_segments(coverage_data)
+    def add_coverage_data(self, coverage_data: Dict[Path, CoverageLines]) -> None:
+        self._coverage_data.add_covered_files(coverage_data)

ddtrace/internal/ci_visibility/encoder.py

@@ -19,10 +19,13 @@
 from ddtrace.internal.ci_visibility.telemetry.payload import record_endpoint_payload_events_count
 from ddtrace.internal.ci_visibility.telemetry.payload import record_endpoint_payload_events_serialization_time
 from ddtrace.internal.encoding import JSONEncoderV2
+from ddtrace.internal.logger import get_logger
 from ddtrace.internal.utils.time import StopWatch
 from ddtrace.internal.writer.writer import NoEncodableSpansError
 
 
+log = get_logger(__name__)
+
 if TYPE_CHECKING:  # pragma: no cover
     from typing import Any  # noqa:F401
     from typing import Dict  # noqa:F401
@@ -158,7 +161,11 @@ def _set_itr_suite_skipping_mode(self, new_value):
         self.itr_suite_skipping_mode = new_value
 
     def put(self, spans):
-        spans_with_coverage = [span for span in spans if COVERAGE_TAG_NAME in span.get_tags()]
+        spans_with_coverage = [
+            span
+            for span in spans
+            if COVERAGE_TAG_NAME in span.get_tags() or span.get_struct_tag(COVERAGE_TAG_NAME) is not None
+        ]
         if not spans_with_coverage:
             raise NoEncodableSpansError()
         return super(CIVisibilityCoverageEncoderV02, self).put(spans_with_coverage)
@@ -194,7 +201,10 @@ def _build_body(self, data):
     def _build_data(self, traces):
         # type: (List[List[Span]]) -> Optional[bytes]
         normalized_covs = [
-            self._convert_span(span, "") for trace in traces for span in trace if COVERAGE_TAG_NAME in span.get_tags()
+            self._convert_span(span, "")
+            for trace in traces
+            for span in trace
+            if (COVERAGE_TAG_NAME in span.get_tags() or span.get_struct_tag(COVERAGE_TAG_NAME) is not None)
         ]
         if not normalized_covs:
             return None
@@ -211,13 +221,23 @@ def _build_payload(self, traces):
 
     def _convert_span(self, span, dd_origin):
         # type: (Span, str) -> Dict[str, Any]
+        files: Dict[str, Any] = {}
+
+        files_struct_tag_value = span.get_struct_tag(COVERAGE_TAG_NAME)
+        if files_struct_tag_value is not None and "files" in files_struct_tag_value:
+            files = files_struct_tag_value["files"]
+        elif COVERAGE_TAG_NAME in span.get_tags():
+            files = json.loads(str(span.get_tag(COVERAGE_TAG_NAME)))["files"]
+
         converted_span = {
             "test_session_id": int(span.get_tag(SESSION_ID) or "1"),
             "test_suite_id": int(span.get_tag(SUITE_ID) or "1"),
-            "files": json.loads(str(span.get_tag(COVERAGE_TAG_NAME)))["files"],
+            "files": files,
         }
 
         if not self.itr_suite_skipping_mode:
             converted_span["span_id"] = span.span_id
 
+        log.debug("Span converted to coverage event: %s", converted_span)
+
         return converted_span