test(debugging): remove flakiness in uploader and encoder tests (#6864)

We remove some residual flakiness in the uploader and encoding tests by
making more operations synchronous in tests to remove potential race
conditions.

## Checklist

- [x] Change(s) are motivated and described in the PR description.
- [x] Testing strategy is described if automated tests are not included
in the PR.
- [x] Risk is outlined (performance impact, potential for breakage,
maintainability, etc).
- [x] Change is maintainable (easy to change, telemetry, documentation).
- [x] [Library release note
guidelines](https://ddtrace.readthedocs.io/en/stable/releasenotes.html)
are followed. If no release note is required, add label
`changelog/no-changelog`.
- [x] Documentation is included (in-code, generated user docs, [public
corp docs](https://github.com/DataDog/documentation/)).
- [x] Backport labels are set (if
[applicable](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting))

## Reviewer Checklist

- [x] Title is accurate.
- [x] No unnecessary changes are introduced.
- [x] Description motivates each change.
- [x] Avoids breaking
[API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces)
changes unless absolutely necessary.
- [x] Testing strategy adequately addresses listed risk(s).
- [x] Change is maintainable (easy to change, telemetry, documentation).
- [x] Release note makes sense to a user of the library.
- [x] Reviewer has explicitly acknowledged and discussed the performance
implications of this PR as reported in the benchmarks PR comment.
- [x] 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)
- [x] If this PR touches code that signs or publishes builds or
packages, or handles credentials of any kind, I've requested a review
from `@DataDog/security-design-and-guidance`.
- [x] This PR doesn't touch any of that.

Author: P403n1x87

tests/debugging/test_debugger.py

@@ -594,7 +594,7 @@ def test_debugger_line_probe_on_wrapped_function(stuff):
 def test_probe_status_logging(remote_config_worker):
     assert remoteconfig_poller.status == ServiceStatus.STOPPED
 
-    with rcm_endpoint(), debugger(diagnostics_interval=0.2, enabled=True) as d:
+    with rcm_endpoint(), debugger(diagnostics_interval=0, enabled=True) as d:
         d.add_probes(
             create_snapshot_line_probe(
                 probe_id="line-probe-ok",
@@ -625,7 +625,7 @@ def count_status(queue):
 def test_probe_status_logging_reemit_on_modify(remote_config_worker):
     assert remoteconfig_poller.status == ServiceStatus.STOPPED
 
-    with rcm_endpoint(), debugger(diagnostics_interval=0.2, enabled=True) as d:
+    with rcm_endpoint(), debugger(diagnostics_interval=0, enabled=True) as d:
         d.add_probes(
             create_snapshot_line_probe(
                 version=1,
@@ -670,7 +670,7 @@ def versions(queue, status):
 def test_debugger_function_probe_duration(duration):
     from tests.submod.stuff import durationstuff
 
-    with debugger(poll_interval=0.1) as d:
+    with debugger(poll_interval=0) as d:
         d.add_probes(
             create_snapshot_function_probe(
                 probe_id="duration-probe",
@@ -688,7 +688,7 @@ def test_debugger_function_probe_duration(duration):
 def test_debugger_condition_eval_then_rate_limit():
     from tests.submod.stuff import Stuff
 
-    with debugger(upload_flush_interval=0.1) as d:
+    with debugger(upload_flush_interval=float("inf")) as d:
         d.add_probes(
             create_snapshot_line_probe(
                 probe_id="foo",
@@ -718,7 +718,7 @@ def test_debugger_condition_eval_then_rate_limit():
 def test_debugger_condition_eval_error_get_reported_once():
     from tests.submod.stuff import Stuff
 
-    with debugger(upload_flush_interval=0.1) as d:
+    with debugger(upload_flush_interval=float("inf")) as d:
         d.add_probes(
             create_snapshot_line_probe(
                 probe_id="foo",
@@ -846,7 +846,7 @@ def __init__(self, age, name):
 def test_debugger_log_live_probe_generate_messages():
     from tests.submod.stuff import Stuff
 
-    with debugger(upload_flush_interval=0.1) as d:
+    with debugger(upload_flush_interval=float("inf")) as d:
         d.add_probes(
             create_log_line_probe(
                 probe_id="foo",
@@ -989,7 +989,7 @@ def child():
 def test_debugger_modified_probe():
     from tests.submod.stuff import Stuff
 
-    with debugger(upload_flush_interval=0.1) as d:
+    with debugger(upload_flush_interval=float("inf")) as d:
         d.add_probes(
             create_log_line_probe(
                 probe_id="foo",

tests/debugging/test_uploader.py

@@ -29,15 +29,15 @@ def __init__(self, size=1 << 10, interval=1):
         )
 
     def on_full(self, item, encoded):
-        self.upload()
+        self.periodic()
 
 
 def test_uploader_batching():
-    with ActiveBatchJsonEncoder(interval=0.1) as uploader:
+    with ActiveBatchJsonEncoder(interval=float("inf")) as uploader:
         for _ in range(5):
             uploader._encoder.put("hello")
             uploader._encoder.put("world")
-            uploader.awake()
+            uploader.periodic()
 
         for _ in range(5):
             assert uploader.queue.get(timeout=1) == "[hello,world]", "iteration %d" % _
@@ -46,7 +46,7 @@ def test_uploader_batching():
 @pytest.mark.xfail(condition=PY2, reason="This test is flaky on Python 2")
 def test_uploader_full_buffer():
     size = 1 << 8
-    with ActiveBatchJsonEncoder(size=size, interval=0.1) as uploader:
+    with ActiveBatchJsonEncoder(size=size, interval=float("inf")) as uploader:
         item = "hello" * 10
         n = size // len(item)
         assert n
@@ -60,5 +60,5 @@ def test_uploader_full_buffer():
         assert uploader.queue.qsize() == 0
 
         # wakeup to mimic next interval
-        uploader.awake()
+        uploader.periodic()
         assert uploader.queue.qsize() == 0