chore(debugger): correct agent endpoint matching

The agent's /info endpoints list may or may not include a leading slash
depending on the agent version. The signal uploader only matched the
leading-slash form, so newer agents that advertise the debugger endpoints
without one were misreported as unsupported: snapshot and log uploads were
disabled and an "unsupported agent" warning was logged on every upload
cycle (as often as once per second).

Normalize the advertised endpoints before matching, consistent with remote
config and CI visibility. Also throttle the agent capability check so /info
is not polled every tick when the agent is unreachable or unsupported. The
throttle is charged only on those back-off paths (never on a supported
agent, so a transient upload failure recovers on the next tick) and is
reset on fork so a child re-checks immediately.

Author: tylfin

ddtrace/debugging/_uploader.py

@@ -118,24 +118,27 @@ def info_check(self, agent_info: Optional[dict[str, Any]]) -> bool:
             log.debug("Unsupported Datadog agent detected. Please upgrade to 7.49.0.")
             return False
 
-        endpoints = set(agent_info.get("endpoints", []))
+        # Agent /info entries may or may not carry a leading slash depending on
+        # the agent version, so normalize before matching (see remoteconfig).
+        endpoints = {endpoint.lstrip("/") for endpoint in agent_info.get("endpoints", [])}
 
         logs_track = self._tracks[SignalTrack.LOGS]
         snapshot_track = self._tracks[SignalTrack.SNAPSHOT]
         logs_track.enabled = True
         snapshot_track.enabled = True
 
-        if "/debugger/v2/input" in endpoints:
+        if "debugger/v2/input" in endpoints:
             log.debug("Detected /debugger/v2/input endpoint")
             logs_track.endpoint = f"/debugger/v2/input{self._endpoint_suffix}"
             snapshot_track.endpoint = f"/debugger/v2/input{self._endpoint_suffix}"
-        elif "/debugger/v1/diagnostics" in endpoints:
+        elif "debugger/v1/diagnostics" in endpoints:
             log.debug("Detected /debugger/v1/diagnostics endpoint fallback")
             logs_track.endpoint = f"/debugger/v1/diagnostics{self._endpoint_suffix}"
             snapshot_track.endpoint = f"/debugger/v1/diagnostics{self._endpoint_suffix}"
         else:
             logs_track.enabled = False
             snapshot_track.enabled = False
+            self._throttle_agent_check()
             log.warning(
                 UNSUPPORTED_AGENT,
                 extra={
@@ -180,6 +183,7 @@ def upload(self) -> None:
 
     def reset(self) -> None:
         """Reset the buffer on fork."""
+        super().reset()
         for track in self._tracks.values():
             track.queue = self.__queue__(encoder=track.queue._encoder, on_full=self._on_buffer_full)
         self._collector._tracks = {t: ut.queue for t, ut in self._tracks.items()}

ddtrace/internal/agent.py

@@ -7,12 +7,16 @@
 from ddtrace.internal.periodic import ForksafeAwakeablePeriodicService
 from ddtrace.internal.process_tags import compute_base_hash
 from ddtrace.internal.settings._agent import config
+from ddtrace.internal.utils.time import HourGlass
 
 from .utils.http import get_connection
 
 
 log = get_logger(__name__)
 
+# Bound how often /info is polled while the agent is unreachable or unsupported.
+AGENT_CHECK_INTERVAL = 60.0
+
 
 def process_info_headers(resp):
     try:
@@ -51,11 +55,24 @@ def __init__(self, interval: float = 0.0):
         super().__init__(interval=interval)
 
         self._state = self._agent_check
+        self._agent_check_throttle = HourGlass(duration=AGENT_CHECK_INTERVAL)
 
     @abc.abstractmethod
     def info_check(self, agent_info: t.Optional[dict]) -> bool: ...
 
+    def _throttle_agent_check(self) -> None:
+        """Back off from checking the agent until the throttle window elapses.
+
+        Only call this when the agent is unreachable or does not support the
+        product, never on a transient upload failure, so that recovery on a
+        supported agent is not delayed by ``AGENT_CHECK_INTERVAL``.
+        """
+        self._agent_check_throttle.turn()
+
     def _agent_check(self) -> None:
+        if self._agent_check_throttle.trickling():
+            return
+
         try:
             agent_info = info()
         except Exception:
@@ -64,6 +81,9 @@ def _agent_check(self) -> None:
         if self.info_check(agent_info):
             self._state = self._online
             self._online()
+        else:
+            # Unreachable or too old to advertise endpoints: back off.
+            self._throttle_agent_check()
 
     def _online(self) -> None:
         try:
@@ -72,6 +92,12 @@ def _online(self) -> None:
             self._state = self._agent_check
             log.debug("Error during online operation, reverting to agent check", exc_info=True)
 
+    def reset(self) -> None:
+        # On fork, re-check the agent immediately rather than inheriting the
+        # parent's throttle window.
+        super().reset()
+        self._agent_check_throttle = HourGlass(duration=AGENT_CHECK_INTERVAL)
+
     @abc.abstractmethod
     def online(self) -> None: ...
 

tests/debugging/test_uploader.py

@@ -114,16 +114,17 @@ def test_info_check_endpoint_selection():
     from ddtrace.debugging._signal.model import SignalTrack
     from ddtrace.debugging._uploader import SignalUploader
 
-    # Test 1: v2 endpoint available - both tracks use v2
+    # Test 1: v2 endpoint available (agent reports endpoints without a leading
+    # slash) - both tracks use v2
     uploader = SignalUploader(interval=LONG_INTERVAL)
-    agent_info = {"endpoints": ["/debugger/v2/input", "/debugger/v1/diagnostics"]}
+    agent_info = {"endpoints": ["debugger/v2/input", "debugger/v1/diagnostics"]}
     assert uploader.info_check(agent_info) is True
     assert uploader._tracks[SignalTrack.LOGS].enabled is True
     assert uploader._tracks[SignalTrack.SNAPSHOT].enabled is True
     assert "/debugger/v2/input" in uploader._tracks[SignalTrack.LOGS].endpoint
     assert "/debugger/v2/input" in uploader._tracks[SignalTrack.SNAPSHOT].endpoint
 
-    # Test 2: Only diagnostics available - both tracks fallback to diagnostics
+    # Test 2: Only diagnostics available (leading-slash form) - both tracks fallback
     uploader = SignalUploader(interval=LONG_INTERVAL)
     agent_info = {"endpoints": ["/debugger/v1/diagnostics"]}
     assert uploader.info_check(agent_info) is True
@@ -313,6 +314,62 @@ def test_online_raises_when_tracks_disabled():
         uploader.online()
 
 
+def test_agent_check_is_throttled():
+    # A missing/unsupported agent must not cause /info to be polled on every
+    # periodic tick; the check is throttled independently of the upload interval.
+    from ddtrace.internal import agent as agent_module
+
+    uploader = SignalUploader(interval=LONG_INTERVAL)
+    agent_info = {"endpoints": ["/some/other/endpoint"]}
+
+    with patch.object(agent_module, "info", return_value=agent_info) as mock_info:
+        uploader._agent_check()
+        uploader._agent_check()
+
+    assert mock_info.call_count == 1
+    assert uploader._state == uploader._agent_check
+
+
+def test_agent_check_recovers_after_online_failure():
+    # A transient upload failure after a healthy online cycle must not be
+    # throttled: the next tick should re-check the agent immediately.
+    from ddtrace.internal import agent as agent_module
+
+    uploader = SignalUploader(interval=LONG_INTERVAL)
+    agent_info = {"endpoints": ["debugger/v2/input"]}
+
+    with patch.object(agent_module, "info", return_value=agent_info) as mock_info:
+        # Healthy cycle: go online and clear the throttle.
+        uploader._agent_check()
+        assert uploader._state == uploader._online
+        assert mock_info.call_count == 1
+
+        # Simulate a transient upload failure reverting us to the agent check.
+        with patch.object(uploader, "online", side_effect=ValueError("transient")):
+            uploader._online()
+        assert uploader._state == uploader._agent_check
+
+        # Recovery must happen on the next tick, not after AGENT_CHECK_INTERVAL.
+        uploader._agent_check()
+        assert mock_info.call_count == 2
+        assert uploader._state == uploader._online
+
+
+def test_agent_check_throttle_reset_on_fork():
+    # A forked child must not inherit the parent's throttle window.
+    from ddtrace.internal import agent as agent_module
+
+    uploader = SignalUploader(interval=LONG_INTERVAL)
+    agent_info = {"endpoints": ["/some/other/endpoint"]}
+
+    with patch.object(agent_module, "info", return_value=agent_info):
+        uploader._agent_check()
+    assert uploader._agent_check_throttle.trickling() is True
+
+    uploader.reset()
+    assert uploader._agent_check_throttle.trickling() is False
+
+
 class _IsolatedUploader(MockSignalUploader):
     """Subclass to isolate class-level _products/_instance state from production code."""