chore(debugger): send debugger snapshots to debugger track (#14402)

Send debugger snapshots to the debugger track. Logs will still go
through the default logs intake.

Refs: [DEBUG-4339](https://datadoghq.atlassian.net/browse/DEBUG-4339)

## Checklist
- [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))

## Reviewer Checklist
- [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)


[DEBUG-4338]:
https://datadoghq.atlassian.net/browse/DEBUG-4338?atlOrigin=eyJpIjoiNWRkNTljNzYxNjVmNDY3MDlhMDU5Y2ZhYzA5YTRkZjUiLCJwIjoiZ2l0aHViLWNvbS1KU1cifQ

[DEBUG-4339]:
https://datadoghq.atlassian.net/browse/DEBUG-4339?atlOrigin=eyJpIjoiNWRkNTljNzYxNjVmNDY3MDlhMDU5Y2ZhYzA5YTRkZjUiLCJwIjoiZ2l0aHViLWNvbS1KU1cifQ

---------

Co-authored-by: Tyler Finethy <[email protected]>

Author: P403n1x87

ddtrace/debugging/_encoding.py

@@ -249,8 +249,18 @@ def __init__(self, service: str, host: Optional[str] = None) -> None:
         self._service = service
         self._host = host
 
+    def _encode(self, item: LogSignal) -> str:
+        return json.dumps(_build_log_track_payload(self._service, item, self._host))
+
+    def encode(self, item: LogSignal) -> bytes:
+        return self._encode(item).encode("utf-8")
+
+
+class SnapshotJsonEncoder(LogSignalJsonEncoder):
+    """Encoder for snapshot signals, with automatic pruning of large snapshots."""
+
     def encode(self, item: LogSignal) -> bytes:
-        return self.pruned(json.dumps(_build_log_track_payload(self._service, item, self._host))).encode("utf-8")
+        return self.pruned(self._encode(item)).encode("utf-8")
 
     def pruned(self, log_signal_json: str) -> str:
         if len(log_signal_json) <= self.MAX_SIGNAL_SIZE:
@@ -310,24 +320,26 @@ def __init__(
     ) -> None:
         self._encoder = encoder
         self._buffer = JsonBuffer(buffer_size)
-        self._lock = forksafe.Lock()
+        self._lock = forksafe.RLock()
         self._on_full = on_full
         self.count = 0
         self.max_size = buffer_size - self._buffer.size
+        self._full = False
 
     def put(self, item: Snapshot) -> int:
         return self.put_encoded(item, self._encoder.encode(item))
 
     def put_encoded(self, item: Snapshot, encoded: bytes) -> int:
-        try:
-            with self._lock:
+        with self._lock:
+            try:
                 size = self._buffer.put(encoded)
                 self.count += 1
                 return size
-        except BufferFull:
-            if self._on_full is not None:
-                self._on_full(item, encoded)
-            raise
+            except BufferFull:
+                self._full = True
+                if self._on_full is not None:
+                    self._on_full(item, encoded)
+                raise
 
     def flush(self) -> Optional[Union[bytes, bytearray]]:
         with self._lock:
@@ -338,4 +350,9 @@ def flush(self) -> Optional[Union[bytes, bytearray]]:
 
             encoded = self._buffer.flush()
             self.count = 0
+            self._full = False
             return encoded
+
+    def is_full(self) -> bool:
+        with self._lock:
+            return self._full

ddtrace/debugging/_exception/replay.py

@@ -189,6 +189,8 @@ def build(cls, exc_id: uuid.UUID, frame: FrameType) -> "SpanExceptionProbe":
 
 @dataclass
 class SpanExceptionSnapshot(Snapshot):
+    __type__ = "er_snapshot"
+
     exc_id: t.Optional[uuid.UUID] = None
 
     @property

ddtrace/debugging/_probe/status.py

@@ -49,6 +49,7 @@ def _payload(
             "timestamp": int(timestamp * 1e3),  # milliseconds
             "message": message,
             "ddsource": "dd_debugger",
+            "type": "diagnostic",
             "debugger": {
                 "diagnostics": {
                     "probeId": probe.probe_id,

ddtrace/debugging/_signal/collector.py

@@ -1,6 +1,7 @@
 import os
 from typing import Any
 from typing import Callable
+from typing import Dict
 from typing import List
 from typing import Tuple
 
@@ -9,6 +10,7 @@
 from ddtrace.debugging._signal.log import LogSignal
 from ddtrace.debugging._signal.model import Signal
 from ddtrace.debugging._signal.model import SignalState
+from ddtrace.debugging._signal.model import SignalTrack
 from ddtrace.internal._encoding import BufferFull
 from ddtrace.internal.compat import ExcInfoType
 from ddtrace.internal.logger import get_logger
@@ -29,18 +31,20 @@ class SignalCollector(object):
     encoded, or the signal status indicate it should be skipped.
     """
 
-    def __init__(self, encoder: BufferedEncoder) -> None:
-        self._encoder = encoder
+    def __init__(self, tracks: Dict[SignalTrack, BufferedEncoder]) -> None:
+        self._tracks = tracks
 
     def _enqueue(self, log_signal: LogSignal) -> None:
         try:
             log.debug(
-                "[%s][P: %s] SignalCollector. _encoder (%s) _enqueue signal", os.getpid(), os.getppid(), self._encoder
+                "[%s][P: %s] SignalCollector enqueu signal on track %s", os.getpid(), os.getppid(), log_signal.__track__
             )
-            self._encoder.put(log_signal)
+            self._tracks[log_signal.__track__].put(log_signal)
         except BufferFull:
             log.debug("Encoder buffer full")
             meter.increment("encoder.buffer.full")
+        except KeyError:
+            log.error("No encoder for signal track %s", log_signal.__track__)
 
     def push(self, signal: Signal) -> None:
         if signal.state is SignalState.SKIP_COND:

ddtrace/debugging/_signal/log.py

@@ -5,6 +5,7 @@
 from ddtrace.debugging._probe.model import FunctionLocationMixin
 from ddtrace.debugging._probe.model import LineLocationMixin
 from ddtrace.debugging._signal.model import Signal
+from ddtrace.debugging._signal.model import SignalTrack
 
 
 @dataclass
@@ -16,6 +17,9 @@ class LogSignal(Signal):
     (e.g. conditions) might need to be reported.
     """
 
+    __type__ = "di_snapshot"
+    __track__: t.ClassVar[SignalTrack] = SignalTrack.LOGS
+
     @property
     @abc.abstractmethod
     def message(self) -> t.Optional[str]:
@@ -61,6 +65,7 @@ def snapshot(self) -> t.Dict[str, t.Any]:
             "evaluationErrors": [{"expr": e.expr, "message": e.message} for e in self.errors],
             "probe": self._probe_details(),
             "language": "python",
+            "type": self.__type__,
         }
         full_data.update(self.data)
 

ddtrace/debugging/_signal/model.py

@@ -50,6 +50,12 @@ class SignalState(str, Enum):
     DONE = "DONE"
 
 
+class SignalTrack(str, Enum):
+    DEFAULT = "default"
+    LOGS = "logs"
+    SNAPSHOT = "snapshot"
+
+
 @dataclass
 class Signal(abc.ABC):
     """Debugger signal base class.
@@ -72,6 +78,7 @@ class Signal(abc.ABC):
     """
 
     __default_timing__: ClassVar[ProbeEvalTiming] = ProbeEvalTiming.EXIT
+    __track__: ClassVar[SignalTrack] = SignalTrack.DEFAULT
 
     probe: Probe
     frame: FrameType

ddtrace/debugging/_signal/snapshot.py

@@ -6,6 +6,7 @@
 from types import FunctionType
 from types import ModuleType
 from typing import Any
+from typing import ClassVar
 from typing import Dict
 from typing import Mapping
 from typing import Optional
@@ -29,6 +30,7 @@
 from ddtrace.debugging._signal import utils
 from ddtrace.debugging._signal.log import LogSignal
 from ddtrace.debugging._signal.model import EvaluationError
+from ddtrace.debugging._signal.model import SignalTrack
 from ddtrace.debugging._signal.model import probe_to_signal
 from ddtrace.debugging._signal.utils import serialize
 from ddtrace.internal.compat import ExcInfoType
@@ -95,6 +97,8 @@ class Snapshot(LogSignal):
     Used to collect the minimum amount of information from a firing probe.
     """
 
+    __track__: ClassVar[SignalTrack] = SignalTrack.SNAPSHOT
+
     entry_capture: Optional[dict] = field(default=None)
     return_capture: Optional[dict] = field(default=None)
     line_capture: Optional[dict] = field(default=None)

ddtrace/debugging/_uploader.py

@@ -1,3 +1,4 @@
+from dataclasses import dataclass
 from enum import Enum
 from typing import Any
 from typing import Optional
@@ -7,8 +8,11 @@
 from ddtrace.debugging._config import di_config
 from ddtrace.debugging._encoding import LogSignalJsonEncoder
 from ddtrace.debugging._encoding import SignalQueue
+from ddtrace.debugging._encoding import SnapshotJsonEncoder
 from ddtrace.debugging._metrics import metrics
 from ddtrace.debugging._signal.collector import SignalCollector
+from ddtrace.debugging._signal.model import SignalTrack
+from ddtrace.internal import agent
 from ddtrace.internal.logger import get_logger
 from ddtrace.internal.periodic import ForksafeAwakeablePeriodicService
 from ddtrace.internal.utils.http import connector
@@ -27,6 +31,12 @@ class UploaderProduct(str, Enum):
     CODE_ORIGIN_SPAN = "code_origin.span"
 
 
+@dataclass
+class UploaderTrack:
+    endpoint: str
+    queue: SignalQueue
+
+
 class LogsIntakeUploaderV1(ForksafeAwakeablePeriodicService):
     """Logs intake uploader.
 
@@ -36,26 +46,48 @@ class LogsIntakeUploaderV1(ForksafeAwakeablePeriodicService):
 
     _instance: Optional["LogsIntakeUploaderV1"] = None
     _products: Set[UploaderProduct] = set()
+    _agent_endpoints: Set[str] = set()
 
     __queue__ = SignalQueue
     __collector__ = SignalCollector
 
-    ENDPOINT = di_config._intake_endpoint
-
     RETRY_ATTEMPTS = 3
 
     def __init__(self, interval: Optional[float] = None) -> None:
         super().__init__(interval if interval is not None else di_config.upload_interval_seconds)
 
-        self._queue = self.__queue__(encoder=LogSignalJsonEncoder(di_config.service_name), on_full=self._on_buffer_full)
-        self._collector = self.__collector__(self._queue)
+        endpoint_suffix = f"?ddtags={quote(di_config.tags)}" if di_config._tags_in_qs and di_config.tags else ""
+        if not self._agent_endpoints:
+            try:
+                agent_info = agent.info()
+                self._agent_endpoints = set(agent_info.get("endpoints", [])) if agent_info is not None else set()
+            except Exception:
+                pass  # nosec B110
+
+        snapshot_track = "/debugger/v1/input"
+        if "/debugger/v2/input" in self._agent_endpoints:
+            snapshot_track = "/debugger/v2/input"
+        elif "/debugger/v1/diagnostics" in self._agent_endpoints:
+            snapshot_track = "/debugger/v1/diagnostics"
+
+        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),
+            ),
+        }
+        self._collector = self.__collector__({t: ut.queue for t, ut in self._tracks.items()})
         self._headers = {
             "Content-type": "application/json; charset=utf-8",
             "Accept": "text/plain",
         }
 
-        if di_config._tags_in_qs and di_config.tags:
-            self.ENDPOINT += f"?ddtags={quote(di_config.tags)}"
         self._connect = connector(di_config._intake_url, timeout=di_config.upload_timeout)
 
         # Make it retry-able
@@ -65,33 +97,31 @@ def __init__(self, interval: Optional[float] = None) -> None:
         )(self._write)
 
         log.debug(
-            "Logs intake uploader initialized (url: %s, endpoint: %s, interval: %f)",
+            "Logs intake uploader initialized (url: %s, endpoints: %s, interval: %f)",
             di_config._intake_url,
-            self.ENDPOINT,
+            {t: ut.endpoint for t, ut in self._tracks.items()},
             self.interval,
         )
 
-    def _write(self, payload: bytes) -> None:
+        self._flush_full = False
+
+    def _write(self, payload: bytes, endpoint: str) -> None:
         try:
             with self._connect() as conn:
-                conn.request(
-                    "POST",
-                    self.ENDPOINT,
-                    payload,
-                    headers=self._headers,
-                )
+                conn.request("POST", endpoint, payload, headers=self._headers)
                 resp = conn.getresponse()
                 if not (200 <= resp.status < 300):
-                    log.error("Failed to upload payload: [%d] %r", resp.status, resp.read())
+                    log.error("Failed to upload payload to endpoint %s: [%d] %r", endpoint, resp.status, resp.read())
                     meter.increment("upload.error", tags={"status": str(resp.status)})
                 else:
                     meter.increment("upload.success")
                     meter.distribution("upload.size", len(payload))
         except Exception:
-            log.error("Failed to write payload", exc_info=True)
+            log.error("Failed to write payload to endpoint %s", endpoint, exc_info=True)
             meter.increment("error")
 
     def _on_buffer_full(self, _item: Any, _encoded: bytes) -> None:
+        self._flush_full = True
         self.upload()
 
     def upload(self) -> None:
@@ -100,20 +130,32 @@ def upload(self) -> None:
 
     def reset(self) -> None:
         """Reset the buffer on fork."""
-        self._queue = self.__queue__(encoder=self._queue._encoder, on_full=self._on_buffer_full)
-        self._collector._encoder = self._queue
+        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()}
+
+    def _flush_track(self, track: UploaderTrack) -> None:
+        queue = track.queue
+        payload = queue.flush()
+        if payload is not None:
+            try:
+                self._write_with_backoff(payload, track.endpoint)
+                meter.distribution("batch.cardinality", queue.count)
+            except Exception:
+                log.debug("Cannot upload logs payload", exc_info=True)
 
     def periodic(self) -> None:
         """Upload the buffer content to the logs intake."""
-        count = self._queue.count
-        if count:
-            payload = self._queue.flush()
-            if payload is not None:
-                try:
-                    self._write_with_backoff(payload)
-                    meter.distribution("batch.cardinality", count)
-                except Exception:
-                    log.debug("Cannot upload logs payload", exc_info=True)
+        if self._flush_full:
+            # We received the signal to flush a full buffer
+            self._flush_full = False
+            for track in self._tracks.values():
+                if track.queue.is_full():
+                    self._flush_track(track)
+
+        for track in self._tracks.values():
+            if track.queue.count:
+                self._flush_track(track)
 
     on_shutdown = periodic
 

ddtrace/internal/symbol_db/symbols.py

@@ -468,6 +468,7 @@ def __init__(self, scopes: t.Optional[t.List[Scope]] = None) -> None:
             "ddsource": "python",
             "service": config.service or DEFAULT_SERVICE_NAME,
             "runtimeId": get_runtime_id(),
+            "type": "symdb",
         }
 
     def add_scope(self, scope: Scope) -> None:

ddtrace/settings/dynamic_instrumentation.py

@@ -52,7 +52,6 @@ class DynamicInstrumentationConfig(DDConfig):
     max_probes = DDConfig.d(int, lambda _: DEFAULT_MAX_PROBES)
     global_rate_limit = DDConfig.d(float, lambda _: DEFAULT_GLOBAL_RATE_LIMIT)
     _tags_in_qs = DDConfig.d(bool, lambda _: True)
-    _intake_endpoint = DDConfig.d(str, lambda _: "/debugger/v1/input")
     tags = DDConfig.d(str, _derive_tags)
 
     enabled = DDConfig.v(