chore: refactor tracer partial flushing logic

itr:noskip

This simplify the code and removes the edge case where we tag a complete trace chunk with unncessary partial flush tag

Author: brettlangdon

ddtrace/_trace/processor/__init__.py

@@ -263,9 +263,20 @@ def process_trace(self, trace: List[Span]) -> Optional[List[Span]]:
 
 
 class _Trace:
-    def __init__(self, spans=None, num_finished=0):
-        self.spans = spans if spans is not None else []
-        self.num_finished = num_finished
+    __slots__ = ("spans", "num_finished")
+
+    def __init__(self, spans: Optional[List[Span]] = None, num_finished: int = 0):
+        self.spans: List[Span] = spans if spans is not None else []
+        self.num_finished: int = num_finished
+
+    def remove_finished(self) -> List[Span]:
+        # perf: Avoid Span.finished which is a computed property and has function call overhead
+        #       so check Span.duration_ns manually.
+        finished = [s for s in self.spans if s.duration_ns is not None]
+        if finished:
+            self.spans[:] = [s for s in self.spans if s.duration_ns is None]
+            self.num_finished = 0
+        return finished
 
 
 class SpanAggregator(SpanProcessor):
@@ -350,31 +361,22 @@ def on_span_finish(self, span: Span) -> None:
                 return
 
             trace = self._traces[span.trace_id]
-            num_buffered = len(trace.spans)
             trace.num_finished += 1
-            should_partial_flush = self.partial_flush_enabled and trace.num_finished >= self.partial_flush_min_spans
-            is_trace_complete = trace.num_finished >= len(trace.spans)
-            if not is_trace_complete and not should_partial_flush:
-                return
-
-            if not is_trace_complete:
-                finished = [s for s in trace.spans if s.finished]
-                if not finished:
-                    return
-                trace.spans[:] = [s for s in trace.spans if not s.finished]  # In-place update
-                trace.num_finished = 0
-            else:
+            num_buffered = len(trace.spans)
+            is_trace_complete = trace.num_finished >= num_buffered
+            num_finished = trace.num_finished
+            should_partial_flush = False
+            if is_trace_complete:
                 finished = trace.spans
                 del self._traces[span.trace_id]
                 # perf: Flush span finish metrics to the telemetry writer after the trace is complete
                 self._queue_span_count_metrics("spans_finished", "integration_name")
-
-        num_finished = len(finished)
-        if should_partial_flush:
-            # FIXME(munir): should_partial_flush should return false if all the spans in the trace are finished.
-            # For example if partial flushing min spans is 10 and the trace has 10 spans, the trace should
-            # not have a partial flush metric. This trace was processed in its entirety.
-            finished[0].set_metric("_dd.py.partial_flush", num_finished)
+            elif self.partial_flush_enabled and num_finished >= self.partial_flush_min_spans:
+                should_partial_flush = True
+                finished = trace.remove_finished()
+                finished[0].set_metric("_dd.py.partial_flush", num_finished)
+            else:
+                return
 
         # perf: Process spans outside of the span aggregator lock
         spans = finished