chore(tracing): track spans_dropped telemetry for the trace writer pipeline

ddtrace/_trace/processor/__init__.py

@@ -347,6 +347,7 @@ def __init__(
         self._span_metrics: dict[str, defaultdict] = {
             "spans_created": defaultdict(int),
             "spans_finished": defaultdict(int),
+            "spans_dropped": defaultdict(int),
         }
         super(SpanAggregator, self).__init__()
 
@@ -423,10 +424,20 @@ def on_span_finish(self, span: Span) -> None:
         ):
             try:
                 spans = tp.process_trace(spans) or []
-                if not spans:
-                    return
             except Exception:
                 log.error("error applying processor %r to trace %d", tp, span.trace_id, exc_info=True)
+                continue
+            if not spans:
+                break
+
+        # Any spans removed by the trace processors (filtered out or the whole trace dropped)
+        # are definitively dropped here and never reach the writer. Batch them alongside the
+        # span created/finished counts.
+        dropped = len(finished) - len(spans)
+        if dropped > 0:
+            with self._lock:
+                self._span_metrics["spans_dropped"]["tp_drop"] += dropped
+                self._queue_span_count_metrics("spans_dropped", "reason")
 
         if spans:
             # Get sampling information from the root span
@@ -476,6 +487,9 @@ def shutdown(self, timeout: Optional[float]) -> None:
         # on_span_finish(...) queues span finish metrics in batches of 100.
         # This ensures all remaining counts are sent before the tracer is shutdown.
         self._queue_span_count_metrics("spans_finished", "integration_name", 1)
+        # on_span_finish(...) queues dropped span metrics in batches of 100.
+        # This ensures all remaining counts are sent before the tracer is shutdown.
+        self._queue_span_count_metrics("spans_dropped", "reason", 1)
         # Log a warning if the tracer is shutdown before spans are finished
         if log.isEnabledFor(logging.WARNING):
             unsent_spans = [
@@ -580,4 +594,5 @@ def reset(
             self._span_metrics = {
                 "spans_created": defaultdict(int),
                 "spans_finished": defaultdict(int),
+                "spans_dropped": defaultdict(int),
             }

ddtrace/_trace/telemetry.py

@@ -4,6 +4,18 @@
 from ddtrace.internal.telemetry.constants import TELEMETRY_NAMESPACE
 
 
+def record_trace_writer_metric(name: str, count: int, tags: Optional[tuple[tuple[str, str], ...]] = None) -> None:
+    """Record a TRACERS-namespace count metric emitted by the trace writer pipeline.
+
+    Used for span/trace_api metrics such as ``spans_enqueued_for_serialization``, ``spans_dropped``
+    (tagged with a ``reason``) and ``trace_api.requests``/``responses``/``errors``. Non-positive
+    counts are ignored so callers can pass raw span counts without guarding.
+    """
+    if count <= 0:
+        return
+    telemetry_writer.add_count_metric(TELEMETRY_NAMESPACE.TRACERS, name, count, tags=tags)
+
+
 def record_span_pointer_calculation(context: str, span_pointer_count: int) -> None:
     telemetry_writer.add_count_metric(
         namespace=TELEMETRY_NAMESPACE.TRACERS,

ddtrace/internal/ci_visibility/writer.py

@@ -123,6 +123,9 @@ class CIVisibilityWriter(HTTPWriter):
     RETRY_ATTEMPTS = 5
     HTTP_METHOD = "POST"
     STATSD_NAMESPACE = "civisibility.writer"
+    # CI Visibility uploads reuse the HTTPWriter pipeline but are not trace writers, so they must not
+    # emit tracer instrumentation telemetry (spans_enqueued/spans_dropped/trace_api.*).
+    _records_trace_telemetry = False
 
     def __init__(
         self,

ddtrace/internal/encoding.py

@@ -188,6 +188,7 @@ def __init__(self, max_size: int, max_item_size: int) -> None:
     def _reset(self) -> None:
         self._buffer = bytearray(self._PREFIX)
         self._count = 0
+        self._n_spans = 0
 
     def __len__(self) -> int:
         with self._lock:
@@ -198,6 +199,17 @@ def size(self) -> int:
         with self._lock:
             return len(self._buffer) + len(self._SUFFIX)
 
+    @property
+    def pending_spans(self) -> int:
+        """Number of spans currently buffered (across all buffered traces).
+
+        Used by the writer to report ``spans_dropped`` telemetry when a payload cannot be
+        encoded or delivered, since the buffer/encode/HTTP layers otherwise only know the
+        trace (chunk) count.
+        """
+        with self._lock:
+            return self._n_spans
+
     def put(self, item) -> None:
         item = typing.cast(list["Span"], item)
 
@@ -222,6 +234,7 @@ def put(self, item) -> None:
                 self._buffer += self._SEPARATOR
             self._buffer += encoded_trace
             self._count += 1
+            self._n_spans += len(item)
 
     def encode(self) -> list[tuple[Optional[bytes], int]]:
         with self._lock:

ddtrace/internal/writer/writer.py

@@ -2,6 +2,7 @@
 import binascii
 from collections import defaultdict
 import gzip
+import socket
 import sys
 import threading
 from typing import TYPE_CHECKING
@@ -11,6 +12,7 @@
 from typing import TextIO
 
 from ddtrace import config
+from ddtrace._trace.telemetry import record_trace_writer_metric
 from ddtrace.internal.dist_computing.utils import in_ray_job
 from ddtrace.internal.hostname import get_hostname
 import ddtrace.internal.native as native
@@ -187,6 +189,10 @@ class HTTPWriter(periodic.PeriodicService, TraceWriter):
     RETRY_ATTEMPTS = 3
     HTTP_METHOD = "PUT"
     STATSD_NAMESPACE = "tracer"
+    # Whether this writer emits span/trace instrumentation telemetry (TRACERS namespace).
+    # CI Visibility reuses this HTTP pipeline for event/coverage uploads but is not a trace
+    # writer, so it opts out to avoid polluting tracer telemetry dashboards.
+    _records_trace_telemetry = True
 
     def __init__(
         self,
@@ -278,6 +284,12 @@ def _set_keep_rate(self, trace):
                 _KEEP_SPANS_RATE_KEY, 1.0 - self._drop_sma.get()
             )  # PERF: avoid setting via Span.set_metric
 
+    def _record_trace_telemetry(
+        self, name: str, count: int, tags: Optional[tuple[tuple[str, str], ...]] = None
+    ) -> None:
+        if self._records_trace_telemetry:
+            record_trace_writer_metric(name, count, tags)
+
     def _reset_connection(self) -> None:
         with self._conn_lck:
             if self._conn:
@@ -357,11 +369,22 @@ def _send_payload(self, payload: bytes, count: int, client: WriterClientBase) ->
         headers = self._get_finalized_headers(count, client)
 
         self._metrics_dist("http.requests")
+        # trace_api.* telemetry is intentionally recorded per HTTP attempt (including retries),
+        # unlike spans_dropped which is only recorded once a payload is definitively dropped.
+        self._record_trace_telemetry("trace_api.requests", 1)
+
+        try:
+            response = self._put(payload, headers, client, no_trace=True)
+        except Exception as e:
+            error_type = "timeout" if isinstance(e, (socket.timeout, TimeoutError)) else "network"
+            self._record_trace_telemetry("trace_api.errors", 1, (("type", error_type),))
+            raise
 
-        response = self._put(payload, headers, client, no_trace=True)
+        self._record_trace_telemetry("trace_api.responses", 1, (("status_code", str(response.status)),))
 
         if response.status >= 400:
             self._metrics_dist("http.errors", tags=["type:%s" % response.status])
+            self._record_trace_telemetry("trace_api.errors", 1, (("type", "status_code"),))
         else:
             self._metrics_dist("http.sent.bytes", len(payload))
             self._metrics["sent_traces"] += count
@@ -422,6 +445,7 @@ def _write_with_client(self, client: WriterClientBase, spans: Optional[list["Spa
             )
             self._metrics_dist("buffer.dropped.traces", 1, tags=["reason:t_too_big"])
             self._metrics_dist("buffer.dropped.bytes", payload_size, tags=["reason:t_too_big"])
+            self._record_trace_telemetry("spans_dropped", len(spans), (("reason", "serialization_error"),))
         except BufferFull as e:
             payload_size = e.args[0]
             _safelog(
@@ -435,11 +459,14 @@ def _write_with_client(self, client: WriterClientBase, spans: Optional[list["Spa
             )
             self._metrics_dist("buffer.dropped.traces", 1, tags=["reason:full"])
             self._metrics_dist("buffer.dropped.bytes", payload_size, tags=["reason:full"])
+            self._record_trace_telemetry("spans_dropped", len(spans), (("reason", "overfull_buffer"),))
         except NoEncodableSpansError:
             self._metrics_dist("buffer.dropped.traces", 1, tags=["reason:incompatible"])
+            self._record_trace_telemetry("spans_dropped", len(spans), (("reason", "serialization_error"),))
         else:
             self._metrics_dist("buffer.accepted.traces", 1)
             self._metrics_dist("buffer.accepted.spans", len(spans))
+            self._record_trace_telemetry("spans_enqueued_for_serialization", len(spans))
 
     def flush_queue(self, raise_exc: bool = False):
         try:
@@ -450,6 +477,10 @@ def flush_queue(self, raise_exc: bool = False):
 
     def _flush_queue_with_client(self, client: WriterClientBase, raise_exc: bool = False) -> None:
         n_traces = len(client.encoder)
+        # Snapshot the number of buffered spans before encoding so we can attribute spans_dropped
+        # if encoding fails. Encoders that do not track spans (e.g. the msgpack encoders) report 0,
+        # and a 0 count is a no-op in the telemetry layer.
+        n_spans = getattr(client.encoder, "pending_spans", 0)
         try:
             if not (encoded_traces := client.encoder.encode()):
                 return
@@ -458,14 +489,26 @@ def _flush_queue_with_client(self, client: WriterClientBase, raise_exc: bool = F
             # FIXME(munir): if client.encoder raises an Exception n_traces may not be accurate due to race conditions
             _safelog(log.error, "failed to encode trace with encoder %r", client.encoder, exc_info=True)
             self._metrics_dist("encoder.dropped.traces", n_traces)
+            self._record_trace_telemetry("spans_dropped", n_spans, (("reason", "serialization_error"),))
             return
 
+        # Per-payload span counts are only meaningful when the encoder produces a single payload
+        # (as the agentless JSON encoder does). With multiple payloads we cannot attribute a span
+        # count to an individual payload, so we report 0 (a no-op) for span-level drops.
+        payload_n_spans = n_spans if len(encoded_traces) == 1 else 0
         for payload in encoded_traces:
             encoded_data, n_traces = payload
-            self._flush_single_payload(encoded_data, n_traces, client=client, raise_exc=raise_exc)
+            self._flush_single_payload(
+                encoded_data, n_traces, client=client, n_spans=payload_n_spans, raise_exc=raise_exc
+            )
 
     def _flush_single_payload(
-        self, encoded: Optional[bytes], n_traces: int, client: WriterClientBase, raise_exc: bool = False
+        self,
+        encoded: Optional[bytes],
+        n_traces: int,
+        client: WriterClientBase,
+        n_spans: int = 0,
+        raise_exc: bool = False,
     ) -> None:
         if encoded is None:
             return
@@ -483,14 +526,18 @@ def _flush_single_payload(
             except Exception:
                 _safelog(log.error, "failed to compress traces with encoder %r", client.encoder, exc_info=True)
                 self._metrics_dist("encoder.dropped.traces", n_traces)
+                self._record_trace_telemetry("spans_dropped", n_spans, (("reason", "serialization_error"),))
                 return
 
         try:
-            self._send_payload_with_backoff(encoded, n_traces, client)
+            response = self._send_payload_with_backoff(encoded, n_traces, client)
         except Exception:
+            # All retries have been exhausted (network/timeout failures are retried until
+            # RETRY_ATTEMPTS). The payload is now definitively dropped, so it is safe to record it.
             self._metrics_dist("http.errors", tags=["type:err"])
             self._metrics_dist("http.dropped.bytes", len(encoded))
             self._metrics_dist("http.dropped.traces", n_traces)
+            self._record_trace_telemetry("spans_dropped", n_spans, (("reason", "api_error"),))
             if raise_exc:
                 raise
             else:
@@ -503,6 +550,11 @@ def _flush_single_payload(
                     exc_info=True,
                     extra={"send_to_telemetry": False},
                 )
+        else:
+            # An HTTP error status (>=400) is not retried (the backoff stops as soon as a Response
+            # is returned), so the payload is dropped after this single, definitive attempt.
+            if isinstance(response, Response) and response.status >= 400:
+                self._record_trace_telemetry("spans_dropped", n_spans, (("reason", "api_error"),))
         finally:
             self._metrics_dist("http.sent.bytes", len(encoded))
             self._metrics_dist("http.sent.traces", n_traces)

tests/tracer/test_processors.py

@@ -100,6 +100,59 @@ def process_trace(self, trace):
     assert span.get_tag("final_processor") == "user"
 
 
+def test_aggregator_records_tp_drop_when_processor_drops_trace():
+    """A trace processor dropping the whole trace accumulates spans_dropped(reason=tp_drop)."""
+
+    class DropProc(TraceProcessor):
+        def process_trace(self, trace):
+            return None
+
+    aggr = SpanAggregator(partial_flush_enabled=False, partial_flush_min_spans=0, user_processors=[DropProc()])
+    aggr.writer = DummyWriter()
+
+    span = Span("span", on_finish=[aggr.on_span_finish])
+    aggr.on_span_start(span)
+    span.finish()
+
+    assert aggr._span_metrics["spans_dropped"]["tp_drop"] == 1
+    assert aggr.writer.pop() == []
+
+
+def test_aggregator_records_tp_drop_for_partial_processor_drop():
+    """A trace processor dropping a subset of spans accumulates only the dropped count as tp_drop."""
+
+    class HalfDropProc(TraceProcessor):
+        def process_trace(self, trace):
+            return trace[:1]
+
+    aggr = SpanAggregator(partial_flush_enabled=False, partial_flush_min_spans=0, user_processors=[HalfDropProc()])
+    aggr.writer = DummyWriter()
+
+    parent = Span("parent", on_finish=[aggr.on_span_finish])
+    aggr.on_span_start(parent)
+    child = Span("child", on_finish=[aggr.on_span_finish])
+    child.trace_id = parent.trace_id
+    child.parent_id = parent.span_id
+    aggr.on_span_start(child)
+    child.finish()
+    parent.finish()
+
+    assert aggr._span_metrics["spans_dropped"]["tp_drop"] == 1
+
+
+def test_aggregator_does_not_record_tp_drop_when_nothing_dropped():
+    """When no spans are dropped by processors, spans_dropped is not accumulated."""
+    aggr = SpanAggregator(partial_flush_enabled=False, partial_flush_min_spans=0)
+    aggr.writer = DummyWriter()
+
+    span = Span("span", on_finish=[aggr.on_span_finish])
+    aggr.on_span_start(span)
+    span.finish()
+
+    assert aggr._span_metrics["spans_dropped"]["tp_drop"] == 0
+    assert aggr.writer.pop() == [span]
+
+
 def test_aggregator_reset_default_args():
     """
     Test that on reset, the aggregator recreates trace writer but not the sampling processor (by default).
@@ -503,6 +556,48 @@ def test_span_creation_metrics():
         )
 
 
+@pytest.mark.subprocess(
+    env={
+        "DD_CIVISIBILITY_ENABLED": "false",
+        "DD_INSTRUMENTATION_TELEMETRY_ENABLED": "true",
+    }
+)
+def test_span_dropped_metrics():
+    """Spans dropped by a trace processor are queued in batches of 100 and the remainder is sent on shutdown"""
+    import mock
+
+    from ddtrace.internal.telemetry.constants import TELEMETRY_NAMESPACE
+    from ddtrace.trace import TraceFilter
+    from ddtrace.trace import tracer
+
+    class DropAllFilter(TraceFilter):
+        def process_trace(self, trace):
+            return None
+
+    tracer.configure(trace_processors=[DropAllFilter()])
+
+    def dropped_calls(mock_tm):
+        # spans_dropped calls are interleaved with spans_created/spans_finished, so filter them out.
+        return [c for c in mock_tm.call_args_list if len(c.args) >= 2 and c.args[1] == "spans_dropped"]
+
+    with mock.patch("ddtrace.internal.telemetry.telemetry_writer.add_count_metric") as mock_tm:
+        for _ in range(250):
+            tracer.trace("span").finish()
+
+        # 250 dropped spans are flushed in batches of 100.
+        assert dropped_calls(mock_tm) == [
+            mock.call(TELEMETRY_NAMESPACE.TRACERS, "spans_dropped", 100, tags=(("reason", "tp_drop"),)),
+            mock.call(TELEMETRY_NAMESPACE.TRACERS, "spans_dropped", 100, tags=(("reason", "tp_drop"),)),
+        ]
+
+        mock_tm.reset_mock()
+        tracer.shutdown()
+        # The remaining 50 dropped spans are flushed on shutdown.
+        assert dropped_calls(mock_tm) == [
+            mock.call(TELEMETRY_NAMESPACE.TRACERS, "spans_dropped", 50, tags=(("reason", "tp_drop"),)),
+        ]
+
+
 def test_changing_tracer_sampler_changes_tracesamplingprocessor_sampler(tracer):
     """Changing the tracer sampler should change the sampling processor's sampler"""
     # get processor