chore(debugger): fix issue with re-initializing uploader tracks (#14635)

`set_track_endpoints` was recently introduced to protect against
scenarios where the agent version can change over time
(#14568).

Unfortunately, this introduced a bug where the track queue would be
reset such that the counter would look empty but the buffer would fill,
leading to unrecoverable BufferFull errors.

This fixes that by preserving the queue metadata when the cache is
reset.

refs: DEBUG-4516

Author: tylfin

ddtrace/debugging/_uploader.py

@@ -1,6 +1,7 @@
 from dataclasses import dataclass
 from enum import Enum
 from typing import Any
+from typing import Dict
 from typing import Optional
 from typing import Set
 from urllib.parse import quote
@@ -59,6 +60,7 @@ def __init__(self, interval: Optional[float] = None) -> None:
 
         self._agent_endpoints_cache: HourGlass = HourGlass(duration=60.0)
 
+        self._tracks: Dict[SignalTrack, UploaderTrack] = {}
         self.set_track_endpoints()
         self._headers = {
             "Content-type": "application/json; charset=utf-8",
@@ -102,18 +104,25 @@ def set_track_endpoints(self) -> None:
 
         endpoint_suffix = f"?ddtags={quote(di_config.tags)}" if di_config._tags_in_qs and di_config.tags else ""
 
-        self._tracks = {
-            SignalTrack.LOGS: UploaderTrack(
-                endpoint=f"/debugger/v1/input{endpoint_suffix}",
-                queue=self.__queue__(
-                    encoder=LogSignalJsonEncoder(di_config.service_name), on_full=self._on_buffer_full
+        # Only create the tracks if they don't exist to preserve the track queue metadata.
+        if not self._tracks:
+            self._tracks = {
+                SignalTrack.LOGS: UploaderTrack(
+                    endpoint=f"/debugger/v1/input{endpoint_suffix}",
+                    queue=self.__queue__(
+                        encoder=LogSignalJsonEncoder(di_config.service_name), on_full=self._on_buffer_full
+                    ),
                 ),
-            ),
-            SignalTrack.SNAPSHOT: UploaderTrack(
-                endpoint=f"{snapshot_track}{endpoint_suffix}",
-                queue=self.__queue__(encoder=SnapshotJsonEncoder(di_config.service_name), on_full=self._on_buffer_full),
-            ),
-        }
+                SignalTrack.SNAPSHOT: UploaderTrack(
+                    endpoint=f"{snapshot_track}{endpoint_suffix}",
+                    queue=self.__queue__(
+                        encoder=SnapshotJsonEncoder(di_config.service_name), on_full=self._on_buffer_full
+                    ),
+                ),
+            }
+        else:
+            self._tracks[SignalTrack.SNAPSHOT].endpoint = f"{snapshot_track}{endpoint_suffix}"
+
         self._collector = self.__collector__({t: ut.queue for t, ut in self._tracks.items()})
 
     def _write(self, payload: bytes, endpoint: str) -> None:

tests/debugging/test_uploader.py

@@ -69,3 +69,102 @@ def test_uploader_full_buffer():
         # wakeup to mimic next interval
         uploader.periodic()
         assert uploader.queue.qsize() == 0
+
+
+def test_uploader_preserves_queue_metadata_on_agent_endpoint_refresh():
+    """Test that track queue metadata is preserved when agent endpoints are refreshed."""
+    import mock
+
+    from ddtrace.debugging._signal.model import SignalTrack
+    from ddtrace.internal import agent
+
+    # Mock agent.info to return initial endpoints
+    initial_agent_info = {"endpoints": ["/debugger/v1/input", "/debugger/v1/diagnostics"]}
+    updated_agent_info = {"endpoints": ["/debugger/v1/input", "/debugger/v2/input"]}
+
+    with mock.patch.object(agent, "info", return_value=initial_agent_info):
+        uploader = MockLogsIntakeUploaderV1(interval=LONG_INTERVAL)
+
+        # Add some data to the queues
+        logs_queue = uploader._tracks[SignalTrack.LOGS].queue
+        snapshot_queue = uploader._tracks[SignalTrack.SNAPSHOT].queue
+
+        # Put some encoded data in the queues
+        logs_queue.put_encoded(None, "log_data".encode("utf-8"))
+        snapshot_queue.put_encoded(None, "snapshot_data".encode("utf-8"))
+
+        # Store queue references and verify they have data
+        original_logs_queue = logs_queue
+        original_snapshot_queue = snapshot_queue
+        original_logs_count = logs_queue.count
+        original_snapshot_count = snapshot_queue.count
+
+        assert original_logs_count > 0, "Logs queue should have data"
+        assert original_snapshot_count > 0, "Snapshot queue should have data"
+
+        # Force the cache to expire by mocking trickling to return False
+        with mock.patch.object(uploader._agent_endpoints_cache, "trickling", return_value=False):
+            # Mock agent.info to return updated endpoints (v2 instead of v1 diagnostics)
+            with mock.patch.object(agent, "info", return_value=updated_agent_info):
+                # This should trigger set_track_endpoints to refresh but preserve queue metadata
+                uploader.set_track_endpoints()
+
+        # Verify that the track queues are the same objects (not recreated)
+        assert uploader._tracks[SignalTrack.LOGS].queue is original_logs_queue
+        assert uploader._tracks[SignalTrack.SNAPSHOT].queue is original_snapshot_queue
+
+        # Verify that queue counts are preserved
+        assert uploader._tracks[SignalTrack.LOGS].queue.count == original_logs_count
+        assert uploader._tracks[SignalTrack.SNAPSHOT].queue.count == original_snapshot_count
+
+        # Verify that the endpoint was updated for snapshot track
+        assert "/debugger/v2/input" in uploader._tracks[SignalTrack.SNAPSHOT].endpoint
+
+        # Verify we can still flush without BufferFull errors
+        uploader.periodic()
+
+        # The data should have been uploaded
+        assert uploader.queue.qsize() == 2  # One payload for logs, one for snapshots
+
+
+def test_uploader_agent_endpoint_refresh_multiple_calls():
+    """Test that multiple calls to set_track_endpoints with cache expiry work correctly."""
+    import mock
+
+    from ddtrace.debugging._signal.model import SignalTrack
+    from ddtrace.internal import agent
+
+    agent_responses = [
+        {"endpoints": ["/debugger/v1/input"]},
+        {"endpoints": ["/debugger/v1/input", "/debugger/v1/diagnostics"]},
+        {"endpoints": ["/debugger/v1/input", "/debugger/v2/input"]},
+    ]
+
+    with mock.patch.object(agent, "info", return_value=agent_responses[0]):
+        uploader = MockLogsIntakeUploaderV1(interval=LONG_INTERVAL)
+
+        # Add data to track buffer state
+        snapshot_queue = uploader._tracks[SignalTrack.SNAPSHOT].queue
+        snapshot_queue.put_encoded(None, "test_data".encode("utf-8"))
+        original_count = snapshot_queue.count
+
+        # Track the original queue object
+        original_queue = snapshot_queue
+
+        # Simulate multiple agent endpoint updates
+        for i, agent_response in enumerate(agent_responses[1:], 1):
+            with mock.patch.object(uploader._agent_endpoints_cache, "trickling", return_value=False):
+                with mock.patch.object(agent, "info", return_value=agent_response):
+                    uploader.set_track_endpoints()
+
+            # Queue should be preserved across all updates
+            assert uploader._tracks[SignalTrack.SNAPSHOT].queue is original_queue
+            assert uploader._tracks[SignalTrack.SNAPSHOT].queue.count == original_count
+
+            # Add more data to ensure buffer state is maintained
+            snapshot_queue.put_encoded(None, f"test_data_{i}".encode("utf-8"))
+            original_count = snapshot_queue.count
+
+        # Final verification - queue should still be functional
+        uploader.periodic()
+        assert uploader.queue.qsize() > 0