fix(ci_visibility): avoid clash with other sys.monitoring tools (#12558)

On Python 3.12+, CI Visibility / Test Optimization uses `sys.monitoring`
to gather code coverage information. The problem is that if other tools,
such as `pyest-cov`, also try to register a `sys.monitoring` tool for
coverage, an exception will be raised because Python only allows one
handler per tool type (see issue
#11256). This PR postpones
tool registering until the first call to `instrument_all_lines()`,
giving other tools the opportunity to register themselves; in this case,
however, we do not gather code coverage information.

- [x] PR author has checked that all the criteria below are met
- The PR description includes an overview of the change
- The PR description articulates the motivation for the change
- The change includes tests OR the PR description describes a testing
strategy
- The PR description notes risks associated with the change, if any
- Newly-added code is easy to change
- The change follows the [library release note
guidelines](https://ddtrace.readthedocs.io/en/stable/releasenotes.html)
- The change includes or references documentation updates if necessary
- Backport labels are set (if
[applicable](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting))

- [x] Reviewer has checked that all the criteria below are met
- Title is accurate
- All changes are related to the pull request's stated goal
- Avoids breaking
[API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces)
changes
- Testing strategy adequately addresses listed risks
- Newly-added code is easy to change
- Release note makes sense to a user of the library
- If necessary, author has acknowledged and discussed the performance
implications of this PR as reported in the benchmarks PR comment
- Backport labels are set in a manner that is consistent with the
[release branch maintenance
policy](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting)

(cherry picked from commit 7074375)

Author: vitor-de-araujo

ddtrace/internal/coverage/instrumentation_py3_12.py

@@ -4,9 +4,13 @@
 import typing as t
 
 from ddtrace.internal.injection import HookType
+from ddtrace.internal.logger import get_logger
 from ddtrace.internal.test_visibility.coverage_lines import CoverageLines
 
 
+log = get_logger(__name__)
+
+
 # This is primarily to make mypy happy without having to nest the rest of this module behind a version check
 assert sys.version_info >= (3, 12)  # nosec
 
@@ -17,126 +21,135 @@
 RETURN_CONST = dis.opmap["RETURN_CONST"]
 EMPTY_MODULE_BYTES = bytes([RESUME, 0, RETURN_CONST, 0])
 
-# Register the coverage tool with the low-impact monitoring system
-try:
-    sys.monitoring.use_tool_id(sys.monitoring.COVERAGE_ID, "datadog")  # noqa
-except ValueError:
-    # TODO: Another coverage tool is already in use. Either warn the user
-    # or free the tool and register ours.
-    def instrument_all_lines(
-        code: CodeType, hook: HookType, path: str, package: str
-    ) -> t.Tuple[CodeType, CoverageLines]:
-        # No-op
+_CODE_HOOKS: t.Dict[CodeType, t.Tuple[HookType, str, t.Dict[int, t.Tuple[str, t.Optional[t.Tuple[str]]]]]] = {}
+
+
+def instrument_all_lines(code: CodeType, hook: HookType, path: str, package: str) -> t.Tuple[CodeType, CoverageLines]:
+    coverage_tool = sys.monitoring.get_tool(sys.monitoring.COVERAGE_ID)
+    if coverage_tool is not None and coverage_tool != "datadog":
+        log.debug("Coverage tool '%s' already registered, not gathering coverage", coverage_tool)
         return code, CoverageLines()
 
-else:
-    _CODE_HOOKS: t.Dict[CodeType, t.Tuple[HookType, str, t.Dict[int, t.Tuple[str, t.Optional[t.Tuple[str]]]]]] = {}
+    if coverage_tool is None:
+        log.debug("Registering code coverage tool")
+        _register_monitoring()
+
+    return _instrument_all_lines_with_monitoring(code, hook, path, package)
+
+
+def _line_event_handler(code: CodeType, line: int) -> t.Any:
+    hook, path, import_names = _CODE_HOOKS[code]
+    import_name = import_names.get(line, None)
+    return hook((line, path, import_name))
+
 
-    def _line_event_handler(code: CodeType, line: int) -> t.Any:
-        hook, path, import_names = _CODE_HOOKS[code]
-        import_name = import_names.get(line, None)
-        return hook((line, path, import_name))
+def _register_monitoring():
+    """
+    Register the coverage tool with the low-impact monitoring system.
+    """
+    sys.monitoring.use_tool_id(sys.monitoring.COVERAGE_ID, "datadog")
 
     # Register the line callback
     sys.monitoring.register_callback(
         sys.monitoring.COVERAGE_ID, sys.monitoring.events.LINE, _line_event_handler
     )  # noqa
 
-    def instrument_all_lines(
-        code: CodeType, hook: HookType, path: str, package: str
-    ) -> t.Tuple[CodeType, CoverageLines]:
-        # Enable local line events for the code object
-        sys.monitoring.set_local_events(sys.monitoring.COVERAGE_ID, code, sys.monitoring.events.LINE)  # noqa
-
-        # Collect all the line numbers in the code object
-        linestarts = dict(dis.findlinestarts(code))
-
-        lines = CoverageLines()
-        import_names: t.Dict[int, t.Tuple[str, t.Optional[t.Tuple[str, ...]]]] = {}
-
-        # The previous two arguments are kept in order to track the depth of the IMPORT_NAME
-        # For example, from ...package import module
-        current_arg: int = 0
-        previous_arg: int = 0
-        _previous_previous_arg: int = 0
-        current_import_name: t.Optional[str] = None
-        current_import_package: t.Optional[str] = None
-
-        line = 0
-
-        ext: list[bytes] = []
-        code_iter = iter(enumerate(code.co_code))
-        try:
-            while True:
-                offset, opcode = next(code_iter)
-                _, arg = next(code_iter)
-
-                if opcode == RESUME:
-                    continue
-
-                if offset in linestarts:
-                    line = linestarts[offset]
-                    lines.add(line)
-
-                    # Make sure that the current module is marked as depending on its own package by instrumenting the
-                    # first executable line
-                    if code.co_name == "<module>" and len(lines) == 1 and package is not None:
-                        import_names[line] = (package, ("",))
-
-                if opcode is EXTENDED_ARG:
-                    ext.append(arg)
-                    continue
+
+def _instrument_all_lines_with_monitoring(
+    code: CodeType, hook: HookType, path: str, package: str
+) -> t.Tuple[CodeType, CoverageLines]:
+    # Enable local line events for the code object
+    sys.monitoring.set_local_events(sys.monitoring.COVERAGE_ID, code, sys.monitoring.events.LINE)  # noqa
+
+    # Collect all the line numbers in the code object
+    linestarts = dict(dis.findlinestarts(code))
+
+    lines = CoverageLines()
+    import_names: t.Dict[int, t.Tuple[str, t.Optional[t.Tuple[str, ...]]]] = {}
+
+    # The previous two arguments are kept in order to track the depth of the IMPORT_NAME
+    # For example, from ...package import module
+    current_arg: int = 0
+    previous_arg: int = 0
+    _previous_previous_arg: int = 0
+    current_import_name: t.Optional[str] = None
+    current_import_package: t.Optional[str] = None
+
+    line = 0
+
+    ext: list[bytes] = []
+    code_iter = iter(enumerate(code.co_code))
+    try:
+        while True:
+            offset, opcode = next(code_iter)
+            _, arg = next(code_iter)
+
+            if opcode == RESUME:
+                continue
+
+            if offset in linestarts:
+                line = linestarts[offset]
+                lines.add(line)
+
+                # Make sure that the current module is marked as depending on its own package by instrumenting the
+                # first executable line
+                if code.co_name == "<module>" and len(lines) == 1 and package is not None:
+                    import_names[line] = (package, ("",))
+
+            if opcode is EXTENDED_ARG:
+                ext.append(arg)
+                continue
+            else:
+                _previous_previous_arg = previous_arg
+                previous_arg = current_arg
+                current_arg = int.from_bytes([*ext, arg], "big", signed=False)
+                ext.clear()
+
+            if opcode == IMPORT_NAME:
+                import_depth: int = code.co_consts[_previous_previous_arg]
+                current_import_name: str = code.co_names[current_arg]
+                # Adjust package name if the import is relative and a parent (ie: if depth is more than 1)
+                current_import_package: str = (
+                    ".".join(package.split(".")[: -import_depth + 1]) if import_depth > 1 else package
+                )
+
+                if line in import_names:
+                    import_names[line] = (
+                        current_import_package,
+                        tuple(list(import_names[line][1]) + [current_import_name]),
+                    )
                 else:
-                    _previous_previous_arg = previous_arg
-                    previous_arg = current_arg
-                    current_arg = int.from_bytes([*ext, arg], "big", signed=False)
-                    ext.clear()
-
-                if opcode == IMPORT_NAME:
-                    import_depth: int = code.co_consts[_previous_previous_arg]
-                    current_import_name: str = code.co_names[current_arg]
-                    # Adjust package name if the import is relative and a parent (ie: if depth is more than 1)
-                    current_import_package: str = (
-                        ".".join(package.split(".")[: -import_depth + 1]) if import_depth > 1 else package
+                    import_names[line] = (current_import_package, (current_import_name,))
+
+            # Also track import from statements since it's possible that the "from" target is a module, eg:
+            # from my_package import my_module
+            # Since the package has not changed, we simply extend the previous import names with the new value
+            if opcode == IMPORT_FROM:
+                import_from_name = f"{current_import_name}.{code.co_names[current_arg]}"
+                if line in import_names:
+                    import_names[line] = (
+                        current_import_package,
+                        tuple(list(import_names[line][1]) + [import_from_name]),
                     )
+                else:
+                    import_names[line] = (current_import_package, (import_from_name,))
+
+    except StopIteration:
+        pass
+
+    # Recursively instrument nested code objects
+    for nested_code in (_ for _ in code.co_consts if isinstance(_, CodeType)):
+        _, nested_lines = instrument_all_lines(nested_code, hook, path, package)
+        lines.update(nested_lines)
+
+    # Register the hook and argument for the code object
+    _CODE_HOOKS[code] = (hook, path, import_names)
+
+    # Special case for empty modules (eg: __init__.py ):
+    # Make sure line 0 is marked as executable, and add package dependency
+    if not lines and code.co_name == "<module>" and code.co_code == EMPTY_MODULE_BYTES:
+        lines.add(0)
+        if package is not None:
+            import_names[0] = (package, ("",))
 
-                    if line in import_names:
-                        import_names[line] = (
-                            current_import_package,
-                            tuple(list(import_names[line][1]) + [current_import_name]),
-                        )
-                    else:
-                        import_names[line] = (current_import_package, (current_import_name,))
-
-                # Also track import from statements since it's possible that the "from" target is a module, eg:
-                # from my_package import my_module
-                # Since the package has not changed, we simply extend the previous import names with the new value
-                if opcode == IMPORT_FROM:
-                    import_from_name = f"{current_import_name}.{code.co_names[current_arg]}"
-                    if line in import_names:
-                        import_names[line] = (
-                            current_import_package,
-                            tuple(list(import_names[line][1]) + [import_from_name]),
-                        )
-                    else:
-                        import_names[line] = (current_import_package, (import_from_name,))
-
-        except StopIteration:
-            pass
-
-        # Recursively instrument nested code objects
-        for nested_code in (_ for _ in code.co_consts if isinstance(_, CodeType)):
-            _, nested_lines = instrument_all_lines(nested_code, hook, path, package)
-            lines.update(nested_lines)
-
-        # Register the hook and argument for the code object
-        _CODE_HOOKS[code] = (hook, path, import_names)
-
-        # Special case for empty modules (eg: __init__.py ):
-        # Make sure line 0 is marked as executable, and add package dependency
-        if not lines and code.co_name == "<module>" and code.co_code == EMPTY_MODULE_BYTES:
-            lines.add(0)
-            if package is not None:
-                import_names[0] = (package, ("",))
-
-        return code, lines
+    return code, lines

releasenotes/notes/ci_visibility-fix-clash-sys-monitoring-bd3ef705324f9f23.yaml

@@ -0,0 +1,5 @@
+---
+fixes:
+  - |
+    CI Visibility: This fix resolves an issue where ddtrace's own sys.monitoring coverage tool in Python 3.12+ would
+    block other sys.monitoring tools such as pytest-cov from being used.

tests/coverage/test_coverage_tool_clash.py

@@ -0,0 +1,50 @@
+import sys
+
+import pytest
+
+
+@pytest.mark.skipif(sys.version_info < (3, 12), reason="sys.monitoring coverage is only used in Python 3.12+")
+@pytest.mark.subprocess
+def test_code_coverage_tool_clash():
+    """If another tool is already registered as the `sys.monitoring` coverage tool, do not collect coverage."""
+    import os
+    from pathlib import Path
+    import sys
+
+    from ddtrace.internal.coverage.code import ModuleCodeCollector
+    from ddtrace.internal.coverage.installer import install
+    from tests.coverage.utils import _get_relpath_dict
+
+    cwd_path = os.getcwd()
+    include_path = Path(cwd_path + "/tests/coverage/included_path/")
+
+    sys.monitoring.use_tool_id(sys.monitoring.COVERAGE_ID, "something_else")
+
+    install(include_paths=[include_path], collect_import_time_coverage=True)
+
+    from tests.coverage.included_path.async_code import call_async_function_and_report_line_number
+
+    ModuleCodeCollector.start_coverage()
+    line_number = call_async_function_and_report_line_number()
+    ModuleCodeCollector.stop_coverage()
+
+    executable = _get_relpath_dict(cwd_path, ModuleCodeCollector._instance.lines)
+    covered = _get_relpath_dict(cwd_path, ModuleCodeCollector._instance._get_covered_lines(include_imported=False))
+    covered_with_imports = _get_relpath_dict(
+        cwd_path, ModuleCodeCollector._instance._get_covered_lines(include_imported=True)
+    )
+
+    expected_executable = {
+        "tests/coverage/included_path/async_code.py": set(),
+    }
+    expected_covered = {}
+    expected_covered_with_imports = {}
+
+    assert (
+        executable == expected_executable
+    ), f"Executable lines mismatch: expected={expected_executable} vs actual={executable}"
+    assert covered == expected_covered, f"Covered lines mismatch: expected={expected_covered} vs actual={covered}"
+    assert (
+        covered_with_imports == expected_covered_with_imports
+    ), f"Covered lines with imports mismatch: expected={expected_covered_with_imports} vs actual={covered_with_imports}"
+    assert line_number == 7