revert(tracer): reusing connections (#2999, #3440) (#3468) (#3473)

Issues with reusing connections were detected when testing in the environment.

With the timing of 1.0 we're pushing this feature off until version 1.1.

(cherry picked from commit 4eaca4a)

Co-authored-by: Kyle Verhoog <[email protected]>

Author: mergify[bot]

ddtrace/internal/writer.py

@@ -5,7 +5,6 @@
 import logging
 import os
 import sys
-from typing import Dict
 from typing import List
 from typing import Optional
 from typing import TYPE_CHECKING
@@ -41,8 +40,6 @@
 if TYPE_CHECKING:
     from ddtrace import Span
 
-    from .agent import ConnectionType
-
 
 log = get_logger(__name__)
 
@@ -57,7 +54,6 @@
 DEFAULT_BUFFER_SIZE = 8 << 20  # 8 MB
 DEFAULT_MAX_PAYLOAD_SIZE = 8 << 20  # 8 MB
 DEFAULT_PROCESSING_INTERVAL = 1.0
-DEFAULT_REUSE_CONNECTIONS = False
 
 
 def get_writer_buffer_size():
@@ -79,11 +75,6 @@ def get_writer_interval_seconds():
     )
 
 
-def get_writer_reuse_connections():
-    # type: () -> bool
-    return asbool(os.getenv("DD_TRACE_WRITER_REUSE_CONNECTIONS", DEFAULT_REUSE_CONNECTIONS))
-
-
 def _human_size(nbytes):
     """Return a human-readable size."""
     i = 0
@@ -255,7 +246,6 @@ def __init__(
         report_metrics=False,  # type: bool
         sync_mode=False,  # type: bool
         api_version=None,  # type: Optional[str]
-        reuse_connections=None,  # type: Optional[bool]
     ):
         # type: (...) -> None
         # Pre-conditions:
@@ -311,7 +301,6 @@ def __init__(
         self._metrics_reset()
         self._drop_sma = SimpleMovingAverage(DEFAULT_SMA_WINDOW)
         self._sync_mode = sync_mode
-        self._conn = None  # type: Optional[ConnectionType]
         self._retry_upload = tenacity.Retrying(
             # Retry RETRY_ATTEMPTS times within the first half of the processing
             # interval, using a Fibonacci policy with jitter
@@ -322,7 +311,6 @@ def __init__(
             retry=tenacity.retry_if_exception_type((compat.httplib.HTTPException, OSError, IOError)),
         )
         self._log_error_payloads = asbool(os.environ.get("_DD_TRACE_WRITER_LOG_ERROR_PAYLOADS", False))
-        self._reuse_connections = get_writer_reuse_connections() if reuse_connections is None else reuse_connections
 
     @property
     def _agent_endpoint(self):
@@ -375,34 +363,22 @@ def recreate(self):
             api_version=self._api_version,
         )
 
-    def _reset_connection(self):
-        # type: () -> None
-        if self._conn:
-            self._conn.close()
-            self._conn = None
-
     def _put(self, data, headers):
-        # type: (bytes, Dict[str, str]) -> Response
+        conn = get_connection(self.agent_url, self._timeout)
+
         with StopWatch() as sw:
-            if self._conn is None:
-                log.debug("creating new agent connection to %s with timeout %d", self.agent_url, self._timeout)
-                self._conn = get_connection(self.agent_url, self._timeout)
             try:
-                self._conn.request("PUT", self._endpoint, data, headers)
-                resp = compat.get_connection_response(self._conn)
+                conn.request("PUT", self._endpoint, data, headers)
+                resp = compat.get_connection_response(conn)
                 t = sw.elapsed()
                 if t >= self.interval:
                     log_level = logging.WARNING
                 else:
                     log_level = logging.DEBUG
                 log.log(log_level, "sent %s in %.5fs to %s", _human_size(len(data)), t, self._agent_endpoint)
                 return Response.from_http_response(resp)
-            except Exception:
-                self._reset_connection()
-                raise
             finally:
-                if self._conn and not self._reuse_connections:
-                    self._reset_connection()
+                conn.close()
 
     def _downgrade(self, payload, response):
         if self._endpoint == "v0.5/traces":
@@ -581,8 +557,4 @@ def _stop_service(  # type: ignore[override]
         super(AgentWriter, self)._stop_service()
         self.join(timeout=timeout)
 
-    def on_shutdown(self):
-        try:
-            self.periodic()
-        finally:
-            self._reset_connection()
+    on_shutdown = periodic

releasenotes/notes/keep-alive-b5ec5febb435daad.yaml

@@ -575,34 +575,3 @@ def test_writer_recreate_api_version(init_api_version, api_version, endpoint, en
     assert writer._api_version == api_version
     assert writer._endpoint == endpoint
     assert isinstance(writer._encoder, encoder_cls)
-
-
-def test_writer_reuse_connections_envvar(monkeypatch):
-    monkeypatch.setenv("DD_TRACE_WRITER_REUSE_CONNECTIONS", "false")
-    writer = AgentWriter(agent_url="http://localhost:9126")
-    assert not writer._reuse_connections
-
-    monkeypatch.setenv("DD_TRACE_WRITER_REUSE_CONNECTIONS", "true")
-    writer = AgentWriter(agent_url="http://localhost:9126")
-    assert writer._reuse_connections
-
-
-def test_writer_reuse_connections():
-    # Ensure connection is not reused
-    writer = AgentWriter(agent_url="http://localhost:9126", reuse_connections=True)
-    # Do an initial flush to get a connection
-    writer.flush_queue()
-    assert writer._conn is None
-    writer.flush_queue()
-    assert writer._conn is None
-
-
-def test_writer_reuse_connections_false():
-    # Ensure connection is reused
-    writer = AgentWriter(agent_url="http://localhost:9126", reuse_connections=False)
-    # Do an initial flush to get a connection
-    writer.flush_queue()
-    conn = writer._conn
-    # And another to potentially have it reset
-    writer.flush_queue()
-    assert writer._conn is conn