chore(tracer): deprecate Tracer.writer (#3223)

The `Tracer.writer` will be removed from the public API in v1.0 so is deprecated here.

Author: majorgreys

benchmarks/threading/scenario.py

@@ -25,6 +25,10 @@ def write(self, spans=None):
         # type: (Optional[List[Span]]) -> None
         pass
 
+    def flush_queue(self):
+        # type: () -> None
+        pass
+
 
 @bm.register
 class Threading(bm.Scenario):

ddtrace/contrib/httplib/patch.py

@@ -165,8 +165,9 @@ def should_skip_request(pin, request):
     if not pin or not pin.enabled():
         return True
 
-    if hasattr(pin.tracer.writer, "agent_url"):
-        parsed = parse.urlparse(pin.tracer.writer.agent_url)
+    agent_url = pin.tracer.agent_trace_url
+    if agent_url:
+        parsed = parse.urlparse(agent_url)
         return request.host == parsed.hostname and request.port == parsed.port
     return False
 

ddtrace/contrib/rq/__init__.py

@@ -177,7 +177,7 @@ def traced_perform_job(rq, pin, func, instance, args, kwargs):
     finally:
         # Force flush to agent since the process `os.exit()`s
         # immediately after this method returns
-        pin.tracer.writer.flush_queue()
+        pin.tracer.flush()
 
 
 @trace_utils.with_traced_module

ddtrace/internal/debug.py

@@ -49,11 +49,11 @@ def collect(tracer):
 
     from ddtrace.internal.runtime.runtime_metrics import RuntimeWorker
 
-    if isinstance(tracer.writer, LogWriter):
+    if isinstance(tracer._writer, LogWriter):
         agent_url = "AGENTLESS"
         agent_error = None
-    elif isinstance(tracer.writer, AgentWriter):
-        writer = tracer.writer
+    elif isinstance(tracer._writer, AgentWriter):
+        writer = tracer._writer
         agent_url = writer.agent_url
         try:
             writer.write([])

ddtrace/internal/writer.py

@@ -173,6 +173,11 @@ def write(self, spans=None):
         # type: (Optional[List[Span]]) -> None
         pass
 
+    @abc.abstractmethod
+    def flush_queue(self):
+        # type: () -> None
+        pass
+
 
 class LogWriter(TraceWriter):
     def __init__(
@@ -210,6 +215,10 @@ def write(self, spans=None):
         self.out.write(encoded + "\n")
         self.out.flush()
 
+    def flush_queue(self):
+        # type: () -> None
+        pass
+
 
 class AgentWriter(periodic.PeriodicService, TraceWriter):
     """Writer to the Datadog Agent.

ddtrace/profiling/profiler.py

@@ -146,8 +146,8 @@ def _build_default_exporters(self):
             )
             endpoint = self.ENDPOINT_TEMPLATE.format(os.environ.get("DD_SITE", "datadoghq.com"))
         else:
-            if isinstance(self.tracer.writer, writer.AgentWriter):
-                endpoint = self.tracer.writer.agent_url
+            if isinstance(self.tracer._writer, writer.AgentWriter):
+                endpoint = self.tracer._writer.agent_url
             else:
                 endpoint = agent.get_trace_url()
 

ddtrace/tracer.py

@@ -61,6 +61,7 @@
 from .sampler import RateByServiceSampler
 from .sampler import RateSampler
 from .span import Span
+from .vendor.debtcollector import removals
 
 
 log = get_logger(__name__)
@@ -147,7 +148,7 @@ def __init__(
                 report_metrics=config.health_metrics_enabled,
                 sync_mode=self._use_sync_mode(),
             )
-        self.writer = writer  # type: TraceWriter
+        self._writer = writer  # type: TraceWriter
 
         # DD_TRACER_... should be deprecated after version 1.0.0 is released
         pfe_default_value = False
@@ -351,8 +352,8 @@ def configure(
         if any(x is not None for x in [hostname, port, uds_path, https]):
             # If any of the parts of the URL have updated, merge them with
             # the previous writer values.
-            if isinstance(self.writer, AgentWriter):
-                prev_url_parsed = compat.parse.urlparse(self.writer.agent_url)
+            if isinstance(self._writer, AgentWriter):
+                prev_url_parsed = compat.parse.urlparse(self._writer.agent_url)
             else:
                 prev_url_parsed = compat.parse.urlparse("")
 
@@ -369,26 +370,26 @@ def configure(
                     port = prev_url_parsed.port
                 scheme = "https" if https else "http"
                 url = "%s://%s:%s" % (scheme, hostname, port)
-        elif isinstance(self.writer, AgentWriter):
+        elif isinstance(self._writer, AgentWriter):
             # Reuse the URL from the previous writer if there was one.
-            url = self.writer.agent_url
+            url = self._writer.agent_url
         else:
             # No URL parts have updated and there's no previous writer to
             # get the URL from.
             url = None
 
         try:
-            self.writer.stop()
+            self._writer.stop()
         except service.ServiceStatusError:
             # It's possible the writer never got started in the first place :(
             pass
 
         if writer is not None:
-            self.writer = writer
+            self._writer = writer
         elif url:
             # Verify the URL and create a new AgentWriter with it.
             agent.verify_url(url)
-            self.writer = AgentWriter(
+            self._writer = AgentWriter(
                 url,
                 sampler=self.sampler,
                 priority_sampler=self.priority_sampler,
@@ -397,11 +398,11 @@ def configure(
                 sync_mode=self._use_sync_mode(),
                 api_version=api_version,
             )
-        elif writer is None and isinstance(self.writer, LogWriter):
+        elif writer is None and isinstance(self._writer, LogWriter):
             # No need to do anything for the LogWriter.
             pass
-        if isinstance(self.writer, AgentWriter):
-            self.writer.dogstatsd = get_dogstatsd_client(self._dogstatsd_url)  # type: ignore[has-type]
+        if isinstance(self._writer, AgentWriter):
+            self._writer.dogstatsd = get_dogstatsd_client(self._dogstatsd_url)  # type: ignore[has-type]
         self._initialize_span_processors()
 
         if context_provider is not None:
@@ -436,7 +437,7 @@ def _child_after_fork(self):
         self._services = set()
 
         # Re-create the background writer thread
-        self.writer = self.writer.recreate()
+        self._writer = self._writer.recreate()
         self._initialize_span_processors()
 
         self._new_process = True
@@ -671,8 +672,8 @@ def _initialize_span_processors(self, appsec_enabled=asbool(get_env("appsec", "e
                 partial_flush_enabled=self._partial_flush_enabled,
                 partial_flush_min_spans=self._partial_flush_min_spans,
                 trace_processors=trace_processors,
-                writer=self.writer,
-            )
+                writer=self._writer,
+            ),
         ]  # type: List[SpanProcessor]
 
         if appsec_enabled:
@@ -811,7 +812,24 @@ def write(self, spans):
             return
 
         if spans is not None:
-            self.writer.write(spans=spans)
+            self._writer.write(spans=spans)
+
+    @removals.removed_property(message="Use Tracer.flush instead to flush buffered traces to agent", version="1.0.0")
+    def writer(self):
+        return self._writer
+
+    @property
+    def agent_trace_url(self):
+        # type: () -> Optional[str]
+        """Trace agent url"""
+        if isinstance(self._writer, AgentWriter):
+            return self._writer.agent_url
+
+        return None
+
+    def flush(self):
+        """Flush the buffer of the trace writer. This does nothing if an unbuffered trace writer is used."""
+        self._writer.flush_queue()
 
     @deprecated(message="Manually setting service info is no longer necessary", version="1.0.0")
     def set_service_info(self, *args, **kwargs):
@@ -974,7 +992,7 @@ def shutdown(self, timeout=None):
         :type timeout: :obj:`int` | :obj:`float` | :obj:`None`
         """
         try:
-            self.writer.stop(timeout=timeout)
+            self._writer.stop(timeout=timeout)
         except service.ServiceStatusError:
             # It's possible the writer never got started in the first place :(
             pass

docs/spelling_wordlist.txt

@@ -143,6 +143,7 @@ submodules
 timestamp
 tweens
 uWSGI
+unbuffered
 unicode
 uninstrumented
 unix

releasenotes/notes/deprecate-tracer-writer-b4921987ad47a832.yaml

@@ -0,0 +1,7 @@
+---
+features:
+  - |
+    Add :py:attr:`ddtrace.Tracer.agent_trace_url` and :py:meth:`ddtrace.Tracer.flush`.
+deprecations:
+  - |
+    :py:attr:`ddtrace.Tracer.writer` is deprecated. To force flushing of buffered traces to the agent, use :py:meth:`ddtrace.Tracer.flush` instead.

tests/benchmarks/test_encoding.py

@@ -109,7 +109,7 @@ def test_dd_origin_tagging_spans_via_encoder(benchmark, trace_size):
             with tracer.trace("") as span:
                 span.set_tag("tag", "value")
                 pass
-    trace = tracer.writer.pop()
+    trace = tracer.pop()
 
     def _(trace):
         trace_encoder.put(trace)