From 24764c7cf1b0ab6a542d8b8093d3b6bad78a49bc Mon Sep 17 00:00:00 2001 From: Anna Garcia Date: Thu, 10 Sep 2026 21:40:11 -0400 Subject: [PATCH 1/4] feat(traces): span export queue with retries Adds SpanExporter, the in-memory queue ended spans wait in until they are batched and sent, separate from the events queue. A timer flushes every flush_interval and a full batch flushes at once; only one flush runs at a time. A full queue drops the incoming span, never a queued parent. Failures back off exponentially with jitter, floored by Retry-After (clamped to 30 s, extended by a later deadline but never shortened), and automatic sends pause while backing off. A batch refused across 8 backoff windows is dropped, a 413 halves the batch and ramps back, and other 4xx drop it. flush(timeout) always sends the first batch, so a serverless handler with no budget left still ships spans. Not reachable from the client. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01TkZAsCciW4PV8ZdcCHmAbA --- posthog/test/tracing/helpers.py | 53 +- posthog/test/tracing/test_export.py | 876 ++++++++++++++++++++++++++++ posthog/tracing/_export.py | 463 +++++++++++++++ 3 files changed, 1390 insertions(+), 2 deletions(-) create mode 100644 posthog/test/tracing/test_export.py create mode 100644 posthog/tracing/_export.py diff --git a/posthog/test/tracing/helpers.py b/posthog/test/tracing/helpers.py index e9786bac..edaed8b9 100644 --- a/posthog/test/tracing/helpers.py +++ b/posthog/test/tracing/helpers.py @@ -1,4 +1,4 @@ -"""Shared fakes for the tracing pipeline tests.""" +"""Shared fakes for the tracing pipeline and export tests.""" import threading import time @@ -8,13 +8,18 @@ import pytest +from posthog.tracing import _export as export_module from posthog.tracing._config import resolve_traces_config from posthog.tracing._drops import DropLog +from posthog.tracing._export import SpanExporter from posthog.tracing._pipeline import PostHogTraces +from posthog.tracing._transport import SendOutcome TRACE_ID = "4bf92f3577b34da6a3ce929d0e0e4736" SPAN_ID = "00f067aa0ba902b7" +RealTimer = threading.Timer + class FakeTimer: """Records the delay it was armed with; fires only when a test says so.""" @@ -39,6 +44,21 @@ def fire(self): self.fn() +class FakeSender: + def __init__(self, *outcomes): + self.outcomes = list(outcomes) + self.payloads: list = [] + + def __call__(self, client, payload): + self.payloads.append(payload) + if len(self.outcomes) > 1: + return self.outcomes.pop(0) + return self.outcomes[0] if self.outcomes else SendOutcome("ok") + + def batches(self): + return [p["resourceSpans"][0]["scopeSpans"][0]["spans"] for p in self.payloads] + + class RecordingExporter: """Stands in for the export queue: keeps every record it is handed.""" @@ -70,6 +90,13 @@ def fake_timers(): yield FakeTimer +@pytest.fixture(autouse=True) +def no_jitter(): + # Backoff delays are asserted exactly; TestJitter covers the spread. + with mock.patch.object(export_module, "_draw_jitter", return_value=1.0): + yield + + @pytest.fixture def clock(): state = {"now": 1000.0} @@ -90,5 +117,27 @@ def make(client=None, context=None, **config): return pipeline, exporter, active +def make_traces(sender=None, client=None, context=None, **config): + """A pipeline over a real exporter whose sender is ``sender``.""" + config.setdefault("flush_interval", 5) + client = client or SimpleNamespace(disabled=False, send=True) + sender = sender or FakeSender(SendOutcome("ok")) + active: ContextVar = ContextVar("active", default=None) + resolved = resolve_traces_config(config) + drops = DropLog(resolved.flush_interval) + pipeline = PostHogTraces( + client, + resolved, + lambda: context or {}, + active, + SpanExporter(client, resolved, drops, send=sender), + drops, + ) + return pipeline, sender, active + + def queued(pipeline): - return pipeline._exporter.records + exporter = pipeline._exporter + if isinstance(exporter, RecordingExporter): + return exporter.records + return exporter._queue diff --git a/posthog/test/tracing/test_export.py b/posthog/test/tracing/test_export.py new file mode 100644 index 00000000..555b1d13 --- /dev/null +++ b/posthog/test/tracing/test_export.py @@ -0,0 +1,876 @@ +import threading +from types import SimpleNamespace +from unittest import mock + +import pytest + +from posthog.test.tracing.helpers import ( + FakeSender, + FakeTimer, + RealTimer, + clock, + fake_timers, + make_traces, + no_jitter, + queued, +) +from posthog.tracing import _export as export_module +from posthog.tracing._config import DEFAULT_MAX_EXPORT_BATCH_SIZE +from posthog.tracing._export import MAX_RETRIES_PER_BATCH, MAX_RETRY_AFTER_SECONDS +from posthog.tracing._span import NOOP_SPAN +from posthog.tracing._transport import TOO_LARGE_LOCALLY, SendOutcome + +__all__ = ["clock", "fake_timers", "no_jitter"] + + +class TestExport: + def test_arms_an_immediate_flush_when_the_queue_reaches_the_batch_size(self): + pipeline, sender, _ = make_traces(max_export_batch_size=2) + pipeline.start_span("a").end() + assert FakeTimer.instances[-1].delay == 5 + pipeline.start_span("b").end() + timer = FakeTimer.instances[-1] + assert timer.delay == 0 + timer.fire() + assert len(sender.batches()) == 1 + assert len(sender.batches()[0]) == 2 + assert queued(pipeline) == [] + + def test_flushes_on_the_interval_timer(self): + pipeline, sender, _ = make_traces() + pipeline.start_span("a").end() + timer = FakeTimer.instances[-1] + assert timer.delay == 5 and timer.started + timer.fire() + assert len(sender.payloads) == 1 + + def test_sends_one_resource_and_one_scope_per_batch(self): + pipeline, sender, _ = make_traces(service_name="api") + pipeline.start_span("a").end() + pipeline.flush() + payload = sender.payloads[0] + assert len(payload["resourceSpans"]) == 1 + assert len(payload["resourceSpans"][0]["scopeSpans"]) == 1 + attrs = { + kv["key"]: kv["value"] + for kv in payload["resourceSpans"][0]["resource"]["attributes"] + } + assert attrs["service.name"] == {"stringValue": "api"} + assert attrs["telemetry.sdk.name"] == {"stringValue": "posthog-python"} + + def test_splits_a_backlog_across_batches(self): + pipeline, sender, _ = make_traces(max_export_batch_size=2) + for i in range(5): + pipeline.start_span(str(i)).end() + pipeline.flush() + assert [len(b) for b in sender.batches()] == [2, 2, 1] + assert queued(pipeline) == [] + + def test_drains_spans_that_arrive_during_a_pass(self): + pipeline, sender, _ = make_traces() + + def send_and_enqueue(client, payload): + sender.payloads.append(payload) + if len(sender.payloads) == 1: + pipeline.start_span("late").end() + return SendOutcome("ok") + + pipeline._exporter._send = send_and_enqueue + pipeline.start_span("early").end() + pipeline.flush() + assert [[s["name"] for s in b] for b in sender.batches()] == [ + ["early"], + ["late"], + ] + + def test_does_not_re_post_on_every_span_end_while_a_flush_is_failing(self): + pipeline, sender, _ = make_traces( + sender=FakeSender(SendOutcome("retry-later")), max_export_batch_size=1 + ) + pipeline.start_span("a").end() + FakeTimer.instances[-1].fire() + assert len(sender.payloads) == 1 + pipeline.start_span("b").end() + assert FakeTimer.instances[-1].delay > 0 + + def test_a_disabled_client_discards_the_queue_instead_of_exporting(self): + client = SimpleNamespace(disabled=False, send=True) + pipeline, sender, _ = make_traces(client=client) + pipeline.start_span("a").end() + client.disabled = True + pipeline.flush() + assert sender.payloads == [] + assert queued(pipeline) == [] + + def test_a_disabled_discard_takes_the_failed_batchs_budget_with_it(self, clock): + client = SimpleNamespace(disabled=False, send=True) + sender = FakeSender(SendOutcome("retry-later"), SendOutcome("ok")) + pipeline, _, _ = make_traces(client=client, sender=sender) + pipeline.start_span("a").end() + pipeline.flush() + assert pipeline._exporter._head_batch_failures == 1 + client.disabled = True + pipeline.flush() + client.disabled = False + assert pipeline._exporter._head_batch_failures == 0 + assert pipeline._exporter._consecutive_failures == 0 + + def test_runs_at_most_one_follow_up_pass(self): + pipeline, sender, _ = make_traces() + + def send_and_enqueue(client, payload): + sender.payloads.append(payload) + pipeline.start_span("late").end() + return SendOutcome("ok") + + pipeline._exporter._send = send_and_enqueue + pipeline.start_span("early").end() + pipeline.flush() + assert len(sender.payloads) == 2 + assert len(queued(pipeline)) == 1 + + def test_stops_starting_requests_once_the_deadline_passes(self, clock): + pipeline, sender, _ = make_traces(max_export_batch_size=1) + + def send_slowly(client, payload): + sender.payloads.append(payload) + clock["now"] += 1 + return SendOutcome("ok") + + pipeline._exporter._send = send_slowly + for _ in range(3): + pipeline.start_span("a").end() + pipeline.flush(timeout=0.1) + assert len(sender.payloads) == 1 + assert len(queued(pipeline)) == 2 + + def test_a_spent_budget_still_ships_one_batch(self): + pipeline, sender, _ = make_traces() + pipeline.start_span("a").end() + pipeline.flush(timeout=0.0) + assert len(sender.payloads) == 1 + assert queued(pipeline) == [] + + def test_returns_without_draining_when_another_flush_holds_the_lock_past_the_deadline( + self, + ): + pipeline, sender, _ = make_traces() + pipeline.start_span("a").end() + timer = FakeTimer.instances[-1] + pipeline._exporter._flush_lock.acquire() + try: + pipeline.flush(timeout=0.01) + finally: + pipeline._exporter._flush_lock.release() + assert sender.payloads == [] + assert pipeline._exporter._flush_timer is timer + assert not timer.cancelled + + def test_re_arms_after_a_timer_that_failed_to_start(self): + class FlakyTimer(FakeTimer): + failed = False + + def start(self): + if self.delay == 0 and not FlakyTimer.failed: + FlakyTimer.failed = True + raise RuntimeError("can't start new thread") + super().start() + + with mock.patch.object(threading, "Timer", FlakyTimer): + pipeline, sender, _ = make_traces(max_export_batch_size=3) + for _ in range(3): + pipeline.start_span("a").end() + assert pipeline._exporter._flush_timer is None + pipeline.start_span("b").end() + timer = pipeline._exporter._flush_timer + assert timer is not None and timer.started and timer.delay == 0 + timer.fire() + assert queued(pipeline) == [] + assert [len(b) for b in sender.batches()] == [3, 1] + + +class TestBackgroundDrainThreads: + def test_at_most_one_follow_up_thread_parks_behind_an_in_flight_flush(self): + created = [] + + class RecordingTimer(RealTimer): + def __init__(self, delay, fn): + super().__init__(delay, fn) + created.append(self) + + sending = threading.Event() + release = threading.Event() + payloads = [] + + def blocking_send(client, payload): + payloads.append(payload) + sending.set() + release.wait(5) + return SendOutcome("ok") + + with mock.patch.object(threading, "Timer", RecordingTimer): + pipeline, _, _ = make_traces( + sender=blocking_send, + max_export_batch_size=10, + max_queue_size=10_000, + flush_interval=60, + ) + for _ in range(10): + pipeline.start_span("a").end() + assert sending.wait(5) + timers_before = len(created) + for _ in range(300): + pipeline.start_span("b").end() + assert len(created) - timers_before <= 2 + release.set() + for timer in list(created): + timer.join(5) + assert queued(pipeline) == [] + assert ( + sum(len(p["resourceSpans"][0]["scopeSpans"][0]["spans"]) for p in payloads) + == 310 + ) + + +class TestExportFailures: + def test_halves_the_batch_and_resends_the_same_spans_on_413(self): + sender = FakeSender( + SendOutcome("too-large"), SendOutcome("ok"), SendOutcome("ok") + ) + pipeline, _, _ = make_traces(sender=sender, max_export_batch_size=4) + for i in range(4): + pipeline.start_span(str(i)).end() + pipeline.flush() + assert [[s["name"] for s in b] for b in sender.batches()] == [ + ["0", "1", "2", "3"], + ["0", "1"], + ["2", "3"], + ] + + def test_shrinks_below_the_queue_depth_rather_than_resending_the_same_body(self): + sender = FakeSender(SendOutcome("too-large"), SendOutcome("ok")) + pipeline, _, _ = make_traces(sender=sender, max_export_batch_size=100) + for i in range(4): + pipeline.start_span(str(i)).end() + pipeline.flush() + assert [len(b) for b in sender.batches()] == [4, 2, 2] + + def test_ramps_the_batch_size_back_up_after_a_413_shrink(self): + sender = FakeSender(SendOutcome("too-large"), SendOutcome("ok")) + pipeline, _, _ = make_traces(sender=sender, max_export_batch_size=8) + for i in range(8): + pipeline.start_span(str(i)).end() + pipeline.flush() + assert pipeline._exporter._max_export_batch_size == 6 + assert [len(b) for b in sender.batches()] == [8, 4, 4] + + def test_a_batch_measured_too_large_locally_splits_only_that_drain(self): + sender = FakeSender(TOO_LARGE_LOCALLY, SendOutcome("ok")) + pipeline, _, _ = make_traces(sender=sender, max_export_batch_size=8) + for i in range(8): + pipeline.start_span(str(i)).end() + pipeline.flush() + assert [len(b) for b in sender.batches()] == [8, 4, 4] + # The next drain starts at full size; after a 413 it would still be + # ramping back up from 4. + assert pipeline._exporter._max_export_batch_size == 8 + for i in range(8): + pipeline.start_span(str(i)).end() + pipeline.flush() + assert [len(b) for b in sender.batches()][3:] == [8] + + def test_drops_a_single_span_the_server_rejects_as_too_large(self, caplog): + caplog.set_level("WARNING", logger="posthog") + pipeline, sender, _ = make_traces(sender=FakeSender(SendOutcome("too-large"))) + pipeline.start_span("huge").end() + pipeline.flush() + assert queued(pipeline) == [] + assert "too large" in caplog.text + + def test_keeps_spans_queued_on_a_retriable_failure(self): + pipeline, sender, _ = make_traces(sender=FakeSender(SendOutcome("retry-later"))) + pipeline.start_span("a").end() + pipeline.flush() + assert len(queued(pipeline)) == 1 + assert len(sender.payloads) == 1 + + def test_does_not_resend_a_refused_batch_in_the_same_flush(self): + sender = FakeSender(SendOutcome("ok"), SendOutcome("retry-later", 30)) + pipeline, _, _ = make_traces(sender=sender, max_export_batch_size=2) + for _ in range(4): + pipeline.start_span("a").end() + pipeline.flush() + assert len(sender.payloads) == 2 + assert len(queued(pipeline)) == 2 + + def test_backs_off_exponentially_while_sends_keep_failing(self, clock): + pipeline, _, _ = make_traces(sender=FakeSender(SendOutcome("retry-later"))) + pipeline.start_span("a").end() + delays = [] + for _ in range(6): + pipeline.flush() + delays.append(FakeTimer.instances[-1].delay) + clock["now"] += 100 + assert delays == [5, 10, 20, 30, 30, 30] + + def test_returns_to_the_base_interval_after_a_send_succeeds(self, clock): + sender = FakeSender( + SendOutcome("retry-later"), SendOutcome("retry-later"), SendOutcome("ok") + ) + pipeline, _, _ = make_traces(sender=sender) + pipeline.start_span("a").end() + pipeline.flush() + clock["now"] += 100 + pipeline.flush() + assert FakeTimer.instances[-1].delay == 10 + clock["now"] += 100 + pipeline.flush() + pipeline.start_span("b").end() + assert FakeTimer.instances[-1].delay == 5 + + def test_drops_a_poison_batch_rather_than_wedging_the_queue(self): + sender = FakeSender(SendOutcome("fatal"), SendOutcome("ok")) + pipeline, _, _ = make_traces(sender=sender, max_export_batch_size=1) + pipeline.start_span("poison").end() + pipeline.start_span("fine").end() + pipeline.flush() + assert [[s["name"] for s in b] for b in sender.batches()] == [ + ["poison"], + ["fine"], + ] + assert queued(pipeline) == [] + + def test_does_not_surface_a_transport_failure_through_span_end(self): + def explode(client, payload): + raise RuntimeError("transport broke") + + pipeline, _, _ = make_traces(sender=explode, max_export_batch_size=1) + pipeline.start_span("a").end() + FakeTimer.instances[-1].fire() + + def test_never_returns_a_span_it_failed_to_encode(self): + pipeline, sender, _ = make_traces() + pipeline.start_span("a").end() + with mock.patch.object( + export_module, "build_otlp_span", side_effect=RuntimeError("bad") + ): + pipeline.flush() + assert sender.payloads == [] + assert queued(pipeline) == [] + + +class TestDroppedBatchesEndTheFailureSequence: + @pytest.mark.parametrize( + "outcome", [SendOutcome("fatal"), SendOutcome("too-large")] + ) + def test_a_dropped_batch_re_enables_the_depth_trigger(self, clock, outcome): + # A batch that is dropped rather than retried leaves nothing to back + # off for, so the next full batch goes out without waiting. + sender = FakeSender(SendOutcome("retry-later"), outcome, SendOutcome("ok")) + pipeline, _, _ = make_traces(sender=sender, max_export_batch_size=1) + pipeline.start_span("a").end() + pipeline.flush() + clock["now"] += 100 + pipeline.flush() + assert pipeline._exporter._consecutive_failures == 0 + pipeline.start_span("b").end() + assert FakeTimer.instances[-1].delay == 0 + + +class TestRetryBudget: + def test_drops_a_batch_the_endpoint_keeps_refusing_and_moves_to_the_next_one( + self, clock + ): + sender = FakeSender( + *([SendOutcome("retry-later")] * MAX_RETRIES_PER_BATCH), SendOutcome("ok") + ) + pipeline, _, _ = make_traces(sender=sender, max_export_batch_size=1) + pipeline.start_span("stuck").end() + pipeline.start_span("next").end() + for _ in range(MAX_RETRIES_PER_BATCH): + pipeline.flush() + clock["now"] += 100 + assert [[s["name"] for s in b] for b in sender.batches()][-1] == ["next"] + assert queued(pipeline) == [] + + def test_charges_the_budget_once_per_backoff_window_not_per_attempt(self, clock): + pipeline, _, _ = make_traces(sender=FakeSender(SendOutcome("retry-later"))) + pipeline.start_span("a").end() + for _ in range(MAX_RETRIES_PER_BATCH * 3): + pipeline.flush() + assert pipeline._exporter._head_batch_failures == 1 + assert len(queued(pipeline)) == 1 + + def test_gives_the_halved_batch_its_own_budget_after_a_413(self, clock): + sender = FakeSender( + SendOutcome("retry-later"), + SendOutcome("too-large"), + SendOutcome("retry-later"), + ) + pipeline, _, _ = make_traces(sender=sender, max_export_batch_size=2) + pipeline.start_span("a").end() + pipeline.start_span("b").end() + pipeline.flush() + clock["now"] += 100 + pipeline.flush() + assert pipeline._exporter._head_batch_failures == 1 + assert pipeline._exporter._head_batch_size == 1 + + def test_does_not_let_a_failing_head_grow_to_sweep_in_fresh_spans(self, clock): + sender = FakeSender(SendOutcome("retry-later")) + pipeline, _, _ = make_traces(sender=sender, max_export_batch_size=4) + pipeline.start_span("a").end() + pipeline.flush() + pipeline.start_span("b").end() + clock["now"] += 100 + pipeline.flush() + assert [len(b) for b in sender.batches()] == [1, 1] + + +class TestRetryAfter: + def test_lengthens_the_backoff_when_the_endpoint_asks_for_a_longer_wait( + self, clock + ): + pipeline, _, _ = make_traces(sender=FakeSender(SendOutcome("retry-later", 20))) + pipeline.start_span("a").end() + pipeline.flush() + assert FakeTimer.instances[-1].delay == 20 + + def test_never_shortens_the_backoff(self, clock): + pipeline, _, _ = make_traces( + sender=FakeSender(SendOutcome("retry-later", 1)), flush_interval=10 + ) + pipeline.start_span("a").end() + pipeline.flush() + assert FakeTimer.instances[-1].delay == 10 + + def test_clamps_an_oversized_retry_after(self, clock): + pipeline, _, _ = make_traces( + sender=FakeSender(SendOutcome("retry-later", 3600)) + ) + pipeline.start_span("a").end() + pipeline.flush() + assert FakeTimer.instances[-1].delay == MAX_RETRY_AFTER_SECONDS + + def test_an_explicit_flush_inside_the_window_still_sends_but_is_not_charged( + self, clock + ): + pipeline, sender, _ = make_traces( + sender=FakeSender(SendOutcome("retry-later", 30)) + ) + pipeline.start_span("a").end() + pipeline.flush() + clock["now"] += 1 + pipeline.flush() + assert len(sender.payloads) == 2 + assert pipeline._exporter._head_batch_failures == 1 + + def test_a_refusal_inside_an_open_window_is_not_charged(self, clock): + # Charged at t=0 with a 5s backoff; an uncharged refusal at t=2 opens a + # 30s window. A flush at t=6 is past the charge point but inside the + # window the endpoint asked for, so it is caller-driven and exempt. + sender = FakeSender( + SendOutcome("retry-later"), + SendOutcome("retry-later", 30), + SendOutcome("retry-later"), + ) + pipeline, _, _ = make_traces(sender=sender) + pipeline.start_span("a").end() + pipeline.flush() + clock["now"] += 2 + pipeline.flush() + clock["now"] += 4 + pipeline.flush() + assert len(sender.payloads) == 3 + assert pipeline._exporter._head_batch_failures == 1 + + def test_a_longer_retry_after_mid_window_extends_the_deadline(self, clock): + window = export_module._RetryAfterWindow() + window.record(SendOutcome("retry-later", 10)) + clock["now"] += 5 + window.record(SendOutcome("retry-later", 20)) + assert window.remaining() == 20 + + def test_a_shorter_retry_after_mid_window_does_not_cut_the_wait(self, clock): + window = export_module._RetryAfterWindow() + window.record(SendOutcome("retry-later", 25)) + clock["now"] += 5 + window.record(SendOutcome("retry-later", 1)) + assert window.remaining() == 20 + + def test_repeated_refusals_cannot_hold_the_window_past_the_ceiling(self, clock): + window = export_module._RetryAfterWindow() + window.record(SendOutcome("retry-later", MAX_RETRY_AFTER_SECONDS)) + clock["now"] += MAX_RETRY_AFTER_SECONDS - 10 + window.record(SendOutcome("retry-later", MAX_RETRY_AFTER_SECONDS)) + # Extended only as far as the ceiling measured from first install. + assert window.remaining() == 10 + clock["now"] += 10 + assert not window.is_open() + # Closed at the ceiling; the next refusal installs a new window. + window.record(SendOutcome("retry-later", 20)) + assert window.remaining() == 20 + + def test_a_refusal_naming_no_wait_leaves_an_open_window_alone(self, clock): + window = export_module._RetryAfterWindow() + window.record(SendOutcome("retry-later", 20)) + clock["now"] += 5 + window.record(SendOutcome("retry-later")) + assert window.remaining() == 15 + + def test_suppresses_the_depth_trigger_while_the_window_is_open(self, clock): + # The failure count is back to 0 after the single-span drop, so only + # the open window holds the depth trigger back. + sender = FakeSender( + SendOutcome("retry-later", 30), SendOutcome("too-large"), SendOutcome("ok") + ) + pipeline, _, _ = make_traces(sender=sender, max_export_batch_size=1) + pipeline.start_span("a").end() + pipeline.flush() + clock["now"] += 1 + pipeline.flush() + assert pipeline._exporter._consecutive_failures == 0 + assert pipeline._exporter._retry_after.is_open() + pipeline.start_span("b").end() + assert FakeTimer.instances[-1].delay > 0 + + def test_a_success_closes_the_window(self, clock): + sender = FakeSender(SendOutcome("retry-later", 30), SendOutcome("ok")) + pipeline, _, _ = make_traces(sender=sender) + pipeline.start_span("a").end() + pipeline.flush() + clock["now"] += 1 + pipeline.flush() + assert not pipeline._exporter._retry_after.is_open() + + def test_a_non_retriable_response_closes_the_window(self, clock): + sender = FakeSender(SendOutcome("retry-later", 30), SendOutcome("fatal")) + pipeline, _, _ = make_traces(sender=sender) + pipeline.start_span("a").end() + pipeline.flush() + clock["now"] += 1 + pipeline.flush() + assert not pipeline._exporter._retry_after.is_open() + + def test_a_too_large_response_leaves_the_window_open(self, clock): + sender = FakeSender(SendOutcome("retry-later", 30), SendOutcome("too-large")) + pipeline, _, _ = make_traces(sender=sender) + pipeline.start_span("a").end() + pipeline.flush() + clock["now"] += 1 + pipeline.flush() + assert pipeline._exporter._retry_after.remaining() == 29 + + def test_retiring_a_batch_inside_the_window_sends_nothing_more_that_pass( + self, clock + ): + sender = FakeSender(SendOutcome("retry-later", 30)) + pipeline, _, _ = make_traces(sender=sender, max_export_batch_size=1) + pipeline.start_span("stuck").end() + pipeline.start_span("next").end() + pipeline._exporter._head_batch_failures = MAX_RETRIES_PER_BATCH - 1 + pipeline._exporter._head_batch_size = 1 + pipeline.flush() + assert [[s["name"] for s in b] for b in sender.batches()] == [["stuck"]] + assert [r.name for r in queued(pipeline)] == ["next"] + + def test_close_cancels_the_timer(self, clock): + pipeline, _, _ = make_traces(sender=FakeSender(SendOutcome("retry-later", 30))) + pipeline.start_span("a").end() + pipeline.flush() + pipeline.close() + assert pipeline._exporter._flush_timer is None + + def test_a_timer_superseded_while_waiting_for_the_flush_lock_does_not_send( + self, clock + ): + # A depth-triggered timer queued behind an in-flight flush that then + # installed a Retry-After window must not send inside that window. + sender = FakeSender(SendOutcome("retry-later", 10)) + pipeline, _, _ = make_traces(sender=sender, max_export_batch_size=1) + exporter = pipeline._exporter + stale = [] + + def send_and_trigger(client, payload): + sender.payloads.append(payload) + pipeline.start_span("mid-flight").end() + stale.append(exporter._flush_timer) + return SendOutcome("retry-later", 10) + + exporter._send = send_and_trigger + pipeline.start_span("a").end() + pipeline.flush() + assert stale[0] is not None and stale[0] is not exporter._flush_timer + exporter.flush(_timer=stale[0]) + assert len(sender.payloads) == 1 + + +class TestTimerRearm: + def test_keeps_a_later_timer_a_span_armed_mid_pass(self, clock): + # Two failures put the backoff at 10s. A span ending mid-pass arms at + # that delay; the pass then succeeds, and its 5s re-arm must not pull + # the pending timer in. + sender = FakeSender(SendOutcome("retry-later")) + pipeline, _, _ = make_traces(sender=sender, flush_interval=5) + exporter = pipeline._exporter + pipeline.start_span("a").end() + pipeline.flush() + clock["now"] += 100 + pipeline.flush() + assert exporter._consecutive_failures == 2 + + def succeed_with_a_span_ending(client, payload): + pipeline.start_span("mid-pass").end() + return SendOutcome("ok") + + exporter._send = succeed_with_a_span_ending + clock["now"] += 100 + pipeline.flush() + live = [t for t in FakeTimer.instances if not t.cancelled] + assert [t.delay for t in live] == [10] + assert live[0] is exporter._flush_timer + + def test_replaces_an_earlier_timer_with_a_longer_backoff(self, clock): + sender = FakeSender(SendOutcome("retry-later", 20)) + pipeline, _, _ = make_traces(sender=sender, flush_interval=5) + exporter = pipeline._exporter + + def fail_with_a_span_ending(client, payload): + pipeline.start_span("mid-pass").end() + return SendOutcome("retry-later", 20) + + exporter._send = fail_with_a_span_ending + pipeline.start_span("a").end() + pipeline.flush() + assert exporter._flush_timer.delay == 20 + assert [t.delay for t in FakeTimer.instances if not t.cancelled] == [20] + + def test_a_depth_trigger_fired_mid_pass_still_drains_at_once(self): + # Two spans end during each send, so a full batch waits after the + # follow-up pass; it must go now, not an interval later. + pipeline, sender, _ = make_traces(max_export_batch_size=2, flush_interval=5) + exporter = pipeline._exporter + + def send_with_spans_ending(client, payload): + sender.payloads.append(payload) + if len(sender.payloads) <= 2: + pipeline.start_span("mid-pass").end() + pipeline.start_span("mid-pass").end() + return SendOutcome("ok") + + exporter._send = send_with_spans_ending + pipeline.start_span("a").end() + pipeline.start_span("b").end() + pipeline.flush() + assert len(sender.payloads) == 2 + assert len(queued(pipeline)) == 2 + assert exporter._flush_timer.delay == 0 + exporter._flush_timer.fire() + assert queued(pipeline) == [] + + +class TestRetryBudgetResets: + def test_a_success_gives_the_next_failure_a_full_budget(self, clock): + sender = FakeSender( + SendOutcome("retry-later"), SendOutcome("ok"), SendOutcome("retry-later") + ) + pipeline, _, _ = make_traces(sender=sender, max_export_batch_size=1) + pipeline.start_span("a").end() + pipeline.flush() + clock["now"] += 100 + pipeline.flush() + pipeline.start_span("b").end() + pipeline.flush() + assert pipeline._exporter._head_batch_failures == 1 + + def test_a_rejected_batch_gives_the_next_one_a_full_budget(self, clock): + sender = FakeSender( + SendOutcome("retry-later"), SendOutcome("fatal"), SendOutcome("retry-later") + ) + pipeline, _, _ = make_traces(sender=sender, max_export_batch_size=1) + pipeline.start_span("a").end() + pipeline.start_span("b").end() + pipeline.flush() + clock["now"] += 100 + pipeline.flush() + assert [r.name for r in queued(pipeline)] == ["b"] + assert pipeline._exporter._head_batch_failures == 1 + + def test_retires_the_head_after_its_windows_are_served_out(self, clock): + # Each Retry-After window elapses before the next attempt, so every + # refusal is new evidence and is charged. + sender = FakeSender(SendOutcome("retry-later", 20)) + pipeline, _, _ = make_traces(sender=sender, max_export_batch_size=1) + pipeline.start_span("stuck").end() + for _ in range(MAX_RETRIES_PER_BATCH): + pipeline.flush() + clock["now"] += 100 + assert queued(pipeline) == [] + assert len(sender.payloads) == MAX_RETRIES_PER_BATCH + + +class TestSenderFailures: + def test_a_sender_that_raises_backs_off_like_a_retriable_failure(self, clock): + def explode(client, payload): + raise UnicodeEncodeError("latin-1", "x", 0, 1, "bad header") + + pipeline, _, _ = make_traces(sender=explode, flush_interval=10) + pipeline.start_span("a").end() + pipeline.flush() + assert [r.name for r in queued(pipeline)] == ["a"] + assert pipeline._exporter._consecutive_failures == 1 + assert pipeline._exporter._head_batch_failures == 1 + assert FakeTimer.instances[-1].delay == 10 + + def test_stops_draining_when_the_client_is_disabled_mid_pass(self, caplog): + caplog.set_level("WARNING", logger="posthog") + client = SimpleNamespace(disabled=False, send=True) + pipeline, sender, _ = make_traces(client=client, max_export_batch_size=1) + + def send_and_disable(c, payload): + sender.payloads.append(payload) + client.disabled = True + return SendOutcome("ok") + + pipeline._exporter._send = send_and_disable + for name in ("a", "b", "c"): + pipeline.start_span(name).end() + pipeline.flush() + assert len(sender.payloads) == 1 + assert queued(pipeline) == [] + assert any("the client is disabled" in r.getMessage() for r in caplog.records) + + +class TestJitter: + @pytest.fixture(autouse=True) + def no_jitter(self): + # Overrides the module fixture: these tests draw real jitter. + yield + + def test_spreads_the_backoff_by_up_to_a_quarter(self, clock): + pipeline, _, _ = make_traces( + sender=FakeSender(SendOutcome("retry-later")), flush_interval=10 + ) + pipeline.start_span("a").end() + with mock.patch.object(export_module.random, "random", return_value=0.0): + pipeline.flush() + assert FakeTimer.instances[-1].delay == pytest.approx(7.5) + clock["now"] += 100 + with mock.patch.object(export_module.random, "random", return_value=1.0): + pipeline.flush() + assert FakeTimer.instances[-1].delay == pytest.approx(25) + + def test_retry_after_is_a_floor_under_the_jittered_delay(self, clock): + pipeline, _, _ = make_traces( + sender=FakeSender(SendOutcome("retry-later", 10)), flush_interval=10 + ) + pipeline.start_span("a").end() + with mock.patch.object(export_module.random, "random", return_value=0.0): + pipeline.flush() + assert FakeTimer.instances[-1].delay == 10 + + def test_the_interval_is_not_jittered_when_nothing_has_failed(self): + pipeline, _, _ = make_traces(flush_interval=10) + with mock.patch.object(export_module.random, "random", return_value=0.0): + pipeline.start_span("a").end() + assert FakeTimer.instances[-1].delay == 10 + + +class TestDropAccounting: + def test_warns_once_per_interval_with_the_total_and_every_reason( + self, clock, caplog + ): + caplog.set_level("WARNING", logger="posthog") + pipeline, _, _ = make_traces( + max_live_spans=1, max_export_batch_size=1, max_queue_size=1 + ) + held = pipeline.start_span("held") + pipeline.start_span("refused-1") + pipeline.start_span("refused-2") + warnings = [r for r in caplog.records if "Dropping" in r.getMessage()] + assert len(warnings) == 1 + assert "Dropping 1 span(s)" in warnings[0].getMessage() + clock["now"] += 6 + pipeline.start_span("refused-3") + warnings = [r for r in caplog.records if "Dropping" in r.getMessage()] + assert len(warnings) == 2 + assert "Dropping 2 span(s)" in warnings[1].getMessage() + assert "live-span limit" in warnings[1].getMessage() + held.end() + + def test_every_flush_reports_its_drops_without_waiting_out_the_interval( + self, clock, caplog + ): + caplog.set_level("WARNING", logger="posthog") + pipeline, _, _ = make_traces(sender=FakeSender(SendOutcome("fatal"))) + for _ in range(2): + pipeline.start_span("poison").end() + pipeline.flush() + warnings = [r for r in caplog.records if "Dropping" in r.getMessage()] + assert len(warnings) == 2 + + +class TestResetWarnings: + def test_warns_with_the_count_when_it_discards_queued_spans(self, caplog): + caplog.set_level("WARNING", logger="posthog") + pipeline, _, _ = make_traces() + pipeline.start_span("a").end() + pipeline.start_span("b").end() + pipeline.close() + assert any("Discarding 2 span(s)" in r.getMessage() for r in caplog.records) + + def test_is_silent_when_nothing_was_queued(self, caplog): + caplog.set_level("WARNING", logger="posthog") + pipeline, _, _ = make_traces() + pipeline.close() + assert not caplog.records + + +class TestQueueBound: + def test_drops_the_incoming_span_when_the_queue_is_full(self, caplog): + caplog.set_level("WARNING", logger="posthog") + pipeline, _, _ = make_traces(max_export_batch_size=2, max_queue_size=2) + for name in ("first", "second", "third"): + pipeline.start_span(name).end() + assert [r.name for r in queued(pipeline)] == ["first", "second"] + assert any("the queue is full" in r.getMessage() for r in caplog.records) + + +class TestCloseAndFork: + def test_close_clears_the_queue_and_cancels_the_timer(self): + pipeline, _, _ = make_traces() + pipeline.start_span("done").end() + pipeline.close() + assert queued(pipeline) == [] + assert FakeTimer.instances[-1].cancelled + assert pipeline._exporter._flush_timer is None + + def test_close_abandons_an_in_flight_pass_and_records_nothing_after(self): + pipeline, sender, _ = make_traces(max_export_batch_size=1) + + def send_and_close(client, payload): + sender.payloads.append(payload) + pipeline.close() + assert pipeline.start_span("after-close") is NOOP_SPAN + return SendOutcome("ok") + + pipeline._exporter._send = send_and_close + pipeline.start_span("a").end() + pipeline.start_span("b").end() + pipeline.flush() + assert len(sender.payloads) == 1 + assert queued(pipeline) == [] + + def test_a_forked_child_drops_the_inherited_queue_and_timer(self): + pipeline, _, _ = make_traces() + pipeline.start_span("parent-span").end() + assert queued(pipeline) and pipeline._exporter._flush_timer is not None + pipeline._exporter._max_export_batch_size = 1 + pipeline.reinit_after_fork() + assert queued(pipeline) == [] + assert pipeline._exporter._flush_timer is None + assert ( + pipeline._exporter._max_export_batch_size == DEFAULT_MAX_EXPORT_BATCH_SIZE + ) + pipeline.start_span("child-span").end() + assert [r.name for r in queued(pipeline)] == ["child-span"] diff --git a/posthog/tracing/_export.py b/posthog/tracing/_export.py new file mode 100644 index 00000000..272604d0 --- /dev/null +++ b/posthog/tracing/_export.py @@ -0,0 +1,463 @@ +"""The span export queue: batches ended spans and ships them to ``/i/v1/traces``. + +Separate from the events queue, with its own timer. Only one flush runs at a +time. Failures back off exponentially, floored by any ``Retry-After``, and a +batch refused across ``MAX_RETRIES_PER_BATCH`` backoff windows is dropped so it +cannot hold the queue. +""" + +import logging +import random +import threading +import time +from typing import Any, Callable, List, Optional, Tuple + +from ._config import ResolvedTracesConfig +from ._drops import DropLog +from ._otlp import ( + SpanRecord, + build_otlp_span, + build_resource_attributes, + build_traces_payload, +) +from ._transport import SendOutcome, send_traces_batch + +log = logging.getLogger("posthog") + +MAX_RETRIES_PER_BATCH = 8 + +MAX_FLUSH_BACKOFF_EXPONENT = 6 +MAX_FLUSH_BACKOFF_SECONDS = 30.0 + +# Nothing upstream bounds the header, and an unbounded value would strand the +# queue. The same ceiling as the SDK's own backoff. +MAX_RETRY_AFTER_SECONDS = 30.0 + +# Spread each backoff by up to a quarter, so clients refused together do not +# return together. +FLUSH_BACKOFF_JITTER = 0.25 + +SendFn = Callable[[Any, dict], SendOutcome] + + +class _RetryAfterWindow: + """The wait the endpoint asked for, as a monotonic deadline. + + A later deadline extends the window, capped at ``MAX_RETRY_AFTER_SECONDS`` + from when it was first installed; a shorter one never pulls it in. + """ + + def __init__(self) -> None: + self._until = 0.0 + self._installed_at = 0.0 + + def record(self, outcome: SendOutcome) -> None: + if outcome.kind == "too-large": + return + if outcome.kind != "retry-later": + self.reset() + return + if not outcome.retry_after or outcome.retry_after <= 0: + return + # Read first: a spent window resets `_installed_at`. + is_open = self.is_open() + now = time.monotonic() + asked = min(outcome.retry_after, MAX_RETRY_AFTER_SECONDS) + if not is_open: + self._installed_at = now + self._until = now + asked + return + self._until = max( + self._until, min(now + asked, self._installed_at + MAX_RETRY_AFTER_SECONDS) + ) + + def remaining(self) -> float: + remaining = min( + MAX_RETRY_AFTER_SECONDS, max(0.0, self._until - time.monotonic()) + ) + if remaining == 0: + self.reset() + return remaining + + def is_open(self) -> bool: + return self.remaining() > 0 + + def reset(self) -> None: + self._until = 0.0 + self._installed_at = 0.0 + + +class SpanExporter: + def __init__( + self, + client: Any, + config: ResolvedTracesConfig, + drops: DropLog, + send: SendFn = send_traces_batch, + ) -> None: + self._client = client + self._config = config + self._drops = drops + self._send = send + self._resource_attributes = build_resource_attributes( + config.service_name, + config.service_version, + config.environment, + config.resource_attributes, + ) + + self._lock = threading.Lock() + self._flush_lock = threading.Lock() + self._closed = False + self._queue: List[SpanRecord] = [] + self._flush_timer: Optional[threading.Timer] = None + self._flush_timer_fires_at = 0.0 + self._max_export_batch_size = config.max_export_batch_size + self._consecutive_failures = 0 + # Drawn once per failure, so the timer and the retry-budget charge see + # the same delay. + self._jitter = 1.0 + self._retry_after = _RetryAfterWindow() + self._head_batch_failures = 0 + self._head_batch_size = 0 + self._head_batch_chargeable_at = 0.0 + + def enqueue(self, record: SpanRecord) -> None: + with self._lock: + if self._closed: + self._drops.record(1, "tracing was shut down") + elif len(self._queue) >= self._config.max_queue_size: + # The incoming span goes, not queued ones: those are parents + # whose children may already have shipped. + self._drops.record( + 1, + "the queue is full ({}); raise the flush frequency or reduce " + "span volume".format(self._config.max_queue_size), + ) + else: + self._queue.append(record) + if self._depth_trigger_due_locked(): + self._arm_timer_locked( + 0.0, replace=self._flush_timer_fires_at > time.monotonic() + ) + else: + self._arm_timer_if_idle_locked() + self._drops.warn_if_due() + + def flush( + self, timeout: Optional[float] = None, _timer: Optional[threading.Timer] = None + ) -> None: + """Drain the queue: one pass over what was queued, then one follow-up pass. + + With a ``timeout``, no request starts once it is spent, except the + first. A request already in flight is bounded by the client's timeout. + """ + deadline = None if timeout is None else time.monotonic() + timeout + # -1 is Lock.acquire's unbounded form. + if not self._flush_lock.acquire( + timeout=-1 if timeout is None else max(0.0, timeout) + ): + return + try: + with self._lock: + # A timer that waited behind another flush may have been + # superseded by the backoff that flush armed. + if _timer is not None and _timer is not self._flush_timer: + return + self._clear_timer_locked() + try: + removed, stop = self._drain(deadline) + if ( + removed + and not stop + and self._queue + and (deadline is None or time.monotonic() < deadline) + ): + self._drain(deadline) + finally: + with self._lock: + self._rearm_after_pass_locked() + finally: + self._flush_lock.release() + self._drops.warn_if_due(force=True) + + def close(self) -> None: + """Stop exporting and discard what is still queued. Called at shutdown.""" + with self._lock: + self._closed = True + self._clear_timer_locked() + discarded = len(self._queue) + self._queue = [] + if discarded: + log.warning( + "Discarding %s span(s) that were still queued when tracing was shut " + "down. Call flush() earlier if they matter.", + discarded, + ) + + def warn_if_queued(self) -> None: + with self._lock: + queued = len(self._queue) + if queued: + log.warning( + "%s span(s) were still queued at exit and may not be sent. Call " + "flush() or shutdown() before exit.", + queued, + ) + + def reinit_after_fork(self) -> None: + # The inherited locks may be held by parent threads that do not exist + # in the child, so they are replaced rather than acquired. + self._lock = threading.Lock() + self._flush_lock = threading.Lock() + self._flush_timer = None + self._flush_timer_fires_at = 0.0 + self._queue = [] + self._max_export_batch_size = self._config.max_export_batch_size + self._retry_after.reset() + self._end_failure_sequence_locked() + + def _drain(self, deadline: Optional[float]) -> Tuple[int, bool]: + """One pass over the queue as it stood at the start. + + Returns ``(removed, stop)``; ``stop`` means no follow-up pass now. + """ + with self._lock: + if self._is_closed() or self._discard_if_disabled_locked(): + return 0, True + remaining = len(self._queue) + removed = 0 + sent_any = False + # A batch the SDK measured as too large says nothing about the batches + # after the oversized span is gone, so only this drain is split. + local_cap: Optional[int] = None + while remaining > 0: + with self._lock: + if self._is_closed() or self._discard_if_disabled_locked(): + return removed, True + if not self._queue: + break + # A retried batch cannot grow to take in spans behind it. + cap = ( + min(self._max_export_batch_size, self._head_batch_size) + if self._head_batch_failures + else self._max_export_batch_size + ) + if local_cap is not None: + cap = min(cap, local_cap) + size = max(1, min(cap, remaining, len(self._queue))) + batch = self._queue[:size] + # Read before the send. A send inside an open Retry-After + # window is caller-driven and exempt from the wait, and so from + # the charge. + chargeable = ( + time.monotonic() >= self._head_batch_chargeable_at + and not self._retry_after.is_open() + ) + + spans = self._encode(batch) + if not spans: + with self._lock: + if self._is_closed(): + return removed, True + del self._queue[:size] + self._reset_head_batch_budget_locked() + remaining -= size + removed += size + continue + + # The first request is exempt, so a flush called with no budget left + # (a serverless handler, say) still ships a batch. + if sent_any and deadline is not None and time.monotonic() >= deadline: + return removed, True + sent_any = True + try: + outcome = self._send( + self._client, build_traces_payload(spans, self._resource_attributes) + ) + except Exception: + log.debug("Span batch send failed", exc_info=True) + outcome = SendOutcome("retry-later") + + with self._lock: + if self._is_closed(): + return removed, True + taken, stop, note = self._apply_outcome_locked( + outcome, size, chargeable + ) + if note: + log.debug(note) + if outcome.measured_locally and not taken: + local_cap = max(1, size // 2) + remaining -= taken + removed += taken + if stop: + return removed, True + return removed, False + + def _apply_outcome_locked( + self, outcome: SendOutcome, size: int, chargeable: bool + ) -> Tuple[int, bool, Optional[str]]: + """Settle one response. Returns ``(spans removed, stop the pass, debug note)``.""" + self._retry_after.record(outcome) + + if outcome.kind == "ok": + del self._queue[:size] + self._end_failure_sequence_locked() + if self._max_export_batch_size < self._config.max_export_batch_size: + self._max_export_batch_size += 1 + return size, False, None + + if outcome.kind == "too-large": + if size == 1: + del self._queue[:1] + self._end_failure_sequence_locked() + self._drops.record(1, "it is too large for the ingestion endpoint") + return 1, False, None + # Halve the refused batch, not the configured size: a shallow queue + # would otherwise resend the same body. + halved = max(1, size // 2) + if not outcome.measured_locally: + self._max_export_batch_size = halved + self._reset_head_batch_budget_locked() + return ( + 0, + False, + "Span batch too large; retrying the same spans in batches of {}".format( + halved + ), + ) + + if outcome.kind == "retry-later": + self._consecutive_failures += 1 + self._jitter = _draw_jitter() + self._head_batch_size = size + # One charge per backoff window: a refusal before it elapses is the + # same refusal seen again. + if chargeable: + self._head_batch_failures += 1 + self._head_batch_chargeable_at = ( + time.monotonic() + self._next_flush_delay_locked() + ) + if self._head_batch_failures < MAX_RETRIES_PER_BATCH: + return 0, True, "Span export failed; retrying on the next flush" + del self._queue[:size] + self._end_failure_sequence_locked() + self._drops.record( + size, + "the ingestion endpoint failed {} times in a row".format( + MAX_RETRIES_PER_BATCH + ), + ) + # Dropping the batch does not end the endpoint's wait. + return size, self._retry_after.is_open(), None + + del self._queue[:size] + self._end_failure_sequence_locked() + self._drops.record(size, "the ingestion endpoint rejected the batch") + return size, False, None + + def _is_closed(self) -> bool: + # A method, so the check after each send is not narrowed away. + return self._closed + + def _encode(self, batch: List[SpanRecord]) -> List[dict]: + encoded: List[dict] = [] + for record in batch: + try: + encoded.append(build_otlp_span(record)) + except Exception: + log.debug("Failed to encode a span; dropping it", exc_info=True) + self._drops.record(1, "its attributes could not be encoded") + return encoded + + def _discard_if_disabled_locked(self) -> bool: + if not getattr(self._client, "disabled", False): + return False + # Spans carry person and session ids: once the client is disabled, + # nothing queued may export. + if self._queue: + self._drops.record(len(self._queue), "the client is disabled") + self._queue = [] + self._end_failure_sequence_locked() + return True + + def _end_failure_sequence_locked(self) -> None: + self._consecutive_failures = 0 + self._jitter = 1.0 + self._reset_head_batch_budget_locked() + + def _reset_head_batch_budget_locked(self) -> None: + self._head_batch_failures = 0 + self._head_batch_chargeable_at = 0.0 + + def _next_flush_delay_locked(self) -> float: + """The flush interval, doubled per failure up to 30 s and jittered, + floored by Retry-After.""" + exponent = min( + max(0, self._consecutive_failures - 1), MAX_FLUSH_BACKOFF_EXPONENT + ) + delay = self._config.flush_interval * 2**exponent + capped = min(delay, max(MAX_FLUSH_BACKOFF_SECONDS, self._config.flush_interval)) + return max(capped * self._jitter, self._retry_after.remaining()) + + def _depth_trigger_due_locked(self) -> bool: + # Not while backing off: the queue stays at depth through an outage, and + # every span end would re-send. + return ( + len(self._queue) >= self._max_export_batch_size + and not self._consecutive_failures + and not self._retry_after.is_open() + ) + + def _arm_timer_if_idle_locked(self) -> None: + if self._flush_timer is None and self._queue: + self._arm_timer_locked(self._next_flush_delay_locked()) + + def _rearm_after_pass_locked(self) -> None: + if not self._queue or self._closed: + return + # The depth trigger, which a span end may have fired mid-pass: the + # replacement below would otherwise push it out a whole interval. + if self._depth_trigger_due_locked(): + self._arm_timer_locked(0.0, replace=True) + return + delay = self._next_flush_delay_locked() + if ( + self._flush_timer is not None + and time.monotonic() + delay <= self._flush_timer_fires_at + ): + return + self._arm_timer_locked(delay, replace=True) + + def _arm_timer_locked(self, delay: float, replace: bool = False) -> None: + if self._flush_timer is not None: + if not replace: + return + self._flush_timer.cancel() + self._flush_timer = None + timer = threading.Timer(delay, lambda: self._timer_flush(timer)) + timer.daemon = True + # Registered only once started, so a failed start leaves nothing + # behind and the next span end arms again. + timer.start() + self._flush_timer = timer + self._flush_timer_fires_at = time.monotonic() + delay + + def _timer_flush(self, fired: threading.Timer) -> None: + with self._lock: + if fired is not self._flush_timer: + return + try: + self.flush(_timer=fired) + except Exception: + log.debug("Background span flush failed", exc_info=True) + + def _clear_timer_locked(self) -> None: + if self._flush_timer is not None: + self._flush_timer.cancel() + self._flush_timer = None + + +def _draw_jitter() -> float: + return 1 - FLUSH_BACKOFF_JITTER + random.random() * FLUSH_BACKOFF_JITTER * 2 From d000111689045c70ffd9f40fb6e1e2d0fd8f6be4 Mon Sep 17 00:00:00 2001 From: Anna Garcia Date: Mon, 14 Sep 2026 18:55:57 -0400 Subject: [PATCH 2/4] fix(traces): keep the scheduled flush when a replacement timer fails to start, and count an unencodable span once The replacement timer now starts before the old one is cancelled, so a thread that cannot be created leaves the earlier flush in place. Records that fail to encode leave the queue before the send is settled, so a failed send no longer counts them a second time. --- posthog/test/tracing/test_export.py | 26 ++++++++++++++++-- posthog/tracing/_export.py | 42 +++++++++++++++++------------ 2 files changed, 49 insertions(+), 19 deletions(-) diff --git a/posthog/test/tracing/test_export.py b/posthog/test/tracing/test_export.py index 555b1d13..aa2c81fb 100644 --- a/posthog/test/tracing/test_export.py +++ b/posthog/test/tracing/test_export.py @@ -180,10 +180,12 @@ def start(self): pipeline, sender, _ = make_traces(max_export_batch_size=3) for _ in range(3): pipeline.start_span("a").end() - assert pipeline._exporter._flush_timer is None + interval_timer = pipeline._exporter._flush_timer + assert interval_timer.delay == 5 and not interval_timer.cancelled pipeline.start_span("b").end() timer = pipeline._exporter._flush_timer - assert timer is not None and timer.started and timer.delay == 0 + assert timer is not interval_timer and timer.started and timer.delay == 0 + assert interval_timer.cancelled timer.fire() assert queued(pipeline) == [] assert [len(b) for b in sender.batches()] == [3, 1] @@ -358,6 +360,26 @@ def test_never_returns_a_span_it_failed_to_encode(self): assert sender.payloads == [] assert queued(pipeline) == [] + def test_counts_a_span_it_failed_to_encode_once(self, caplog): + caplog.set_level("WARNING", logger="posthog") + sender = FakeSender(SendOutcome("fatal")) + pipeline, _, _ = make_traces(sender=sender, max_export_batch_size=2) + pipeline.start_span("bad").end() + pipeline.start_span("fine").end() + encode = export_module.build_otlp_span + + def encode_unless_bad(record): + if record.name == "bad": + raise RuntimeError("bad") + return encode(record) + + with mock.patch.object(export_module, "build_otlp_span", encode_unless_bad): + pipeline.flush() + assert [[s["name"] for s in b] for b in sender.batches()] == [["fine"]] + assert queued(pipeline) == [] + (record,) = [r for r in caplog.records if "Dropping" in r.getMessage()] + assert "Dropping 2 span(s)" in record.getMessage() + class TestDroppedBatchesEndTheFailureSequence: @pytest.mark.parametrize( diff --git a/posthog/tracing/_export.py b/posthog/tracing/_export.py index 272604d0..4d1404e5 100644 --- a/posthog/tracing/_export.py +++ b/posthog/tracing/_export.py @@ -255,16 +255,22 @@ def _drain(self, deadline: Optional[float]) -> Tuple[int, bool]: and not self._retry_after.is_open() ) - spans = self._encode(batch) - if not spans: + spans, failed = self._encode(batch) + if failed: + # Out of the queue before the send is settled, so a failed + # send does not count them a second time. with self._lock: if self._is_closed(): return removed, True - del self._queue[:size] - self._reset_head_batch_budget_locked() - remaining -= size - removed += size - continue + for index in reversed(failed): + del self._queue[index] + if not spans: + self._reset_head_batch_budget_locked() + size -= len(failed) + remaining -= len(failed) + removed += len(failed) + if not spans: + continue # The first request is exempt, so a flush called with no budget left # (a serverless handler, say) still ships a batch. @@ -361,15 +367,18 @@ def _is_closed(self) -> bool: # A method, so the check after each send is not narrowed away. return self._closed - def _encode(self, batch: List[SpanRecord]) -> List[dict]: + def _encode(self, batch: List[SpanRecord]) -> Tuple[List[dict], List[int]]: + """The batch as OTLP spans, and the indexes of records that could not be encoded.""" encoded: List[dict] = [] - for record in batch: + failed: List[int] = [] + for index, record in enumerate(batch): try: encoded.append(build_otlp_span(record)) except Exception: log.debug("Failed to encode a span; dropping it", exc_info=True) self._drops.record(1, "its attributes could not be encoded") - return encoded + failed.append(index) + return encoded, failed def _discard_if_disabled_locked(self) -> bool: if not getattr(self._client, "disabled", False): @@ -431,16 +440,15 @@ def _rearm_after_pass_locked(self) -> None: self._arm_timer_locked(delay, replace=True) def _arm_timer_locked(self, delay: float, replace: bool = False) -> None: - if self._flush_timer is not None: - if not replace: - return - self._flush_timer.cancel() - self._flush_timer = None + if self._flush_timer is not None and not replace: + return timer = threading.Timer(delay, lambda: self._timer_flush(timer)) timer.daemon = True - # Registered only once started, so a failed start leaves nothing - # behind and the next span end arms again. + # Started before the old timer is cancelled, so a failed start keeps + # whatever flush was already scheduled. timer.start() + if self._flush_timer is not None: + self._flush_timer.cancel() self._flush_timer = timer self._flush_timer_fires_at = time.monotonic() + delay From 9c4e8937c0f8a5c0b4dbee09da2be7196d5638d2 Mon Sep 17 00:00:00 2001 From: Anna Garcia Date: Wed, 16 Sep 2026 21:03:42 -0400 Subject: [PATCH 3/4] fix(traces): keep backing off after a refused batch is dropped Dropping the head batch after eight windows reset the failure count, so the next batch went straight to an endpoint that had just failed eight times and the depth trigger came back on. The backoff ceiling is now the events lane's single constant, and FakeTimer refuses to fire a timer the code cancelled. --- posthog/test/tracing/helpers.py | 1 + posthog/test/tracing/test_export.py | 15 +++++++++++++++ posthog/tracing/_export.py | 11 +++++++---- 3 files changed, 23 insertions(+), 4 deletions(-) diff --git a/posthog/test/tracing/helpers.py b/posthog/test/tracing/helpers.py index edaed8b9..9c0790bc 100644 --- a/posthog/test/tracing/helpers.py +++ b/posthog/test/tracing/helpers.py @@ -41,6 +41,7 @@ def cancel(self): self.cancelled = True def fire(self): + assert not self.cancelled, "fired a timer the code cancelled" self.fn() diff --git a/posthog/test/tracing/test_export.py b/posthog/test/tracing/test_export.py index aa2c81fb..d6866da3 100644 --- a/posthog/test/tracing/test_export.py +++ b/posthog/test/tracing/test_export.py @@ -415,6 +415,21 @@ def test_drops_a_batch_the_endpoint_keeps_refusing_and_moves_to_the_next_one( assert [[s["name"] for s in b] for b in sender.batches()][-1] == ["next"] assert queued(pipeline) == [] + def test_a_dropped_batch_keeps_the_backoff_for_the_next_one(self, clock): + sender = FakeSender(*([SendOutcome("retry-later")] * MAX_RETRIES_PER_BATCH)) + pipeline, _, _ = make_traces(sender=sender, max_export_batch_size=1) + pipeline.start_span("stuck").end() + pipeline.start_span("next").end() + for _ in range(MAX_RETRIES_PER_BATCH): + pipeline.flush() + clock["now"] += 100 + exporter = pipeline._exporter + assert [s.name for s in queued(pipeline)] == ["next"] + assert exporter._consecutive_failures >= MAX_RETRIES_PER_BATCH + assert exporter._head_batch_failures == 1 + pipeline.start_span("c").end() + assert FakeTimer.instances[-1].delay > 0 + def test_charges_the_budget_once_per_backoff_window_not_per_attempt(self, clock): pipeline, _, _ = make_traces(sender=FakeSender(SendOutcome("retry-later"))) pipeline.start_span("a").end() diff --git a/posthog/tracing/_export.py b/posthog/tracing/_export.py index 4d1404e5..24010c3f 100644 --- a/posthog/tracing/_export.py +++ b/posthog/tracing/_export.py @@ -12,6 +12,7 @@ import time from typing import Any, Callable, List, Optional, Tuple +from ..capture_v1 import _MAX_BACKOFF_SECONDS from ._config import ResolvedTracesConfig from ._drops import DropLog from ._otlp import ( @@ -27,11 +28,12 @@ MAX_RETRIES_PER_BATCH = 8 MAX_FLUSH_BACKOFF_EXPONENT = 6 -MAX_FLUSH_BACKOFF_SECONDS = 30.0 +# The events lane's ceiling, so both backoffs and the Retry-After clamp share one. +MAX_FLUSH_BACKOFF_SECONDS = float(_MAX_BACKOFF_SECONDS) # Nothing upstream bounds the header, and an unbounded value would strand the -# queue. The same ceiling as the SDK's own backoff. -MAX_RETRY_AFTER_SECONDS = 30.0 +# queue. +MAX_RETRY_AFTER_SECONDS = MAX_FLUSH_BACKOFF_SECONDS # Spread each backoff by up to a quarter, so clients refused together do not # return together. @@ -348,7 +350,8 @@ def _apply_outcome_locked( if self._head_batch_failures < MAX_RETRIES_PER_BATCH: return 0, True, "Span export failed; retrying on the next flush" del self._queue[:size] - self._end_failure_sequence_locked() + # The endpoint is still failing: the next batch keeps backing off. + self._reset_head_batch_budget_locked() self._drops.record( size, "the ingestion endpoint failed {} times in a row".format( From 71d931ed3a7620e28a21fe292715e78bbb1acc65 Mon Sep 17 00:00:00 2001 From: Anna Garcia Date: Wed, 16 Sep 2026 21:58:40 -0400 Subject: [PATCH 4/4] fix(traces): retry a refused batch within a timed flush, and start the budget once the lock is held A flush with budget left stopped at the first retriable failure, so a 30 s shutdown flush made one attempt and then discarded the backlog. A caller-driven flush now waits out the backoff and retries while budget remains, with one last attempt at the deadline; timer flushes and untimed flushes are unchanged. The budget starts once no other flush is in flight, and a flush that never gets the lock says so at debug. A size-1 413 restores the batch size it halved from, and the ramp doubles instead of adding one. The resource is encoded once per exporter. --- posthog/test/tracing/test_export.py | 126 +++++++++++++++++++++++++++- posthog/tracing/_export.py | 90 ++++++++++++++++---- 2 files changed, 196 insertions(+), 20 deletions(-) diff --git a/posthog/test/tracing/test_export.py b/posthog/test/tracing/test_export.py index d6866da3..aa1cc15c 100644 --- a/posthog/test/tracing/test_export.py +++ b/posthog/test/tracing/test_export.py @@ -1,4 +1,5 @@ import threading +import time from types import SimpleNamespace from unittest import mock @@ -152,8 +153,9 @@ def test_a_spent_budget_still_ships_one_batch(self): assert queued(pipeline) == [] def test_returns_without_draining_when_another_flush_holds_the_lock_past_the_deadline( - self, + self, caplog ): + caplog.set_level("DEBUG", logger="posthog") pipeline, sender, _ = make_traces() pipeline.start_span("a").end() timer = FakeTimer.instances[-1] @@ -165,6 +167,31 @@ def test_returns_without_draining_when_another_flush_holds_the_lock_past_the_dea assert sender.payloads == [] assert pipeline._exporter._flush_timer is timer assert not timer.cancelled + assert "another flush was still in flight" in caplog.text + + def test_the_budget_starts_once_the_lock_is_held(self, clock): + pipeline, sender, _ = make_traces(max_export_batch_size=1) + exporter = pipeline._exporter + + def send_slowly(client, payload): + sender.payloads.append(payload) + clock["now"] += 0.1 + return SendOutcome("ok") + + exporter._send = send_slowly + for _ in range(2): + pipeline.start_span("a").end() + exporter._flush_lock.acquire() + + def release_after_a_while(): + time.sleep(0.05) + clock["now"] += 1.0 + exporter._flush_lock.release() + + threading.Thread(target=release_after_a_while).start() + pipeline.flush(timeout=0.5) + assert len(sender.payloads) == 2 + assert queued(pipeline) == [] def test_re_arms_after_a_timer_that_failed_to_start(self): class FlakyTimer(FakeTimer): @@ -263,9 +290,19 @@ def test_ramps_the_batch_size_back_up_after_a_413_shrink(self): for i in range(8): pipeline.start_span(str(i)).end() pipeline.flush() - assert pipeline._exporter._max_export_batch_size == 6 + assert pipeline._exporter._max_export_batch_size == 8 assert [len(b) for b in sender.batches()] == [8, 4, 4] + def test_restores_the_batch_size_once_a_413_isolates_the_oversized_span(self): + sender = FakeSender(*([SendOutcome("too-large")] * 4), SendOutcome("ok")) + pipeline, _, _ = make_traces(sender=sender, max_export_batch_size=8) + for i in range(8): + pipeline.start_span(str(i)).end() + pipeline.flush() + assert [len(b) for b in sender.batches()] == [8, 4, 2, 1, 7] + assert pipeline._exporter._max_export_batch_size == 8 + assert queued(pipeline) == [] + def test_a_batch_measured_too_large_locally_splits_only_that_drain(self): sender = FakeSender(TOO_LARGE_LOCALLY, SendOutcome("ok")) pipeline, _, _ = make_traces(sender=sender, max_export_batch_size=8) @@ -911,3 +948,88 @@ def test_a_forked_child_drops_the_inherited_queue_and_timer(self): ) pipeline.start_span("child-span").end() assert [r.name for r in queued(pipeline)] == ["child-span"] + + +def waits_advance(clock, pipeline): + """Make the exporter's backoff wait move the fake clock instead of sleeping.""" + waited = [] + + def wait(seconds): + waited.append(seconds) + clock["now"] += seconds + return False + + pipeline._exporter._wait_for_retry = wait + return waited + + +class TestRetryWithinBudget: + def test_retries_a_retriable_failure_after_its_backoff(self, clock): + sender = FakeSender(SendOutcome("retry-later"), SendOutcome("ok")) + pipeline, _, _ = make_traces(sender=sender) + waited = waits_advance(clock, pipeline) + pipeline.start_span("a").end() + pipeline.flush(timeout=30) + assert len(sender.payloads) == 2 + assert waited == [5] + assert queued(pipeline) == [] + + def test_makes_a_last_attempt_at_the_deadline(self, clock): + sender = FakeSender(SendOutcome("retry-later")) + pipeline, _, _ = make_traces(sender=sender) + waited = waits_advance(clock, pipeline) + pipeline.start_span("a").end() + pipeline.flush(timeout=12) + # Attempts at 0, 5 and 12: the second backoff of 10 is cut to the budget. + assert len(sender.payloads) == 3 + assert waited == [5, 7] + assert len(queued(pipeline)) == 1 + + def test_honours_a_retry_after_within_the_budget(self, clock): + sender = FakeSender(SendOutcome("retry-later", 60), SendOutcome("ok")) + pipeline, _, _ = make_traces(sender=sender) + waited = waits_advance(clock, pipeline) + pipeline.start_span("a").end() + pipeline.flush(timeout=40) + assert waited == [MAX_RETRY_AFTER_SECONDS] + assert queued(pipeline) == [] + + def test_a_timer_flush_does_not_retry(self, clock): + sender = FakeSender(SendOutcome("retry-later")) + pipeline, _, _ = make_traces(sender=sender) + waited = waits_advance(clock, pipeline) + pipeline.start_span("a").end() + FakeTimer.instances[-1].fire() + assert len(sender.payloads) == 1 + assert waited == [] + + def test_a_flush_without_a_timeout_does_not_retry(self, clock): + sender = FakeSender(SendOutcome("retry-later")) + pipeline, _, _ = make_traces(sender=sender) + waited = waits_advance(clock, pipeline) + pipeline.start_span("a").end() + pipeline.flush() + assert len(sender.payloads) == 1 + assert waited == [] + + def test_close_cuts_the_wait_short(self, clock): + sender = FakeSender(SendOutcome("retry-later")) + pipeline, _, _ = make_traces(sender=sender) + exporter = pipeline._exporter + pipeline.start_span("a").end() + + def close_during_wait(seconds): + exporter.close() + return True + + exporter._wait_for_retry = close_during_wait + pipeline.flush(timeout=30) + assert len(sender.payloads) == 1 + + def test_the_real_wait_returns_when_close_is_called(self): + pipeline, _, _ = make_traces() + exporter = pipeline._exporter + threading.Thread(target=lambda: (time.sleep(0.05), exporter.close())).start() + started = time.monotonic() + assert exporter._wait_for_retry(5) is True + assert time.monotonic() - started < 2 diff --git a/posthog/tracing/_export.py b/posthog/tracing/_export.py index 24010c3f..18e86dbc 100644 --- a/posthog/tracing/_export.py +++ b/posthog/tracing/_export.py @@ -20,6 +20,7 @@ build_otlp_span, build_resource_attributes, build_traces_payload, + to_resource_key_value_list, ) from ._transport import SendOutcome, send_traces_batch @@ -101,20 +102,28 @@ def __init__( self._config = config self._drops = drops self._send = send - self._resource_attributes = build_resource_attributes( - config.service_name, - config.service_version, - config.environment, - config.resource_attributes, + # Encoded once: the resource is the same for every batch. + self._resource = to_resource_key_value_list( + build_resource_attributes( + config.service_name, + config.service_version, + config.environment, + config.resource_attributes, + ) ) self._lock = threading.Lock() self._flush_lock = threading.Lock() + # Set by close(), so a flush waiting out a backoff returns at once. + self._closing = threading.Event() self._closed = False self._queue: List[SpanRecord] = [] self._flush_timer: Optional[threading.Timer] = None self._flush_timer_fires_at = 0.0 self._max_export_batch_size = config.max_export_batch_size + # The size before a server 413 halved it, restored once the oversized + # span is isolated and dropped. + self._batch_size_before_halving: Optional[int] = None self._consecutive_failures = 0 # Drawn once per failure, so the timer and the retry-budget charge see # the same delay. @@ -151,15 +160,24 @@ def flush( ) -> None: """Drain the queue: one pass over what was queued, then one follow-up pass. - With a ``timeout``, no request starts once it is spent, except the - first. A request already in flight is bounded by the client's timeout. + With a ``timeout``, the budget starts once no other flush is in flight; + one still in flight after that long a wait is left to finish and + nothing is sent here. No request starts once the budget is spent, + except the first, and a retriable failure is retried after its backoff + while budget remains, once more at the deadline. Without a timeout a + retriable failure is left to the timer. A request already in flight is + bounded by the client's timeout. """ - deadline = None if timeout is None else time.monotonic() + timeout # -1 is Lock.acquire's unbounded form. if not self._flush_lock.acquire( timeout=-1 if timeout is None else max(0.0, timeout) ): + log.debug( + "Skipping a span flush: another flush was still in flight after %ss", + timeout, + ) return + deadline = None if timeout is None else time.monotonic() + timeout try: with self._lock: # A timer that waited behind another flush may have been @@ -168,14 +186,7 @@ def flush( return self._clear_timer_locked() try: - removed, stop = self._drain(deadline) - if ( - removed - and not stop - and self._queue - and (deadline is None or time.monotonic() < deadline) - ): - self._drain(deadline) + self._drain_within_budget(deadline, retry=_timer is None) finally: with self._lock: self._rearm_after_pass_locked() @@ -183,10 +194,42 @@ def flush( self._flush_lock.release() self._drops.warn_if_due(force=True) + def _drain_within_budget(self, deadline: Optional[float], retry: bool) -> None: + while True: + removed, stop = self._drain(deadline) + if not stop: + if ( + removed + and self._queue + and (deadline is None or time.monotonic() < deadline) + ): + self._drain(deadline) + return + if deadline is None or not retry: + return + wait = self._retry_wait_locked() + remaining = deadline - time.monotonic() + if wait is None or remaining <= 0: + return + if self._wait_for_retry(min(wait, remaining)): + return + + def _retry_wait_locked(self) -> Optional[float]: + """The backoff to wait out before retrying, or ``None`` when there is nothing to retry.""" + with self._lock: + if self._closed or not self._queue or not self._consecutive_failures: + return None + return self._next_flush_delay_locked() + + def _wait_for_retry(self, seconds: float) -> bool: + """Wait out a backoff; ``True`` when close() cut the wait short.""" + return self._closing.wait(seconds) + def close(self) -> None: """Stop exporting and discard what is still queued. Called at shutdown.""" with self._lock: self._closed = True + self._closing.set() self._clear_timer_locked() discarded = len(self._queue) self._queue = [] @@ -212,10 +255,12 @@ def reinit_after_fork(self) -> None: # in the child, so they are replaced rather than acquired. self._lock = threading.Lock() self._flush_lock = threading.Lock() + self._closing = threading.Event() self._flush_timer = None self._flush_timer_fires_at = 0.0 self._queue = [] self._max_export_batch_size = self._config.max_export_batch_size + self._batch_size_before_halving = None self._retry_after.reset() self._end_failure_sequence_locked() @@ -281,7 +326,7 @@ def _drain(self, deadline: Optional[float]) -> Tuple[int, bool]: sent_any = True try: outcome = self._send( - self._client, build_traces_payload(spans, self._resource_attributes) + self._client, build_traces_payload(spans, self._resource) ) except Exception: log.debug("Span batch send failed", exc_info=True) @@ -312,8 +357,11 @@ def _apply_outcome_locked( if outcome.kind == "ok": del self._queue[:size] self._end_failure_sequence_locked() + self._batch_size_before_halving = None if self._max_export_batch_size < self._config.max_export_batch_size: - self._max_export_batch_size += 1 + self._max_export_batch_size = min( + self._config.max_export_batch_size, self._max_export_batch_size * 2 + ) return size, False, None if outcome.kind == "too-large": @@ -321,11 +369,17 @@ def _apply_outcome_locked( del self._queue[:1] self._end_failure_sequence_locked() self._drops.record(1, "it is too large for the ingestion endpoint") + # The oversized span is gone; the batches after it are not suspect. + if self._batch_size_before_halving is not None: + self._max_export_batch_size = self._batch_size_before_halving + self._batch_size_before_halving = None return 1, False, None # Halve the refused batch, not the configured size: a shallow queue # would otherwise resend the same body. halved = max(1, size // 2) if not outcome.measured_locally: + if self._batch_size_before_halving is None: + self._batch_size_before_halving = self._max_export_batch_size self._max_export_batch_size = halved self._reset_head_batch_budget_locked() return (