Add env variables to configure Writer behavior (#2581)

* Add env variables to configure Writer behavior
DD_TRACE_WRITER_BUFFER_SIZE_BYTES - How many bytes of traces to buffer between flushes to the agent
DD_TRACE_WRITER_MAX_PAYLOAD_SIZE_BYTES - Max size of any payload to write to the trace agent
DD_TRACE_WRITER_INTERVAL_SECONDS - How many seconds between flushes to the agent

* move around and update

* go back to DD_TRACE_WRITER_*

* fix flake8

* Add release note

* Update releasenotes/notes/add-writer-env-config-7c0916dff4f7f80c.yaml

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

* fix tests

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

Author: brettlangdon

ddtrace/internal/writer.py

@@ -21,6 +21,7 @@
 from ..constants import KEEP_SPANS_RATE_KEY
 from ..sampler import BasePrioritySampler
 from ..sampler import BaseSampler
+from ..utils.formats import get_env
 from ..utils.time import StopWatch
 from .agent import get_connection
 from .buffer import BufferFull
@@ -47,6 +48,29 @@
 # to 10 buckets of 1s duration.
 DEFAULT_SMA_WINDOW = 10
 
+DEFAULT_BUFFER_SIZE = 8 * 1000000  # 8mb
+DEFAULT_MAX_PAYLOAD_SIZE = 8 * 1000000  # 8mb
+DEFAULT_PROCESSING_INTERVAL = 1.0
+
+
+def get_writer_buffer_size():
+    # type: () -> int
+    return int(get_env("trace", "writer_buffer_size_bytes", default=DEFAULT_BUFFER_SIZE))  # type: ignore[arg-type]
+
+
+def get_writer_max_payload_size():
+    # type: () -> int
+    return int(
+        get_env("trace", "writer_max_payload_size_bytes", default=DEFAULT_MAX_PAYLOAD_SIZE)  # type: ignore[arg-type]
+    )
+
+
+def get_writer_interval_seconds():
+    # type: () -> float
+    return float(
+        get_env("trace", "writer_interval_seconds", default=DEFAULT_PROCESSING_INTERVAL)  # type: ignore[arg-type]
+    )
+
 
 def _human_size(nbytes):
     """Return a human-readable size."""
@@ -200,11 +224,11 @@ def __init__(
         agent_url,  # type: str
         sampler=None,  # type: Optional[BaseSampler]
         priority_sampler=None,  # type: Optional[BasePrioritySampler]
-        processing_interval=1.0,  # type: float
+        processing_interval=get_writer_interval_seconds(),  # type: float
         # Match the payload size since there is no functionality
         # to flush dynamically.
-        buffer_size=8 * 1000000,  # type: int
-        max_payload_size=8 * 1000000,  # type: int
+        buffer_size=get_writer_buffer_size(),  # type: int
+        max_payload_size=get_writer_max_payload_size(),  # type: int
         timeout=agent.get_trace_agent_timeout(),  # type: float
         dogstatsd=None,  # type: Optional[DogStatsd]
         report_metrics=False,  # type: bool

docs/configuration.rst

@@ -83,6 +83,18 @@ below:
      - Float
      - 2.0
      - The timeout in float to use to connect to the Datadog agent.
+   * - ``DD_TRACE_WRITER_BUFFER_SIZE_BYTES``
+     - Int
+     - 8000000
+     - The max size in bytes of traces to buffer between flushes to the agent.
+   * - ``DD_TRACE_WRITER_MAX_PAYLOAD_SIZE_BYTES``
+     - Int
+     - 8000000
+     - The max size in bytes of each payload sent to the trace agent. If max payload size is less than buffer size, multiple payloads will be sent to the trace agent.
+   * - ``DD_TRACE_WRITER_INTERVAL_SECONDS``
+     - Float
+     - 1.0
+     - The time between each flush of traces to the trace agent.
    * - ``DD_TRACE_STARTUP_LOGS``
      - Boolean
      - False

releasenotes/notes/add-writer-env-config-7c0916dff4f7f80c.yaml

@@ -0,0 +1,7 @@
+---
+features:
+  - |
+    Add new environment variables to configure the internal trace writer.
+
+    ``DD_TRACE_WRITER_MAX_BUFFER_SIZE``, ``DD_TRACE_WRITER_INTERVAL_SECONDS``,
+    ``DD_TRACE_WRITER_MAX_PAYLOAD_SIZE_BYTES``

tests/integration/test_integration.py

@@ -497,3 +497,67 @@ def test_call_basic_config(ddtrace_run_python_code_in_subprocess, call_basic_con
         assert out == six.b("1\n")
     else:
         assert out == six.b("0\n")
+
+
+def test_writer_env_configuration(run_python_code_in_subprocess):
+    env = os.environ.copy()
+    env["DD_TRACE_WRITER_BUFFER_SIZE_BYTES"] = "1000"
+    env["DD_TRACE_WRITER_MAX_PAYLOAD_SIZE_BYTES"] = "5000"
+    env["DD_TRACE_WRITER_INTERVAL_SECONDS"] = "5.0"
+
+    out, err, status, pid = run_python_code_in_subprocess(
+        """
+import ddtrace
+
+assert ddtrace.tracer.writer._buffer.max_size == 1000
+assert ddtrace.tracer.writer._buffer.max_item_size == 5000
+assert ddtrace.tracer.writer._interval == 5.0
+""",
+        env=env,
+    )
+    assert status == 0, (out, err)
+
+
+def test_writer_env_configuration_defaults(run_python_code_in_subprocess):
+    out, err, status, pid = run_python_code_in_subprocess(
+        """
+import ddtrace
+
+assert ddtrace.tracer.writer._buffer.max_size == 8000000
+assert ddtrace.tracer.writer._buffer.max_item_size == 8000000
+assert ddtrace.tracer.writer._interval == 1.0
+""",
+    )
+    assert status == 0, (out, err)
+
+
+def test_writer_env_configuration_ddtrace_run(ddtrace_run_python_code_in_subprocess):
+    env = os.environ.copy()
+    env["DD_TRACE_WRITER_BUFFER_SIZE_BYTES"] = "1000"
+    env["DD_TRACE_WRITER_MAX_PAYLOAD_SIZE_BYTES"] = "5000"
+    env["DD_TRACE_WRITER_INTERVAL_SECONDS"] = "5.0"
+
+    out, err, status, pid = ddtrace_run_python_code_in_subprocess(
+        """
+import ddtrace
+
+assert ddtrace.tracer.writer._buffer.max_size == 1000
+assert ddtrace.tracer.writer._buffer.max_item_size == 5000
+assert ddtrace.tracer.writer._interval == 5.0
+""",
+        env=env,
+    )
+    assert status == 0, (out, err)
+
+
+def test_writer_env_configuration_ddtrace_run_defaults(ddtrace_run_python_code_in_subprocess):
+    out, err, status, pid = ddtrace_run_python_code_in_subprocess(
+        """
+import ddtrace
+
+assert ddtrace.tracer.writer._buffer.max_size == 8000000
+assert ddtrace.tracer.writer._buffer.max_item_size == 8000000
+assert ddtrace.tracer.writer._interval == 1.0
+""",
+    )
+    assert status == 0, (out, err)