chore(ci): enabling partial flushing by default (#4041)

## Description
Enabling payload partial flush by default

Resolves #1632

## Checklist
- [x] Title must conform to [conventional commit](https://github.com/conventional-changelog/commitlint/tree/master/%40commitlint/config-conventional).
- [x] Add additional sections for `feat` and `fix` pull requests.
- [x] Ensure tests are passing for affected code.
- [x] [Library documentation](https://github.com/DataDog/dd-trace-py/tree/1.x/docs) and/or [Datadog's documentation site](https://github.com/DataDog/documentation/) is updated. Link to doc PR in description.

## Reviewer Checklist
- [ ] Title is accurate.
- [ ] Description motivates each change.
- [ ] No unnecessary changes were introduced in this PR.
- [ ] PR cannot be broken up into smaller PRs.
- [ ] Avoid breaking [API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces) changes unless absolutely necessary.
- [ ] Tests provided or description of manual testing performed is included in the code or PR.
- [ ] Release note has been added for fixes and features, or else `changelog/no-changelog` label added.
- [ ] All relevant GitHub issues are correctly linked.
- [ ] Backports are identified and tagged with Mergifyio.
- [ ] Add to milestone.

Author: mawais54013

ddtrace/tracer.py

@@ -215,7 +215,7 @@ def __init__(
             )
         self._single_span_sampling_rules = get_span_sampling_rules()  # type: List[SpanSamplingRule]
         self._writer = writer  # type: TraceWriter
-        self._partial_flush_enabled = asbool(os.getenv("DD_TRACE_PARTIAL_FLUSH_ENABLED", default=False))
+        self._partial_flush_enabled = asbool(os.getenv("DD_TRACE_PARTIAL_FLUSH_ENABLED", default=True))
         self._partial_flush_min_spans = int(os.getenv("DD_TRACE_PARTIAL_FLUSH_MIN_SPANS", default=500))
         self._appsec_enabled = config._appsec_enabled
 

docs/configuration.rst

@@ -233,6 +233,12 @@ below:
      - 512
      - The maximum length of ``x-datadog-tags`` header allowed in the Datadog propagation style. Must be a value between 0 to 512. If 0, propagation of ``x-datadog-tags`` is disabled.
 
+       .. _dd-trace-partial-flush-enabled:
+   * - ``DD_TRACE_PARTIAL_FLUSH_ENABLED``
+     - Boolean
+     - True
+     - Prevents large payloads being sent to APM.
+
        .. _dd-profiling-enabled:
    * - ``DD_PROFILING_ENABLED``
      - Boolean

tests/commands/test_runner.py

@@ -410,7 +410,7 @@ def test_info_no_configs():
     Log injection enabled: False
     Health metrics enabled: False
     Priority sampling enabled: True
-    Partial flushing enabled: False
+    Partial flushing enabled: True
     Partial flush minimum number of spans: 500
     \x1b[92m\x1b[1mTagging:\x1b[0m
     DD Service: None

tests/contrib/asyncio/test_tracer_safety.py

@@ -45,11 +45,12 @@ async def coro():
         with tracer.trace("coroutine"):
             await asyncio.sleep(0.01)
 
-    futures = [asyncio.ensure_future(coro()) for x in range(1000)]
+    # partial flushing is enabled, ensure the number of spans generated is less than 500
+    futures = [asyncio.ensure_future(coro()) for x in range(400)]
     for future in futures:
         await future
 
     # the trace is wrong but the Context is finished
     traces = tracer.pop_traces()
     assert 1 == len(traces)
-    assert 1000 == len(traces[0])
+    assert 400 == len(traces[0])

tests/integration/test_integration.py

@@ -1,3 +1,4 @@
+# -*- coding: utf-8 -*-
 import itertools
 import logging
 import os
@@ -219,6 +220,47 @@ def test_metrics(encoding, monkeypatch):
 
     with override_global_config(dict(health_metrics_enabled=True)):
         t = Tracer()
+        assert t._partial_flush_min_spans == 500
+        statsd_mock = mock.Mock()
+        t._writer.dogstatsd = statsd_mock
+        assert t._writer._report_metrics
+        with mock.patch("ddtrace.internal.writer.log") as log:
+            for _ in range(5):
+                spans = []
+                for i in range(3000):
+                    spans.append(t.trace("op"))
+                # Since _partial_flush_min_spans is set to 500 we will flush spans in 6 batches
+                # each batch will contain 500 spans
+                for s in spans:
+                    s.finish()
+
+            t.shutdown()
+            log.warning.assert_not_called()
+            log.error.assert_not_called()
+
+        statsd_mock.distribution.assert_has_calls(
+            [
+                mock.call("datadog.tracer.http.sent.bytes", AnyInt()),
+                mock.call("datadog.tracer.http.sent.traces", 30),
+                mock.call("datadog.tracer.writer.accepted.traces", 30, tags=[]),
+                mock.call("datadog.tracer.buffer.accepted.traces", 30, tags=[]),
+                mock.call("datadog.tracer.buffer.accepted.spans", 15000, tags=[]),
+                mock.call("datadog.tracer.http.requests", 1, tags=[]),
+                mock.call("datadog.tracer.http.sent.bytes", AnyInt(), tags=[]),
+            ],
+            any_order=True,
+        )
+
+
+@allencodings
+def test_metrics_partial_flush_disabled(encoding, monkeypatch):
+    monkeypatch.setenv("DD_TRACE_API_VERSION", encoding)
+
+    with override_global_config(dict(health_metrics_enabled=True)):
+        t = Tracer()
+        t.configure(
+            partial_flush_enabled=False,
+        )
         statsd_mock = mock.Mock()
         t._writer.dogstatsd = statsd_mock
         assert t._writer._report_metrics
@@ -248,8 +290,37 @@ def test_metrics(encoding, monkeypatch):
 @allencodings
 def test_single_trace_too_large(encoding, monkeypatch):
     monkeypatch.setenv("DD_TRACE_API_VERSION", encoding)
+    # setting writer interval to 5 seconds so that buffer can fit larger traces
+    monkeypatch.setenv("DD_TRACE_WRITER_INTERVAL_SECONDS", "5.0")
 
     t = Tracer()
+    assert t._partial_flush_enabled is True
+    with mock.patch("ddtrace.internal.writer.log") as log:
+        key = "a" * 250
+        with t.trace("huge"):
+            for i in range(200000):
+                with t.trace("operation") as s:
+                    # Need to make the strings unique so that the v0.5 encoding doesn’t compress the data
+                    s.set_tag(key + str(i), key + str(i))
+        t.shutdown()
+        log.warning.assert_any_call(
+            "trace buffer (%s traces %db/%db) cannot fit trace of size %db, dropping",
+            AnyInt(),
+            AnyInt(),
+            AnyInt(),
+            AnyInt(),
+        )
+        log.error.assert_not_called()
+
+
+@allencodings
+def test_single_trace_too_large_partial_flush_disabled(encoding, monkeypatch):
+    monkeypatch.setenv("DD_TRACE_API_VERSION", encoding)
+
+    t = Tracer()
+    t.configure(
+        partial_flush_enabled=False,
+    )
     with mock.patch("ddtrace.internal.writer.log") as log:
         with t.trace("huge"):
             for i in range(200000):
@@ -717,7 +788,6 @@ def test_partial_flush_log(run_python_code_in_subprocess, encoding, monkeypatch)
     t = Tracer()
 
     t.configure(
-        partial_flush_enabled=True,
         partial_flush_min_spans=partial_flush_min_spans,
     )