chore(tracing): removes tracer._configure from products and tests (#12696)

## Background

In ddtrace v3.0, most functionality previously handled by
tracer.configure(...) was internalized. As a result, ddtrace >3.0 no
longer provides a public API for configuring agent connections, custom
samplers, UDS, or enabling tracing. To avoid refactoring all tests and
products relying on tracer.configure(...), we introduced the private
method tracer._configure.

This method has been used across ddtrace products (e.g., ASM, MLObs, and
CiVis), integrations (e.g., Tornado, Vertex AI), and hundreds of tests.
Now, it’s time to fully migrate away from this API. 🫡

## Motivation

This refactor encourages the use of environment variables and/or remote
configuration for setting up tracing features. Even though
tracer._configure is internal, it can still be used to enable
unsupported behavior. This PR aims to put an end to that. 🫡

## Changes

- Removes ddtrace.Tracer._configure from the library.
- Updates tests requiring global tracer reconfiguration to run in a
subprocess.
- Introduces Tracer._reset.
- Removes tests validating unsupported behavior (e.g., using
tracer.configure to change the agent URL).
- Removes tests validating unsupported behavior (e.g., using
tracer.configure to change the agent URL).

Note: This PR also introduces Tracer._recreate(...), a private method
used in ASM and CIVis tests to recreate the agent writer and span
processors. This was added to limit the scope of changes in this PR.
Ideally, tests requiring tracer reconfiguration/restart should be run in
a subprocess. This is a temporary workaround (hopefully).

Follow up:
- Deprecate `tracer.configure(...)`. 
- Provide a public method for setting trace processors (ex:
Tracer.set_processor)
- Encourage ASM users to enable features via environment variables
- Remove `tracer.configure(...)` from all tests.

## Checklist
- [x] PR author has checked that all the criteria below are met
- The PR description includes an overview of the change
- The PR description articulates the motivation for the change
- The change includes tests OR the PR description describes a testing
strategy
- The PR description notes risks associated with the change, if any
- Newly-added code is easy to change
- The change follows the [library release note
guidelines](https://ddtrace.readthedocs.io/en/stable/releasenotes.html)
- The change includes or references documentation updates if necessary
- Backport labels are set (if
[applicable](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting))

## Reviewer Checklist
- [x] Reviewer has checked that all the criteria below are met 
- Title is accurate
- All changes are related to the pull request's stated goal
- Avoids breaking
[API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces)
changes
- Testing strategy adequately addresses listed risks
- Newly-added code is easy to change
- Release note makes sense to a user of the library
- If necessary, author has acknowledged and discussed the performance
implications of this PR as reported in the benchmarks PR comment
- Backport labels are set in a manner that is consistent with the
[release branch maintenance
policy](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting)

Author: mabdinur

benchmarks/threading/scenario.py

@@ -2,28 +2,16 @@
 import random
 from typing import Callable
 from typing import Generator
-from typing import List
-from typing import Optional
 
 import bm
 
-from ddtrace._trace.span import Span
-from ddtrace._trace.tracer import Tracer
-from ddtrace.internal.writer import TraceWriter
+from ddtrace.trace import TraceFilter
+from ddtrace.trace import Tracer
 
 
-class NoopWriter(TraceWriter):
-    def recreate(self) -> TraceWriter:
-        return NoopWriter()
-
-    def stop(self, timeout: Optional[float] = None) -> None:
-        pass
-
-    def write(self, spans: Optional[List[Span]] = None) -> None:
-        pass
-
-    def flush_queue(self) -> None:
-        pass
+class _DropTraces(TraceFilter):
+    def process_trace(self, trace):
+        return
 
 
 class Threading(bm.Scenario):
@@ -42,7 +30,7 @@ def run(self) -> Generator[Callable[[int], None], None, None]:
         from ddtrace.trace import tracer
 
         # configure global tracer to drop traces rather
-        tracer.configure(writer=NoopWriter())
+        tracer.configure(trace_processors=[_DropTraces()])
 
         def _(loops: int) -> None:
             for _ in range(loops):

ddtrace/_trace/tracer.py

@@ -14,7 +14,6 @@
 from typing import Tuple
 from typing import TypeVar
 from typing import Union
-from urllib import parse
 
 from ddtrace import _hooks
 from ddtrace import config
@@ -411,48 +410,6 @@ def configure(
         :param List[TraceProcessor] trace_processors: This parameter sets TraceProcessor (ex: TraceFilters).
            Trace processors are used to modify and filter traces based on certain criteria.
         """
-        return self._configure(
-            context_provider=context_provider,
-            trace_processors=trace_processors,
-            compute_stats_enabled=compute_stats_enabled,
-            appsec_enabled=appsec_enabled,
-            iast_enabled=iast_enabled,
-            apm_tracing_disabled=apm_tracing_disabled,
-        )
-
-    def _configure(
-        self,
-        enabled: Optional[bool] = None,
-        hostname: Optional[str] = None,
-        port: Optional[int] = None,
-        uds_path: Optional[str] = None,
-        https: Optional[bool] = None,
-        sampler: Optional[DatadogSampler] = None,
-        context_provider: Optional[BaseContextProvider] = None,
-        wrap_executor: Optional[Callable] = None,
-        priority_sampling: Optional[bool] = None,
-        trace_processors: Optional[List[TraceProcessor]] = None,
-        dogstatsd_url: Optional[str] = None,
-        writer: Optional[TraceWriter] = None,
-        partial_flush_enabled: Optional[bool] = None,
-        partial_flush_min_spans: Optional[int] = None,
-        api_version: Optional[str] = None,
-        compute_stats_enabled: Optional[bool] = None,
-        appsec_enabled: Optional[bool] = None,
-        iast_enabled: Optional[bool] = None,
-        apm_tracing_disabled: Optional[bool] = None,
-    ) -> None:
-        if enabled is not None:
-            self.enabled = enabled
-
-        if trace_processors is not None:
-            self._user_trace_processors = trace_processors
-
-        if partial_flush_enabled is not None:
-            self._partial_flush_enabled = partial_flush_enabled
-
-        if partial_flush_min_spans is not None:
-            self._partial_flush_min_spans = partial_flush_min_spans
 
         if appsec_enabled is not None:
             asm_config._asm_enabled = appsec_enabled
@@ -468,114 +425,34 @@ def _configure(
             # Disable compute stats (neither agent or tracer should compute them)
             config._trace_compute_stats = False
             # Update the rate limiter to 1 trace per minute when tracing is disabled
-            if sampler is not None:
-                sampler._rate_limit_always_on = True
-                sampler.limiter.rate_limit = 1
-                sampler.limiter.time_window = 60e9
-            else:
-                sampler = DatadogSampler(rate_limit=1, rate_limit_window=60e9, rate_limit_always_on=True)
+            self._sampler = DatadogSampler(rate_limit=1, rate_limit_window=60e9, rate_limit_always_on=True)
             log.debug("ASM standalone mode is enabled, traces will be rate limited at 1 trace per minute")
 
-        if sampler is not None:
-            self._sampler = sampler
-
-        if dogstatsd_url is not None:
-            self._dogstatsd_url = dogstatsd_url
-
-        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.
-            prev_url_parsed = parse.urlparse(self._agent_url)
-
-            if uds_path is not None:
-                if hostname is None and prev_url_parsed.scheme == "unix":
-                    hostname = prev_url_parsed.hostname
-                new_url = "unix://%s%s" % (hostname or "", uds_path)
-            else:
-                if https is None:
-                    https = prev_url_parsed.scheme == "https"
-                if hostname is None:
-                    hostname = prev_url_parsed.hostname or ""
-                if port is None:
-                    port = prev_url_parsed.port
-                scheme = "https" if https else "http"
-                new_url = "%s://%s:%s" % (scheme, hostname, port)
-            verify_url(new_url)
-            self._agent_url = new_url
-        else:
-            new_url = None
-
         if compute_stats_enabled is not None:
             self._compute_stats = compute_stats_enabled
 
-        try:
-            self._writer.stop()
-        except ServiceStatusError:
-            # It's possible the writer never got started
-            pass
-
-        if writer is not None:
-            self._writer = writer
-        elif any(x is not None for x in [new_url, api_version, sampler, dogstatsd_url, appsec_enabled]):
-            if asm_config._asm_enabled:
-                api_version = "v0.4"
-            self._writer = AgentWriter(
-                self._agent_url,
-                dogstatsd=get_dogstatsd_client(self._dogstatsd_url),
-                sync_mode=self._use_sync_mode(),
-                api_version=api_version,
-                # if apm opt out, neither agent or tracer should compute the stats
-                headers=(
-                    {"Datadog-Client-Computed-Stats": "yes"}
-                    if (compute_stats_enabled or asm_config._apm_opt_out)
-                    else {}
-                ),
-                report_metrics=not asm_config._apm_opt_out,
-                response_callback=self._agent_response_callback,
-            )
-        elif writer is None and isinstance(self._writer, LogWriter):
-            # No need to do anything for the LogWriter.
-            pass
         if isinstance(self._writer, AgentWriter):
+            if appsec_enabled:
+                self._writer._api_version = "v0.4"
             self._writer.dogstatsd = get_dogstatsd_client(self._dogstatsd_url)
 
+        if trace_processors:
+            self._user_trace_processors = trace_processors
+
         if any(
             x is not None
             for x in [
-                partial_flush_min_spans,
-                partial_flush_enabled,
-                writer,
-                dogstatsd_url,
-                hostname,
-                port,
-                https,
-                uds_path,
-                api_version,
-                sampler,
                 trace_processors,
                 compute_stats_enabled,
                 appsec_enabled,
                 iast_enabled,
             ]
         ):
-            self._span_processors, self._appsec_processor, self._deferred_processors = _default_span_processors_factory(
-                self._user_trace_processors,
-                self._writer,
-                self._partial_flush_enabled,
-                self._partial_flush_min_spans,
-                self._compute_stats,
-                self._single_span_sampling_rules,
-                self._agent_url,
-                self._sampler,
-                self._endpoint_call_counter_span_processor,
-            )
+            self._recreate()
 
         if context_provider is not None:
             self.context_provider = context_provider
 
-        if wrap_executor is not None:
-            self._wrap_executor = wrap_executor
-
         self._generate_diagnostic_logs()
 
     def _agent_response_callback(self, resp: AgentResponse) -> None:
@@ -611,8 +488,22 @@ def _generate_diagnostic_logs(self):
 
     def _child_after_fork(self):
         self._pid = getpid()
+        self._recreate()
+        self._new_process = True
+
+    def _recreate(self):
+        """Re-initialize the tracer's processors and trace writer. This method should only be used in tests."""
+        # Stop the writer.
+        # This will stop the periodic thread in HTTPWriters, preventing memory leaks and unnecessary I/O.
+        try:
+            self._writer.stop()
+        except ServiceStatusError:
+            # Some writers (ex: AgentWriter), start when the first trace chunk is encoded. Stopping
+            # the writer before that point will raise a ServiceStatusError.
+            pass
         # Re-create the background writer thread
         self._writer = self._writer.recreate()
+        # Recreate the trace and span processors
         self._span_processors, self._appsec_processor, self._deferred_processors = _default_span_processors_factory(
             self._user_trace_processors,
             self._writer,
@@ -625,8 +516,6 @@ def _child_after_fork(self):
             self._endpoint_call_counter_span_processor,
         )
 
-        self._new_process = True
-
     def _start_span_after_shutdown(
         self,
         name: str,

ddtrace/appsec/_remoteconfiguration.py

@@ -221,12 +221,12 @@ def _appsec_1click_activation(features: Mapping[str, Any], test_tracer: Optional
 
             if rc_asm_enabled:
                 if not asm_config._asm_enabled:
-                    tracer._configure(appsec_enabled=True)
+                    tracer.configure(appsec_enabled=True)
                 else:
                     asm_config._asm_enabled = True
             else:
                 if asm_config._asm_enabled:
-                    tracer._configure(appsec_enabled=False)
+                    tracer.configure(appsec_enabled=False)
                 else:
                     asm_config._asm_enabled = False
 

ddtrace/contrib/internal/tornado/application.py

@@ -1,3 +1,5 @@
+from urllib.parse import urlparse
+
 from tornado import template
 
 import ddtrace
@@ -34,21 +36,25 @@ def tracer_config(__init__, app, args, kwargs):
     service = settings["default_service"]
 
     # extract extra settings
+    # TODO: Remove `FILTERS` from supported settings
     trace_processors = settings.get("settings", {}).get("FILTERS")
 
-    # the tracer must use the right Context propagation and wrap executor;
-    # this action is done twice because the patch() method uses the
-    # global tracer while here we can have a different instance (even if
-    # this is not usual).
-    tracer._configure(
+    tracer.configure(
         context_provider=context_provider,
-        wrap_executor=decorators.wrap_executor,
-        enabled=settings.get("enabled", None),
-        hostname=settings.get("agent_hostname", None),
-        port=settings.get("agent_port", None),
         trace_processors=trace_processors,
     )
-
+    tracer._wrap_executor = decorators.wrap_executor
+    # TODO: Remove `enabled`, `hostname` and `port` settings in v4.0
+    # Tracer should be configured via environment variables
+    if settings.get("enabled") is not None:
+        tracer.enabled = settings["enabled"]
+    if settings.get("hostname") is not None or settings.get("port") is not None:
+        curr_agent_url = urlparse(tracer._agent_url)
+        hostname = settings.get("hostname", curr_agent_url.hostname)
+        port = settings.get("port", curr_agent_url.port)
+        tracer._agent_url = tracer._writer.intake_url = f"{curr_agent_url.scheme}://{hostname}:{port}"
+        tracer._recreate()
+    # TODO: Remove tags from settings, tags should be set via `DD_TAGS` environment variable
     # set global tags if any
     tags = settings.get("tags", None)
     if tags:

ddtrace/contrib/internal/tornado/patch.py

@@ -50,10 +50,8 @@ def patch():
     _w("tornado.template", "Template.generate", template.generate)
 
     # configure the global tracer
-    ddtrace.tracer._configure(
-        context_provider=context_provider,
-        wrap_executor=decorators.wrap_executor,
-    )
+    ddtrace.tracer.context_provider = context_provider
+    ddtrace.tracer._wrap_executor = decorators.wrap_executor
 
 
 def unpatch():

ddtrace/internal/ci_visibility/recorder.py

@@ -175,7 +175,9 @@ def __init__(
 
             # Partial traces are required for ITR to work in suite-level skipping for long test sessions, but we
             # assume that a tracer is already configured if it's been passed in.
-            self.tracer._configure(partial_flush_enabled=True, partial_flush_min_spans=TRACER_PARTIAL_FLUSH_MIN_SPANS)
+            self.tracer._partial_flush_enabled = True
+            self.tracer._partial_flush_min_spans = TRACER_PARTIAL_FLUSH_MIN_SPANS
+            self.tracer._recreate()
 
         self._api_client: Optional[_TestVisibilityAPIClientBase] = None
 
@@ -395,7 +397,8 @@ def _configure_writer(
                 itr_suite_skipping_mode=self._suite_skipping_mode,
             )
         if writer is not None:
-            self.tracer._configure(writer=writer)
+            self.tracer._writer = writer
+            self.tracer._recreate()
 
     def _agent_evp_proxy_is_available(self) -> bool:
         try:
@@ -602,7 +605,7 @@ def _start_service(self) -> None:
         tracer_filters = self.tracer._user_trace_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)
+            self.tracer.configure(trace_processors=tracer_filters)
 
         if self.test_skipping_enabled():
             self._fetch_tests_to_skip()

ddtrace/internal/ci_visibility/writer.py

@@ -108,7 +108,6 @@ def __init__(
         dogstatsd=None,  # type: Optional[DogStatsd]
         sync_mode=False,  # type: bool
         report_metrics=False,  # type: bool
-        api_version=None,  # type: Optional[str]
         reuse_connections=None,  # type: Optional[bool]
         headers=None,  # type: Optional[Dict[str, str]]
         use_evp=False,  # type: bool
@@ -129,10 +128,13 @@ def __init__(
         if not intake_url:
             intake_url = "%s.%s" % (AGENTLESS_BASE_URL, os.getenv("DD_SITE", AGENTLESS_DEFAULT_SITE))
 
+        self._use_evp = use_evp
         clients = (
-            [CIVisibilityProxiedEventClient()] if use_evp else [CIVisibilityAgentlessEventClient()]
+            [CIVisibilityProxiedEventClient()] if self._use_evp else [CIVisibilityAgentlessEventClient()]
         )  # type: List[WriterClientBase]
-        if coverage_enabled:
+        self._coverage_enabled = coverage_enabled
+        self._itr_suite_skipping_mode = itr_suite_skipping_mode
+        if self._coverage_enabled:
             if not intake_cov_url:
                 intake_cov_url = "%s.%s" % (AGENTLESS_COVERAGE_BASE_URL, os.getenv("DD_SITE", AGENTLESS_DEFAULT_SITE))
             clients.append(
@@ -154,6 +156,8 @@ def __init__(
             timeout=timeout,
             dogstatsd=dogstatsd,
             sync_mode=sync_mode,
+            # FIXME(munir): report_metrics is not used in the CI Visibility Writers
+            # self._report_metrics = report_metrics,
             reuse_connections=reuse_connections,
             headers=headers,
         )
@@ -170,6 +174,12 @@ def recreate(self):
             timeout=self._timeout,
             dogstatsd=self.dogstatsd,
             sync_mode=self._sync_mode,
+            report_metrics=self._report_metrics,
+            reuse_connections=self._reuse_connections,
+            headers=self._headers,
+            use_evp=self._use_evp,
+            coverage_enabled=self._coverage_enabled,
+            itr_suite_skipping_mode=self._itr_suite_skipping_mode,
         )
 
     def _put(self, data, headers, client, no_trace):