remove writer arg from SpanAggregator

Author: mabdinur

ddtrace/_trace/processor/__init__.py

@@ -272,7 +272,6 @@ def __init__(
         partial_flush_enabled: bool,
         partial_flush_min_spans: int,
         trace_processors: Iterable[TraceProcessor],
-        writer: Optional[TraceWriter] = None,
     ):
         # Set partial flushing
         self.partial_flush_enabled = partial_flush_enabled
@@ -283,13 +282,7 @@ def __init__(
         )
         self.tags_processor = TraceTagsProcessor()
         self.trace_processors = trace_processors
-        # Initialize writer
-        if writer is not None:
-            if isinstance(writer, AgentWriter):
-                # Ensure the active writer is set to update the active sampler
-                writer._response_cb = self._agent_response_callback
-            self.writer: TraceWriter = writer
-        elif SpanAggregator._use_log_writer():
+        if SpanAggregator._use_log_writer():
             self.writer = LogWriter()
         else:
             verify_url(agent_config.trace_agent_url)

ddtrace/_trace/tracer.py

@@ -184,7 +184,6 @@ def __init__(self) -> None:
             partial_flush_enabled=config._partial_flush_enabled,
             partial_flush_min_spans=config._partial_flush_min_spans,
             trace_processors=[PeerServiceProcessor(_ps_config), BaseServiceProcessor()],
-            writer=None,
         )
         if config._data_streams_enabled:
             # Inline the import to avoid pulling in ddsketch or protobuf

tests/tracer/test_processors.py

@@ -70,9 +70,9 @@ def process_trace(self, trace):
         trace_processors=[
             mock_proc1,
             mock_proc2,
-        ],
-        writer=writer,
+        ]
     )
+    aggr.writer = writer
 
     span = Span("span", on_finish=[aggr.on_span_finish])
     aggr.on_span_start(span)
@@ -104,8 +104,8 @@ def process_trace(self, trace):
             mock_bad,
             mock_good_after,
         ],
-        writer=writer,
     )
+    aggr.writer = writer
 
     span = Span("span", on_finish=[aggr.on_span_finish])
     aggr.on_span_start(span)
@@ -119,7 +119,8 @@ def process_trace(self, trace):
 
 def test_aggregator_multi_span():
     writer = DummyWriter()
-    aggr = SpanAggregator(partial_flush_enabled=False, partial_flush_min_spans=0, trace_processors=[], writer=writer)
+    aggr = SpanAggregator(partial_flush_enabled=False, partial_flush_min_spans=0, trace_processors=[])
+    aggr.writer = writer
 
     # Normal usage
     parent = Span("parent", on_finish=[aggr.on_span_finish])
@@ -152,7 +153,8 @@ def test_aggregator_multi_span():
 
 def test_aggregator_partial_flush_0_spans():
     writer = DummyWriter()
-    aggr = SpanAggregator(partial_flush_enabled=True, partial_flush_min_spans=0, trace_processors=[], writer=writer)
+    aggr = SpanAggregator(partial_flush_enabled=True, partial_flush_min_spans=0, trace_processors=[])
+    aggr.writer = writer
 
     # Normal usage
     parent = Span("parent", on_finish=[aggr.on_span_finish])
@@ -187,7 +189,8 @@ def test_aggregator_partial_flush_0_spans():
 
 def test_aggregator_partial_flush_2_spans():
     writer = DummyWriter()
-    aggr = SpanAggregator(partial_flush_enabled=True, partial_flush_min_spans=2, trace_processors=[], writer=writer)
+    aggr = SpanAggregator(partial_flush_enabled=True, partial_flush_min_spans=2, trace_processors=[])
+    aggr.writer = writer
 
     # Normal usage
     parent = Span("parent", on_finish=[aggr.on_span_finish])
@@ -331,7 +334,8 @@ def test_trace_128bit_processor(trace_id):
 def test_span_creation_metrics():
     """Test that telemetry metrics are queued in batches of 100 and the remainder is sent on shutdown"""
     writer = DummyWriter()
-    aggr = SpanAggregator(partial_flush_enabled=False, partial_flush_min_spans=0, trace_processors=[], writer=writer)
+    aggr = SpanAggregator(partial_flush_enabled=False, partial_flush_min_spans=0, trace_processors=[])
+    aggr.writer = writer
 
     with override_global_config(dict(_telemetry_enabled=True)):
         with mock.patch("ddtrace.internal.telemetry.telemetry_writer.add_count_metric") as mock_tm: