Merge branch 'main' into 4.0-breaking-changes

Author: emmettbutler

ddtrace/internal/_unpatched.py

@@ -10,9 +10,13 @@
 # to get a reference to the right threading module.
 import threading as _threading  # noqa
 import gc as _gc  # noqa
-
 import sys
 
+threading_Lock = _threading.Lock
+threading_RLock = _threading.RLock
+threading_Event = _threading.Event
+
+
 previous_loaded_modules = frozenset(sys.modules.keys())
 from subprocess import Popen as unpatched_Popen  # noqa # nosec B404
 from os import close as unpatched_close  # noqa: F401, E402

ddtrace/internal/forksafe.py

@@ -5,12 +5,13 @@
 import functools
 import logging
 import os
-import threading
 import typing
 import weakref
 
 import wrapt
 
+from ddtrace.internal import _unpatched
+
 
 log = logging.getLogger(__name__)
 
@@ -138,13 +139,13 @@ def _reset_object(self):
         self.__wrapped__ = self._self_wrapped_class()
 
 
-def Lock() -> threading.Lock:
-    return ResetObject(threading.Lock)  # type: ignore
+def Lock() -> _unpatched.threading_Lock:
+    return ResetObject(_unpatched.threading_Lock)  # type: ignore
 
 
-def RLock() -> threading.RLock:
-    return ResetObject(threading.RLock)  # type: ignore
+def RLock() -> _unpatched.threading_RLock:
+    return ResetObject(_unpatched.threading_RLock)  # type: ignore
 
 
-def Event() -> threading.Event:
-    return ResetObject(threading.Event)  # type: ignore
+def Event() -> _unpatched.threading_Event:
+    return ResetObject(_unpatched.threading_Event)  # type: ignore

ddtrace/internal/ipc.py

@@ -1,4 +1,3 @@
-from contextlib import contextmanager
 import os
 import secrets
 import tempfile
@@ -57,9 +56,7 @@ class ReadLock(BaseUnixLock):
     class WriteLock(BaseUnixLock):
         __acquire_mode__ = fcntl.LOCK_EX
 
-    @contextmanager
-    def open_file(path, mode):
-        yield unpatched_open(path, mode)
+    open_file = unpatched_open
 
 except ModuleNotFoundError:
     # Availability: Windows
@@ -78,7 +75,7 @@ def release(self):
 
     ReadLock = WriteLock = BaseWinLock  # type: ignore
 
-    def open_file(path, mode):
+    def open_file(path, mode):  # type: ignore
         import _winapi
 
         # force all modes to be read/write binary
@@ -93,14 +90,17 @@ def open_file(path, mode):
         return unpatched_open(fd, mode)
 
 
-TMPDIR = Path(tempfile.gettempdir())
+try:
+    TMPDIR: typing.Optional[Path] = Path(tempfile.gettempdir())
+except FileNotFoundError:
+    TMPDIR = None
 
 
 class SharedStringFile:
     """A simple shared-file implementation for multiprocess communication."""
 
     def __init__(self) -> None:
-        self.filename: typing.Optional[str] = str(TMPDIR / secrets.token_hex(8))
+        self.filename: typing.Optional[str] = str(TMPDIR / secrets.token_hex(8)) if TMPDIR is not None else None
 
     def put(self, data: str) -> None:
         """Put a string into the file."""

releasenotes/notes/fix-ipc-open-file-ensure-closed-98d9707f56c416da.yaml

@@ -0,0 +1,4 @@
+---
+fixes:
+  - |
+    Prevent a potential ``ResourceWarning`` in multiprocess scenarios.

releasenotes/notes/fix-ipc-valid-tmp-dir-063d8abcc1835747.yaml

@@ -0,0 +1,4 @@
+---
+fixes:
+  - |
+    Prevent startup failure when a temporary directory is not available.

releasenotes/notes/forksafe-lock-06ca188ef798c1f8.yaml

@@ -0,0 +1,4 @@
+---
+fixes:
+  - |
+    core: This fix resolves an issue where forksafe locks used patched threading primitives from the profiling module, causing performance issues. The forksafe module now uses unpatched threading primitives (``Lock``, ``RLock``, ``Event``).

scripts/cformat.sh

@@ -42,7 +42,7 @@ enumerate_files() {
         for ext in "${extensions[@]}"; do
             find_conditions+=("-o" "-name" "$ext")
         done
-        unset 'find_conditions[-1]'
+        find_conditions=("${find_conditions[@]:1}")  # Remove first -o
         find "$BASE_DIR" -type f \( "${find_conditions[@]}" \)
     else
         git ls-files "${extensions[@]}"
@@ -52,7 +52,7 @@ enumerate_files() {
 # Script defaults
 UPDATE_MODE=false
 ENUM_ALL=false
-BASE_DIR=$(dirname "$(realpath "$0")")
+BASE_DIR=$(dirname "$(dirname "$(realpath "$0")")")
 CLANG_FORMAT=clang-format
 
 # NB: consumes the arguments
@@ -67,20 +67,21 @@ while (( "$#" )); do
         *)
             ;;
     esac
+    shift
 done
 
 # Environment variable overrides
 [[ -n "${CFORMAT_FIX:-}" ]] && UPDATE_MODE=true
 [[ -n "${CFORMAT_ALL:-}" ]] && ENUM_ALL=true
-[[ -n "${CFORMAT_BIN:-}" ]] && CLANG_FORMAT="$CLANG_FORMAT_BIN"
+[[ -n "${CFORMAT_BIN:-}" ]] && CLANG_FORMAT="$CFORMAT_BIN"
 
 if [[ "$UPDATE_MODE" == "true" ]]; then
     # Update mode: Format files in-place
     enumerate_files \
         | exclude_patterns \
         | while IFS= read -r file; do
+            echo "Formatting $file";
             ${CLANG_FORMAT} -i "$file"
-            echo "Formatting $file"
         done
 else
     # Check mode: Compare formatted output to existing files

scripts/run-tests

@@ -248,6 +248,45 @@ class TestRunner:
             print(f"Warning: Failed to get riot venvs for pattern '{pattern}': {e}")
             return []
 
+    def get_all_riot_venvs(self) -> List[RiotVenv]:
+        """Get all available riot venvs without pattern filtering."""
+        try:
+            venvs = []
+
+            # Get all venv instances without pattern filtering
+            for n, inst in enumerate(riotfile.venv.instances()):
+                if not inst.name:
+                    continue
+
+                # Extract package information from the instance
+                packages_info = ""
+                if hasattr(inst, 'pkgs') and inst.pkgs:
+                    all_packages = [f"{pkg}: {version}" for pkg, version in inst.pkgs.items()]
+                    packages_info = ", ".join(all_packages) if all_packages else "standard packages"
+
+                # Extract command from the instance
+                command = ""
+                if hasattr(inst, 'cmd'):
+                    command = str(inst.cmd)
+                elif hasattr(inst, 'command'):
+                    command = str(inst.command)
+
+                venvs.append(RiotVenv(
+                    number=n,
+                    hash=inst.short_hash if hasattr(inst, 'short_hash') else f"hash{n}",
+                    name=inst.name,
+                    python_version=str(inst.py._hint) if hasattr(inst, 'py') and hasattr(inst.py, '_hint') else "3.10",
+                    packages=packages_info,
+                    suite_name="",
+                    command=command
+                ))
+
+            return venvs
+
+        except Exception as e:
+            print(f"Warning: Failed to get all riot venvs: {e}")
+            return []
+
     def start_services(self, services: Set[str]) -> bool:
         """Start required Docker services."""
         if not services:
@@ -539,6 +578,25 @@ class TestRunner:
 
         return selected_venvs
 
+    def get_venvs_by_hash_direct(self, venv_hashes: List[str]) -> List[RiotVenv]:
+        """Get specific venvs by their hashes from all available venvs.
+
+        This method doesn't require suite information and is used when --venv is provided directly.
+        """
+        all_venvs = self.get_all_riot_venvs()
+        venv_hashes_set = set(venv_hashes)
+        matched_venvs = [venv for venv in all_venvs if venv.hash in venv_hashes_set]
+
+        if not matched_venvs:
+            return []
+
+        # Print info about found venvs
+        print(f"📌 Found {len(matched_venvs)} venv(s):")
+        for venv in matched_venvs:
+            print(f"   • {venv.hash}: {venv.name} ({venv.display_name})")
+
+        return matched_venvs
+
 
 def main():
     parser = argparse.ArgumentParser(
@@ -612,7 +670,19 @@ Examples:
 
     runner = TestRunner()
 
-    # Determine which files to check
+    # Special handling for --venv: skip all file discovery and suite matching
+    if args.venv:
+        print("🎯 Using directly specified venvs (skipping file/suite analysis)")
+        selected_venvs = runner.get_venvs_by_hash_direct(args.venv)
+        if not selected_venvs:
+            print(f"❌ No venvs found matching hashes: {', '.join(args.venv)}")
+            return 1
+        # When using --venv directly, skip service management
+        print(f"⚠️  Skipping service management (run manually if needed)")
+        success = runner.run_tests(selected_venvs, {}, riot_args=riot_args, dry_run=args.dry_run)
+        return 0 if success else 1
+
+    # Normal flow: determine which files to check
     if args.files:
         # Use explicitly provided files
         files = set(args.files)
@@ -646,17 +716,8 @@ Examples:
         runner.output_suites_json(matching_suites)
         return 0
 
-    # Determine venv selection method
-    if args.venv:
-        # Use provided venvs (no interactive prompts)
-        selected_venvs = runner.select_venvs_by_hash(matching_suites, args.venv)
-        if not selected_venvs:
-            print(f"❌ No venvs found matching hashes: {', '.join(args.venv)}")
-            return 1
-        print(f"📌 Selected {len(selected_venvs)} venv(s) from provided hashes")
-    else:
-        # Interactive venv selection
-        selected_venvs = runner.interactive_venv_selection(matching_suites)
+    # Interactive venv selection
+    selected_venvs = runner.interactive_venv_selection(matching_suites)
 
     # Execute tests
     success = runner.run_tests(selected_venvs, matching_suites, riot_args=riot_args, dry_run=args.dry_run)

tests/internal/test_forksafe.py

@@ -1,5 +1,6 @@
 from collections import Counter
 import os
+import sys
 
 import pytest
 
@@ -204,6 +205,37 @@ def test_lock_fork():
     assert exit_code == 12
 
 
+@pytest.mark.skipif(sys.version_info >= (3, 14), reason="Profiling is not supported on Python 3.14 yet")
+@pytest.mark.subprocess(
+    env=dict(DD_PROFILING_ENABLED="1"),
+    ddtrace_run=True,
+)
+def test_lock_unpatched():
+    """Check that a forksafe.Lock is not patched when profiling is enabled."""
+
+    from ddtrace.internal import forksafe
+    from ddtrace.profiling import bootstrap
+    from ddtrace.profiling.collector.threading import ThreadingLockCollector
+
+    # When Profiler is started, bootstrap.profiler is set to the Profiler
+    # instance. We explicitly access the Profiler instance and the collector list
+    # to verify that the forksafe.Lock is not using the same class that is
+    # patched by ThreadingLockCollector that's running.
+    profiler = bootstrap.profiler._profiler
+    lock_collector = None
+    for c in profiler._collectors:
+        if isinstance(c, ThreadingLockCollector):
+            lock_collector = c
+            break
+
+    assert lock_collector is not None, "ThreadingLockCollector not found in profiler collectors"
+
+    lock = forksafe.Lock()
+    assert (
+        lock_collector._get_patch_target() is not lock._self_wrapped_class
+    ), "forksafe.Lock is using the same class that is patched by ThreadingLockCollector"
+
+
 def test_rlock_basic():
     # type: (...) -> None
     """Check that a forksafe.RLock implements the correct threading.RLock interface"""

tests/internal/test_tracer_flare.py

@@ -23,6 +23,7 @@
 from ddtrace.internal.flare.handler import _handle_tracer_flare
 from ddtrace.internal.logger import get_logger
 from ddtrace.internal.remoteconfig._connectors import PublisherSubscriberConnector
+from ddtrace.internal.utils.retry import fibonacci_backoff_with_jitter
 from tests.utils import remote_config_build_payload as build_payload
 
 
@@ -214,6 +215,7 @@ def test_json_logs(self):
         self.flare.clean_up_files()
         self.flare.revert_configs()
 
+    @fibonacci_backoff_with_jitter(attempts=5, initial_wait=0.1)
     def confirm_cleanup(self):
         assert not self.flare.flare_dir.exists(), f"The directory {self.flare.flare_dir} still exists"
         # Only check for file handler cleanup if prepare() was called