fix(tracing): do not raise exception if partial flush is triggered without any spans

ddtrace/_trace/processor/__init__.py

@@ -297,8 +297,7 @@ class _Trace(object):
         type=Dict[str, DefaultDict],
     )
 
-    def on_span_start(self, span):
-        # type: (Span) -> None
+    def on_span_start(self, span: Span) -> None:
         with self._lock:
             trace = self._traces[span.trace_id]
             trace.spans.append(span)
@@ -309,6 +308,17 @@ def on_span_finish(self, span):
         # type: (Span) -> None
         with self._lock:
             self._span_metrics["spans_finished"][span._span_api] += 1
+
+            # Calling finish on a span that we did not see the start for
+            # DEV: This can occur if the SpanAggregator is recreated while there is a span in progress
+            #      e.g. `tracer.configure()` is called after starting a span
+            if span.trace_id not in self._traces:
+                log_msg = "finished span not connected to a trace"
+                if config._telemetry_enabled:
+                    telemetry.telemetry_writer.add_log("WARN", log_msg)
+                log.warning("%s: %s", log_msg, span)
+                return
+
             trace = self._traces[span.trace_id]
             trace.num_finished += 1
             should_partial_flush = self._partial_flush_enabled and trace.num_finished >= self._partial_flush_min_spans
@@ -326,16 +336,27 @@ def on_span_finish(self, span):
                     finished = trace_spans
 
                 num_finished = len(finished)
+                trace.num_finished -= num_finished
+                if trace.num_finished != 0:
+                    log_msg = "unexpected finished span count"
+                    if config._telemetry_enabled:
+                        telemetry.telemetry_writer.add_log("WARN", log_msg)
+                    log.warning("%s (%s) for span %s", log_msg, num_finished, span)
+                    trace.num_finished = 0
+
+                # If we have removed all spans from this trace, then delete the trace from the traces dict
+                if len(trace.spans) == 0:
+                    del self._traces[span.trace_id]
+
+                # No spans to process, return early
+                if not finished:
+                    return
 
+                # Set partial flush tag on the first span
                 if should_partial_flush:
                     log.debug("Partially flushing %d spans for trace %d", num_finished, span.trace_id)
                     finished[0].set_metric("_dd.py.partial_flush", num_finished)
 
-                trace.num_finished -= num_finished
-
-                if len(trace.spans) == 0:
-                    del self._traces[span.trace_id]
-
                 spans = finished  # type: Optional[List[Span]]
                 for tp in self._trace_processors:
                     try:

releasenotes/notes/fix-tracing-dont_raise_exception_on_empty_partial_flush-131cd3268101f255.yaml

@@ -0,0 +1,4 @@
+---
+fixes:
+  - |
+    tracing: fixes a potential crash where using partial flushes and ``tracer.configure()`` could result in an IndexError

tests/contrib/subprocess/test_subprocess.py

@@ -219,7 +219,13 @@ def test_fork(tracer):
         with tracer.trace("ossystem_test"):
             pid = os.fork()
             if pid == 0:
-                return
+                # Exit, otherwise the rest of this process will continue to be pytest
+                from ddtrace.contrib.coverage import unpatch
+
+                unpatch()
+                import pytest
+
+                pytest.exit("in forked child", returncode=0)
 
         spans = tracer.pop()
         assert spans

tests/tracer/test_processors.py

@@ -393,9 +393,9 @@ def test_changing_tracer_sampler_changes_tracesamplingprocessor_sampler():
     tracer = Tracer()
     # get processor
     for aggregator in tracer._deferred_processors:
-        if type(aggregator) == SpanAggregator:
+        if type(aggregator) is SpanAggregator:
             for processor in aggregator._trace_processors:
-                if type(processor) == TraceSamplingProcessor:
+                if type(processor) is TraceSamplingProcessor:
                     sampling_processor = processor
 
     assert sampling_processor.sampler is tracer._sampler
@@ -588,11 +588,11 @@ def assert_span_sampling_decision_tags(
 
 def switch_out_trace_sampling_processor(tracer, sampling_processor):
     for aggregator in tracer._deferred_processors:
-        if type(aggregator) == SpanAggregator:
+        if type(aggregator) is SpanAggregator:
             i = 0
             while i < len(aggregator._trace_processors):
                 processor = aggregator._trace_processors[i]
-                if type(processor) == TraceSamplingProcessor:
+                if type(processor) is TraceSamplingProcessor:
                     aggregator._trace_processors[i] = sampling_processor
                     break
                 i += 1
@@ -692,3 +692,18 @@ def on_span_finish(self, span):
     with tracer.trace("test") as span:
         assert span.get_tag("on_start") is None
     assert span.get_tag("on_finish") is None
+
+
+def _stderr_contains_log(stderr: str) -> bool:
+    return "finished span not connected to a trace" in stderr
+
+
+@pytest.mark.subprocess(err=_stderr_contains_log)
+def test_tracer_reconfigured_with_active_span_does_not_crash():
+    import ddtrace
+
+    with ddtrace.tracer.trace("regression1") as exploding_span:
+        # Reconfiguring the tracer clears active traces
+        # Calling .finish() manually bypasses the code that catches the exception
+        ddtrace.tracer.configure(partial_flush_enabled=True, partial_flush_min_spans=1)
+        exploding_span.finish()