Add sender queue that expires oldest first - #986
Conversation
There was a problem hiding this comment.
🟡 Changes recommended
Unresolved critical and moderate findings remain.
Once you've addressed the issues Copilot identified, you can request another Copilot review.
Pull request overview
This pull request adds a bounded sender queue that evicts oldest entries, expires stale metrics, and retries transient failures.
Changes:
- Introduces
SenderQueueandPendingPayload. - Integrates expiry, eviction telemetry, and retry handling.
- Adds unit tests and a performance benchmark.
File summaries
| File | Review summary |
|---|---|
tests/unit/dogstatsd/test_statsd.py |
One nit regarding a timing-dependent queue test. |
tests/performance/test_sender_queue_benchmark.py |
Reviewed with no final comments. |
datadog/dogstatsd/sender_queue.py |
One critical producer-notification issue and one moderate shutdown requeue issue. |
datadog/dogstatsd/base.py |
One critical shutdown issue and three moderate expiry, batching, and retry issues. |
Review details
Suppressed comments (2)
datadog/dogstatsd/base.py:1792
- Queue-mode retry only handles
sent is None, but_xmit_packet_attempt()returnsFalseforsocket.timeout;_get_uds_socket()can raise that exception when a UDS connect attempt times out. A connection timeout therefore falls through to the writer-drop path and permanently loses the queued payload instead of letting the sender queue retry it. Distinguish connect timeouts from send timeouts or propagate connect timeouts as transient failures in queue mode.
if sent is None and queue_mode:
# Connection trouble, and the caller is the background sender
# queue: let it requeue the payload and retry once reconnected,
# instead of dropping it here.
return None
datadog/dogstatsd/sender_queue.py:196
Stopis allowed past the capacity limit input(), butrequeue_front()counts that sentinel inlen(self._deque). During shutdown, if the sender has an in-flight payload andstop()appendsStop, a failed send sees the queue as full and drops the payload instead of requeueing it; this loses metrics that shutdown is supposed to drain. Coordinate sentinel insertion with in-flight work or make the requeue path ignore the shutdown sentinel when enforcing capacity.
if self._maxsize > 0 and len(self._deque) >= self._maxsize:
self._on_drop_queue_full(item)
self._finish_task_locked()
return
- Files reviewed: 4/4 changed files
- Comments generated: 5
- Review effort level: Lite
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| if sent is None: | ||
| # Connection trouble: keep the payload for the next attempt | ||
| # instead of losing it. The queue's own expiry check (on a | ||
| # future get()) is what eventually gives up on a payload | ||
| # that's been stuck for too long, unless it's replay-safe. |
There was a problem hiding this comment.
This will be addressed in a followup PR that will place a timeout on these functions.
This comment has been minimized.
This comment has been minimized.
Codex Review SummaryThis comment shows the latest Codex review activity on this pull request.
ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings. |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: dce1a3b4e7
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| # instead of losing it. The queue's own expiry check (on a | ||
| # future get()) is what eventually gives up on a payload | ||
| # that's been stuck for too long, unless it's replay-safe. | ||
| pending_queue.requeue_front(item) # type: ignore[arg-type] |
There was a problem hiding this comment.
When a replay-safe payload encounters a transient UDS failure with a positive socket_connect_timeout, it is requeued at the front forever because it never expires. If stop(), disable_background_sender(), or pre_fork() appends Stop during the outage, this payload remains ahead of the sentinel on every retry, so _sender_thread.join() never returns and the application cannot shut down or fork. Shutdown needs to stop retrying or otherwise prioritize the sentinel.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
uhmmm, isn't this important enough to be fixed here? I mean, the fact that an application hangs when being shutdown is not great :D
There was a problem hiding this comment.
uhmmm, isn't this important enough to be fixed here? I mean, the fact that an application hangs when being shutdown is not great :D
🤔 I can merge the two PRs if you feel it would be better? Just wanted to keep concerns separate..
There was a problem hiding this comment.
Up to you! Approved! Definitely you have more context here :)
| pending_queue.requeue_front(item) # type: ignore[arg-type] | ||
| time.sleep(backoff) | ||
| backoff = min(backoff * 2, UDS_CONNECT_RETRY_MAX_BACKOFF) |
There was a problem hiding this comment.
When the bounded queue is refilled while a send is in flight, or when that payload expires during the send attempt, requeue_front() drops it and completes its task rather than requeuing it. The caller cannot distinguish that outcome and still sleeps and increases backoff, delaying already-queued work by up to a second and making additional payloads more likely to expire under load; have requeue_front() report whether it retained the item and only back off on success.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
I disagree with this. When the connection is down this indicates a downstream issue and we want to backoff the connection attempts regardless of the state of the items on the queue.
_send_to_buffer/_should_flush indexed {False:..., True:...} dicts by
replay_safe, costing a bool() coercion plus a subscript on the single
hottest path in the library (once per metric, ~18k/s per benchmark
client). Measured in the benchmark image: _send_to_buffer 0.6388us on
master -> 0.7344us on this branch (+15.0%), which was ~72% of the whole
per-metric regression (+0.132us, +7.2%).
Hold the two batches as four plain attributes and branch on replay_safe
instead, and inline the size comparison so the hot path makes no call to
_should_flush (which master also paid). _should_flush is retained for the
pre-existing surface and for tests.
Also stop reading the clock in SenderQueue.get() for entries that can
never expire: the argument to _expired() is evaluated before the call, so
every bare-string (replay-safe) get() computed a monotonic() it then
discarded.
Measured after, best-of-7: _send_to_buffer -16.5%, full per-metric -8.5%.
Unit tests: 184 passed, 1 skipped with DD_ORIGIN_DETECTION_ENABLED=false
(the 45 failures otherwise seen are a pre-existing container-id artifact
of the dev host, identical before and after this change).
These four are development aids, not part of the change under review: tests/manual/emulate_reconnect_then_write_fails.py tests/manual/test_sender_queue_manual.py tests/manual/test_shutdown_bound.py tests/performance/test_sender_queue_benchmark.py They are ad-hoc drivers and a benchmark harness rather than tests the suite runs, and together they accounted for 1278 of the ~1970 added lines under tests/, which buried the unit coverage that does matter. Retained locally via .git/info/exclude (which is not committed) so they stay available for development without shipping in the review.
tests/manual/ holds ad-hoc development drivers, not tests the suite runs, so none of it belongs in the review. This removes the one remaining tracked file, test_gauge_with_timestamp_aggregation.py, leaving tests/manual/ entirely untracked. Retained on disk via .git/info/exclude, which now excludes the whole directory so future scratch scripts there cannot be added by accident.
StephenWakely
added
the
changelog/Added
| "Gave up reconnecting after socket_connect_timeout (%ss), dropping the packet", | ||
| self.socket_connect_timeout, | ||
| ) | ||
| sent = None |
There was a problem hiding this comment.
Don't we actually requeue packet here instead of dropping?
There was a problem hiding this comment.
This is all removed in the followup PR where we no longer give up.
| item = self._deque.popleft() | ||
| # A slot just opened up: wake one thread blocked in put()'s | ||
| # wait-for-room loop, if any (harmless no-op otherwise). | ||
| self._not_full.notify() |
There was a problem hiding this comment.
Can here be a potential race condition with requeue_front?
There was a problem hiding this comment.
Maybe. There is an assumption that get and requeue_front have to be on the same thread. With a single sender thread that we have here, they are. There's nothing stopping someone coming along and breaking this in future. I'll look into that.
There was a problem hiding this comment.
This checks to ensure we don't double count.. 18d3018
gh-worker-dd-devflow-36fce6
Bot
added
mergequeue-status: queued
mergequeue-status: in_progress
and removed
mergequeue-status: queued
labels
Requirements for Contributing to this repository
What does this PR do?
Changes the sender queue from a simple
queue.Queueto a queue with the following properties:Description of the Change
With the recent addition of a socket connect timeout, metrics can be retried whilst facing network outages or agent downtimes. As it is retrying the sender queue can fill up, and eventually will overflow. When overflowing we want to make sure the most relevant metrics are kept.
Metrics sent without a timestamp have a limited time during which they are relevant. If we queue metrics for 2 minutes, and then send them all in one go, this would lead to 2 minutes worth of metrics being aggregated in a single time window, causing a huge spike. Metric expiry ensures this cannot happen.
Metrics sent with timestamps cannot expire as they are sent through to the Datadog backend as is.
Alternate Designs
Possible Drawbacks
There are performance implications with this change. Benchmarking:
3 scenarios:
During the restart the client queue grows to ~217k. (Each item in the queue is a payload which can consist of more than one metric, nontimestamped queue is less because items start expiring earlier).
Memory usage is roughly the same at the peak - no timestamps is slightly less because the payloads don't include the timestamp string
Settling down to within ~30mb.
I don't get accurate cpu measurements during the agent downtime since the agent is down, but during typical usage CPU seems largely unaffected.
Dropped metrics
It is worth highlighting that this PR will cause more non-timestamped metrics to be dropped over the previous release since we now expire metrics that are stuck in a queue for more than 10 seconds if they are not sent with a timestamp.
Verification Process
Additional Notes
There is a followup PR:
wait_for_pendingandstopfunctions.Release Notes
Review checklist (to be filled by reviewers)
changelog/label attached. If applicable it should have thebackward-incompatiblelabel attached.do-not-merge/label attached.kind/andseverity/labels attached at least.