fix userprocessor overwriting

Author: mabdinur

ddtrace/_trace/processor/__init__.py

@@ -271,6 +271,7 @@ def __init__(
         partial_flush_enabled: bool,
         partial_flush_min_spans: int,
         trace_processors: List[TraceProcessor],
+        user_processors: Optional[List[TraceProcessor]] = None,
     ):
         # Set partial flushing
         self.partial_flush_enabled = partial_flush_enabled
@@ -280,7 +281,8 @@ def __init__(
             config._trace_compute_stats, get_span_sampling_rules(), asm_config._apm_opt_out
         )
         self.tags_processor = TraceTagsProcessor()
-        self.trace_processors = trace_processors
+        self.trace_processors = trace_processors or []
+        self.user_processors = user_processors or []
         if SpanAggregator._use_log_writer():
             self.writer: TraceWriter = LogWriter()
         else:
@@ -314,6 +316,8 @@ def __repr__(self) -> str:
             f"{self.sampling_processor},"
             f"{self.tags_processor},"
             f"{self.trace_processors}, "
+            f"{self.user_processors}, "
+            f"{self._span_metrics}, "
             f"{self.writer})"
         )
 
@@ -375,7 +379,9 @@ def on_span_finish(self, span: Span) -> None:
                     finished[0].set_metric("_dd.py.partial_flush", num_finished)
 
                 spans: Optional[List[Span]] = finished
-                for tp in chain(self.trace_processors, [self.sampling_processor, self.tags_processor]):
+                for tp in chain(
+                    self.trace_processors, self.user_processors, [self.sampling_processor, self.tags_processor]
+                ):
                     try:
                         if spans is None:
                             return

ddtrace/_trace/tracer.py

@@ -373,7 +373,7 @@ def configure(
                 self._span_aggregator.writer._api_version = "v0.4"
 
         if trace_processors:
-            self._span_aggregator.trace_processors = trace_processors
+            self._span_aggregator.user_processors = trace_processors
 
         if any(
             x is not None

ddtrace/internal/ci_visibility/recorder.py

@@ -631,7 +631,7 @@ def disable(cls) -> None:
         log.debug("%s disabled", cls.__name__)
 
     def _start_service(self) -> None:
-        tracer_filters = self.tracer._span_aggregator.trace_processors
+        tracer_filters = self.tracer._span_aggregator.user_processors
         if not any(isinstance(tracer_filter, TraceCiVisibilityFilter) for tracer_filter in tracer_filters):
             tracer_filters += [TraceCiVisibilityFilter(self._tags, self._service)]  # type: ignore[arg-type]
             self.tracer.configure(trace_processors=tracer_filters)

ddtrace/opentracer/tracer.py

@@ -104,7 +104,7 @@ def __init__(
         trace_processors = None
         if isinstance(self._config.get(keys.SETTINGS), dict) and self._config[keys.SETTINGS].get("FILTERS"):  # type: ignore[union-attr]
             trace_processors = self._config[keys.SETTINGS]["FILTERS"]  # type: ignore[index]
-            self._dd_tracer._span_aggregator.trace_processors = trace_processors
+            self._dd_tracer._span_aggregator.user_processors = trace_processors
 
         if self._config[keys.ENABLED]:
             self._dd_tracer.enabled = self._config[keys.ENABLED]