From dd6ebee2aa27b2ecb7274b3b5f6a7db348f1abd8 Mon Sep 17 00:00:00 2001 From: Stephen Wakely Date: Wed, 9 Sep 2026 11:31:01 +0100 Subject: [PATCH 1/5] Add sender queue that expires oldest first --- datadog/dogstatsd/base.py | 182 +++++--- datadog/dogstatsd/sender_queue.py | 247 +++++++++++ .../test_sender_queue_benchmark.py | 406 ++++++++++++++++++ tests/unit/dogstatsd/test_statsd.py | 258 ++++++++++- 4 files changed, 1032 insertions(+), 61 deletions(-) create mode 100644 datadog/dogstatsd/sender_queue.py create mode 100644 tests/performance/test_sender_queue_benchmark.py diff --git a/datadog/dogstatsd/base.py b/datadog/dogstatsd/base.py index 5922e1183..fc969d953 100644 --- a/datadog/dogstatsd/base.py +++ b/datadog/dogstatsd/base.py @@ -23,12 +23,6 @@ if sys.version_info[:2] >= (3, 5): from typing import TYPE_CHECKING # noqa: F401 -try: - import queue -except ImportError: - # pypy has the same module, but capitalized. - import Queue as queue # type: ignore[no-redef] - # pylint: disable=unused-import if sys.version_info[:2] >= (3, 5): @@ -49,6 +43,13 @@ ) from datadog.dogstatsd.route import get_default_route from datadog.dogstatsd.container import Cgroup +from datadog.dogstatsd.sender_queue import ( + SenderQueue, + PendingPayload, + Stop, + PENDING_PAYLOAD_EXPIRY_SECONDS, + coalesce_enqueue_time, +) from datadog.util.compat import text, urlparse from datadog.util.format import normalize_tags, validate_cardinality from datadog.version import __version__ @@ -224,8 +225,6 @@ def reverse(self): ] ) + "\n" -Stop = object() - SUPPORTS_FORKING = hasattr(os, "register_at_fork") and not os.environ.get("DD_DOGSTATSD_DISABLE_FORK_SUPPORT", None) TRACK_INSTANCES = not os.environ.get("DD_DOGSTATSD_DISABLE_INSTANCE_TRACKING", None) @@ -490,16 +489,19 @@ def __init__( Default: True. :type disable_background_sender: boolean - :param sender_queue_size: Set the maximum number of packets to queue for the sender. Optional - How may packets to queue before blocking or dropping the packet if the packet queue is already full. + :param sender_queue_size: Set the maximum number of packets to queue for the sender. Optional. + Once the queue is full, adding a new packet drops the oldest queued packet (and any additional + expired packets at the front of the queue) to make room, instead of blocking or dropping the new + packet. Packets aren't held indefinitely either: a queued packet that hasn't been sent within + PENDING_PAYLOAD_EXPIRY_SECONDS is dropped when it's pulled off the queue, unless it carries its own + explicit timestamp (e.g. gauge_with_timestamp, or count/service_check/event with an explicit + timestamp), in which case it's kept until it can actually be sent. Default: 0 (unlimited). :type sender_queue_size: integer - :param sender_queue_timeout: Set timeout for packet queue operations, in seconds. Optional. - How long the application thread is willing to wait for the queue clear up before dropping the metric packet. - If set to None, wait forever. - If set to zero drop the packet immediately if the queue is full. - Default: 0 (no wait) + :param sender_queue_timeout: Deprecated and ignored. The sender queue no longer blocks: it always + makes room for a new packet by dropping older or expired entries instead. Kept only for backwards + compatibility with existing call sites. :type sender_queue_timeout: float :param track_instance: Keep track of this instance and automatically handle cleanup when os.fork() is called, @@ -634,7 +636,7 @@ def __init__( else: log.debug("Statsd buffering and aggregation is disabled") - self._queue = None # type: Optional[queue.Queue[Union[str, object]]] + self._queue = None # type: Optional[SenderQueue] self._sender_thread = None # type: Optional[threading.Thread] self._sender_enabled = False @@ -716,25 +718,19 @@ def enable_background_sender(self, sender_queue_size=0, sender_queue_timeout=0): to os.fork(). :param sender_queue_size: Set the maximum number of packets to queue for the sender. - How many packets to queue before blocking or dropping the packet if the packet queue is already full. + Once the queue is full, adding a new packet drops the oldest queued packet (and any additional + expired packets at the front of the queue) to make room, instead of blocking or dropping the new + packet. Default: 0 (unlimited). :type sender_queue_size: integer, optional - :param sender_queue_timeout: Set timeout for packet queue operations, in seconds. - How long the application thread is willing to wait for the queue clear up before dropping the metric packet. - If set to None, wait forever. If set to zero drop the packet immediately if the queue is full. - Default: 0 (no wait). + :param sender_queue_timeout: Deprecated and ignored: the sender queue no longer blocks. Kept only + for backwards compatibility with existing call sites. :type sender_queue_timeout: float, optional """ with self._config_lock: self._sender_enabled = True self._sender_queue_size = sender_queue_size - if sender_queue_timeout is None: - self._queue_blocking = True - self._queue_timeout = None - else: - self._queue_blocking = sender_queue_timeout > 0 - self._queue_timeout = max(0, sender_queue_timeout) self._start_sender_thread() @@ -1169,6 +1165,10 @@ def _reset_buffer(self): with self._buffer_lock: self._current_buffer_total_size = 0 self._buffer = [] + # A freshly (re)started buffer starts out replay-safe; it's + # downgraded to False as soon as anything not-replay-safe is + # appended to it. See _send_to_buffer(). + self._buffer_replay_safe = True def flush(self): # type: () -> None @@ -1182,7 +1182,7 @@ def flush_buffered_metrics(self): with self._buffer_lock: # Only send packets if there are packets to send if self._buffer: - self._send_to_server("\n".join(self._buffer)) + self._send_to_server("\n".join(self._buffer), self._buffer_replay_safe) self._reset_buffer() def flush_aggregated_metrics(self): @@ -1569,8 +1569,13 @@ def _report(self, metric, metric_type, value, tags, sample_rate, timestamp=0, sa metric, metric_type, value, tags, sample_rate, timestamp, cardinality ) + # A metric carrying its own explicit timestamp is replay-safe: sending + # it late (e.g. after sitting in the background sender queue) doesn't + # change what it means. + replay_safe = timestamp > 0 + # Send it - self._send(payload) + self._send(payload, replay_safe) def _reset_telemetry(self): # type: () -> None @@ -1580,21 +1585,35 @@ def _reset_telemetry(self): self.bytes_sent = 0 self.bytes_dropped_queue = 0 self.bytes_dropped_writer = 0 + self.bytes_dropped_expired = 0 self.packets_sent = 0 self.packets_dropped_queue = 0 self.packets_dropped_writer = 0 + self.packets_dropped_expired = 0 self._last_flush_time = time.time() # Aliases for backwards compatibility. @property def packets_dropped(self): # type: () -> int - return self.packets_dropped_queue + self.packets_dropped_writer + return self.packets_dropped_queue + self.packets_dropped_writer + self.packets_dropped_expired @property def bytes_dropped(self): # type: () -> int - return self.bytes_dropped_queue + self.bytes_dropped_writer + return self.bytes_dropped_queue + self.bytes_dropped_writer + self.bytes_dropped_expired + + def _account_dropped_queue_full(self, item): + # type: (PendingPayload) -> None + """A payload was evicted from the sender queue to make room for a new one.""" + self.packets_dropped_queue += 1 + self.bytes_dropped_queue += len(item.payload.encode(self.encoding)) + + def _account_dropped_expired(self, item): + # type: (PendingPayload) -> None + """A payload sat in the sender queue longer than PENDING_PAYLOAD_EXPIRY_SECONDS.""" + self.packets_dropped_expired += 1 + self.bytes_dropped_expired += len(item.payload.encode(self.encoding)) def _flush_telemetry(self): # type: () -> str @@ -1633,26 +1652,34 @@ def _is_telemetry_flush_time(self): return self._telemetry and \ self._last_flush_time + self._telemetry_flush_interval < time.time() - def _send_to_server(self, packet): - # type: (str) -> None + def _send_to_server(self, packet, replay_safe=False): + # type: (str, bool) -> None # Skip the lock if the queue is None. There is no race with enable_background_sender. if self._queue is not None: # Prevent a race with disable_background_sender. with self._buffer_lock: packet_with_newline = packet + '\n' if self._queue is not None: - try: - self._queue.put(packet_with_newline, self._queue_blocking, self._queue_timeout) - except queue.Full: - self.packets_dropped_queue += 1 - self.bytes_dropped_queue += len(packet_with_newline.encode(self.encoding)) + # replay_safe payloads never have their enqueued_at read + # (see SenderQueue._expired()'s short-circuit), so skip + # both the clock read and the float allocation for them. + enqueued_at = None if replay_safe else coalesce_enqueue_time() + self._queue.put(PendingPayload(packet_with_newline, enqueued_at, replay_safe)) return self._xmit_packet_with_telemetry(packet + '\n') - def _xmit_packet_with_telemetry(self, packet): - # type: (str) -> None - self._xmit_packet(packet, False) + def _xmit_packet_with_telemetry(self, packet, queue_mode=False): + # type: (str, bool) -> Optional[bool] + """Send one packet, optionally piggy-backing a telemetry flush. + + :param queue_mode: True when called from the background sender + thread on behalf of a queued PendingPayload. In that mode, a + connection failure is reported back as None (rather than being + accounted for and dropped) so the caller can requeue the payload + and retry once reconnected, instead of losing it. + """ + sent = self._xmit_packet(packet, False, queue_mode=queue_mode) if self._is_telemetry_flush_time(): telemetry = self._flush_telemetry() @@ -1666,6 +1693,8 @@ def _xmit_packet_with_telemetry(self, packet): self.bytes_dropped_writer += len(telemetry) self.packets_dropped_writer += 1 + return sent + def _installed_socket(self, is_telemetry): # type: (bool) -> Optional[_Socket] """ @@ -1680,8 +1709,16 @@ def _installed_socket(self, is_telemetry): return self.telemetry_socket return self.socket - def _xmit_packet(self, packet, is_telemetry): - # type: (str, bool) -> bool + def _xmit_packet(self, packet, is_telemetry, queue_mode=False): + # type: (str, bool, bool) -> Optional[bool] + """Attempt to send packet, retrying a reconnect within this call as budget allows. + + Returns True if sent. Otherwise returns False for a definitive, + non-retryable failure (already accounted for as a dropped packet), + or -- only when queue_mode is True -- None for a connection failure + that the sender queue should retry by requeuing the payload rather + than have accounted for here as a drop. + """ if is_telemetry and self._dedicated_telemetry_destination(): uses_uds = self.telemetry_socket_path is not None @@ -1693,6 +1730,7 @@ def _xmit_packet(self, packet, is_telemetry): retry_deadline = time.time() + self.socket_connect_timeout backoff = UDS_CONNECT_RETRY_INITIAL_BACKOFF + sent = None # type: Optional[bool] while True: # Cheap fast-path check before even trying to acquire _socket_lock. if ( @@ -1704,6 +1742,7 @@ def _xmit_packet(self, packet, is_telemetry): "Gave up reconnecting after socket_connect_timeout (%ss), dropping the packet", self.socket_connect_timeout, ) + sent = None break sent = self._xmit_packet_attempt( @@ -1730,6 +1769,12 @@ def _xmit_packet(self, packet, is_telemetry): time.sleep(min(backoff, remaining)) backoff = min(backoff * 2, UDS_CONNECT_RETRY_MAX_BACKOFF) + 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 + if not is_telemetry and self._telemetry: self.bytes_dropped_writer += len(packet) self.packets_dropped_writer += 1 @@ -1836,8 +1881,8 @@ def _xmit_packet_attempt(self, packet, is_telemetry, retry_eligible, retry_deadl return False - def _send_to_buffer(self, packet): - # type: (str) -> None + def _send_to_buffer(self, packet, replay_safe=False): + # type: (str, bool) -> None with self._buffer_lock: if self._should_flush(len(packet)): self.flush_buffered_metrics() @@ -1846,6 +1891,10 @@ def _send_to_buffer(self, packet): # Update the current buffer length, including line break to anticipate # the final packet size self._current_buffer_total_size += len(packet) + 1 + # The flushed batch is only as replay-safe as its least safe + # member: if anything in it needs to be treated as time-sensitive, + # treat the whole batch that way. + self._buffer_replay_safe = self._buffer_replay_safe and replay_safe def _should_flush(self, length_to_be_added): # type: (int) -> bool @@ -1938,7 +1987,9 @@ def event( if self._telemetry: self.events_count += 1 - self._send(string) + # An event carrying its own explicit date_happened is replay-safe: + # sending it late doesn't change what it means. + self._send(string, replay_safe=bool(date_happened)) def service_check( self, @@ -1984,7 +2035,9 @@ def service_check( if self._telemetry: self.service_checks_count += 1 - self._send(string) + # A service check carrying its own explicit timestamp is replay-safe: + # sending it late doesn't change what it means. + self._send(string, replay_safe=bool(timestamp)) @staticmethod def _normalize_and_join_tags(tags): @@ -2062,7 +2115,12 @@ def _start_sender_thread(self): if self._queue is not None: return - self._queue = queue.Queue(self._sender_queue_size) + self._queue = SenderQueue( + self._sender_queue_size, + PENDING_PAYLOAD_EXPIRY_SECONDS, + self._account_dropped_queue_full, + self._account_dropped_expired, + ) log.debug("Starting background sender thread") self._sender_thread = threading.Thread( @@ -2086,19 +2144,35 @@ def _stop_sender_thread(self): self._sender_thread.join() self._sender_thread = None - def _sender_main_loop(self, queue): - # type: (queue.Queue[Union[str, object]]) -> None + def _sender_main_loop(self, pending_queue): + # type: (SenderQueue) -> None + backoff = UDS_CONNECT_RETRY_INITIAL_BACKOFF while True: - item = queue.get() + item = pending_queue.get() if item is Stop: - queue.task_done() + pending_queue.task_done() return # next line has type ignore because the type checker cannot # know that 'if item is Stop' is the only case where item is # of object type. - self._xmit_packet_with_telemetry(item) # type: ignore[arg-type] # noqa: F821 - queue.task_done() + sent = self._xmit_packet_with_telemetry(item.payload, queue_mode=True) # type: ignore[attr-defined] # noqa: F821 + + 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. + pending_queue.requeue_front(item) # type: ignore[arg-type] + time.sleep(backoff) + backoff = min(backoff * 2, UDS_CONNECT_RETRY_MAX_BACKOFF) + continue + + # Sent, or a definitive failure that _xmit_packet already + # accounted for as a dropped packet -- either way, this + # payload's story is over. + pending_queue.task_done() + backoff = UDS_CONNECT_RETRY_INITIAL_BACKOFF def wait_for_pending(self): # type: () -> None diff --git a/datadog/dogstatsd/sender_queue.py b/datadog/dogstatsd/sender_queue.py new file mode 100644 index 000000000..8cf202ade --- /dev/null +++ b/datadog/dogstatsd/sender_queue.py @@ -0,0 +1,247 @@ +import collections +import sys +import threading + +try: + # Python 3.3+ + from time import monotonic +except ImportError: + # Python 2: no monotonic clock available, fall back to wall clock. + from time import time as monotonic + +if sys.version_info[:2] >= (3, 5): + from typing import Callable, Optional, Union # noqa: F401 + + +# Sentinel telling the background sender thread to shut down. +Stop = object() + +# How long (in seconds) a non-replay-safe payload may sit in the background +# sender queue before it's considered stale and dropped instead of sent. +# Payloads that carry their own explicit timestamp (replay_safe) are exempt: +# delivering those late doesn't change what they mean, so they're kept +# around until they can actually be sent. +PENDING_PAYLOAD_EXPIRY_SECONDS = 10.0 + +# Granularity for coalesce_enqueue_time() below. Deliberately far below +# PENDING_PAYLOAD_EXPIRY_SECONDS (by two orders of magnitude with the default +# above), so it has no meaningful effect on expiry accuracy, but lets many +# payloads enqueued within the same short window share one float object +# instead of each allocating their own -- which is exactly when it matters: +# under sustained load or a backlog, not when the queue is lightly used. +_TIMESTAMP_COALESCE_SECONDS = 0.1 + +# Bucket + cached value for coalesce_enqueue_time(). Plain module globals, +# not a lock: under a race between threads, the worst outcome is a +# redundant allocation (two threads each compute a fresh reading for the +# same bucket), never an incorrect timestamp. +_coalesce_bucket = None # type: Optional[int] +_coalesce_value = 0.0 # type: float + + +def coalesce_enqueue_time(): + # type: () -> float + """A monotonic() reading coalesced to _TIMESTAMP_COALESCE_SECONDS granularity. + + Only meant for stamping payloads that DO need expiry tracking (see + PendingPayload.enqueued_at). The slop this introduces (at most one + bucket width, 0.1s by default) is negligible next to the multi-second + expiry window it feeds into. + """ + global _coalesce_bucket, _coalesce_value + raw = monotonic() + bucket = int(raw / _TIMESTAMP_COALESCE_SECONDS) + if bucket != _coalesce_bucket: + _coalesce_bucket = bucket + _coalesce_value = raw + return _coalesce_value + + +class PendingPayload(object): + """A single packet queued for the background sender. + + :ivar payload: The already-serialized packet text (including its + trailing newline), ready to be written to the socket. + :ivar enqueued_at: A monotonic timestamp recorded when the payload + became eligible for sending (i.e. when it was put on the queue). + Used to decide whether it has been sitting in the queue for too + long to still be worth sending. None when replay_safe is True: it's + never read in that case (see SenderQueue._expired()'s short-circuit), + so skipping the allocation costs nothing. + :ivar replay_safe: True when delayed delivery preserves the payload's + meaning because it carries its own explicit timestamp. Such + payloads are never dropped for being stale, and never need + enqueued_at. + """ + + __slots__ = ("payload", "enqueued_at", "replay_safe") + + def __init__(self, payload, enqueued_at, replay_safe): + # type: (str, Optional[float], bool) -> None + self.payload = payload + self.enqueued_at = enqueued_at + self.replay_safe = replay_safe + + +class SenderQueue(object): + """Bounded hand-off queue between application threads and the background sender thread. + + Unlike queue.Queue, put() never blocks and never rejects a payload. When + the queue is already at its maximum size, the oldest entry is dropped to + make room, along with any additional expired entries left at the front, + so a backlog of stale payloads can't shut out fresh metrics indefinitely. + + get() drops expired entries lazily too, from the front, before returning + the next payload actually worth handing to the sender. + + A payload that fails to send (e.g. because the connection is down) can be + handed back with requeue_front() so it's retried first. That still + respects both the expiry check and the size limit though: the queue + must never grow past maxsize, and a payload that's gone stale while it + was being (re)tried is dropped rather than requeued. + """ + + def __init__(self, maxsize, expiry_seconds, on_drop_queue_full, on_drop_expired): + # type: (int, float, Callable[[PendingPayload], None], Callable[[PendingPayload], None]) -> None + self._maxsize = maxsize + self._expiry_seconds = expiry_seconds + self._on_drop_queue_full = on_drop_queue_full + self._on_drop_expired = on_drop_expired + self._deque = collections.deque() # type: collections.deque + self._lock = threading.Lock() + self._not_empty = threading.Condition(self._lock) + self._all_tasks_done = threading.Condition(self._lock) + + # Keep track of the tasks that are being processed. A task pulled from the queue may + # be returned if the connection fails, so we don't consider the queue empty until + # all tasks have been dropped or sent. + self._unfinished_tasks = 0 + + def _expired(self, item, now): + # type: (PendingPayload, float) -> bool + if item.replay_safe: + return False + # enqueued_at is only ever None for replay_safe items (see + # PendingPayload), which are already excluded above -- it's a plain + # float here. mypy can't correlate that invariant across the two + # attributes, hence the ignore. + return (now - item.enqueued_at) > self._expiry_seconds # type: ignore[operator] + + def _make_room_locked(self): + # type: () -> None + """Drop the oldest entry, plus any further expired entries at the front. + + Called with self._lock already held, and only when the queue is at + capacity. Never touches the Stop sentinel: by the time it's queued, + nothing else is ever put on the queue again, so it can only ever be + the newest entry, never the one being evicted here. + """ + if not self._deque or self._deque[0] is Stop: + return + + now = monotonic() + oldest = self._deque.popleft() + # The oldest entry is always dropped to make room. If it happens to + # also be expired, attribute it to staleness rather than to the + # queue being full, since that's the more useful signal. + if self._expired(oldest, now): + self._on_drop_expired(oldest) + else: + self._on_drop_queue_full(oldest) + self._finish_task_locked() + + # Keep clearing out additional stale entries left at the front: they + # would otherwise just sit there consuming a slot until they're + # eventually popped. + while self._deque and self._deque[0] is not Stop and self._expired(self._deque[0], now): + self._on_drop_expired(self._deque.popleft()) + self._finish_task_locked() + + def put(self, item): + # type: (Union[PendingPayload, object]) -> None + """Queue a payload (or the Stop sentinel), evicting old entries if needed.""" + with self._not_empty: + if item is not Stop and self._maxsize > 0 and len(self._deque) >= self._maxsize: + self._make_room_locked() + + self._deque.append(item) + self._unfinished_tasks += 1 + self._not_empty.notify() + + def requeue_front(self, item): + # type: (PendingPayload) -> None + """Put an in-flight payload back at the front after a failed send attempt. + + The payload was already accounted for by the put() that originally + queued it (its task isn't done yet), so a successful requeue here + doesn't touch _unfinished_tasks. But it's still subject to the same + rules as any other entry: an item that's expired while it was being + (re)tried is dropped instead of requeued, and the queue is never + allowed to grow past maxsize -- if it's already full, the requeue is + dropped too rather than evicting something else to make room for it. + Either way, a drop here finishes the task that put() started. + """ + with self._not_empty: + if self._expired(item, monotonic()): + self._on_drop_expired(item) + self._finish_task_locked() + return + + if self._maxsize > 0 and len(self._deque) >= self._maxsize: + self._on_drop_queue_full(item) + self._finish_task_locked() + return + + self._deque.appendleft(item) + self._not_empty.notify() + + def get(self): + # type: () -> Union[PendingPayload, object] + """Block for the next payload, silently dropping expired entries along the way.""" + while True: + with self._not_empty: + while not self._deque: + self._not_empty.wait() + item = self._deque.popleft() + + if item is Stop: + return item + + if self._expired(item, monotonic()): + self._on_drop_expired(item) + self.task_done() + continue + + return item + + def _finish_task_locked(self): + # type: () -> None + # Caller already holds self._lock (shared by _not_empty / _all_tasks_done). + unfinished = self._unfinished_tasks - 1 + if unfinished < 0: + raise ValueError("task_done() called too many times") + self._unfinished_tasks = unfinished + if unfinished == 0: + self._all_tasks_done.notify_all() + + def task_done(self): + # type: () -> None + with self._all_tasks_done: + self._finish_task_locked() + + def join(self): + # type: () -> None + with self._all_tasks_done: + while self._unfinished_tasks: + self._all_tasks_done.wait() + + def qsize(self): + # type: () -> int + with self._lock: + return len(self._deque) + + def empty(self): + # type: () -> bool + with self._lock: + return not self._deque + diff --git a/tests/performance/test_sender_queue_benchmark.py b/tests/performance/test_sender_queue_benchmark.py new file mode 100644 index 000000000..b755c569f --- /dev/null +++ b/tests/performance/test_sender_queue_benchmark.py @@ -0,0 +1,406 @@ +""" +Microbenchmark: SenderQueue vs. stdlib queue.Queue. + +This isolates just the hand-off queue's own overhead -- no sockets, no +network variance -- because that's the piece that changed when the +background sender moved off queue.Queue. put() runs synchronously on every +metric emission's calling thread (the application's hot path), so its +*latency* matters at least as much as raw throughput; get() runs on the +background sender thread. + +queue.Queue is used as the baseline throughout via a thin adapter +(_OldStyleQueueAdapter) that reproduces the OLD behavior being replaced: +put_nowait() and drop-with-a-counter on queue.Full, get()+task_done() to +drain. That's the fairest apples-to-apples comparison, since it's literally +what SenderQueue's put()/get() replaced. + +Scenarios: + 1. Unbounded put() then get(), single-threaded (best case for both -- + no eviction, no contention). + 2. Bounded queue, kept permanently full: every put() forces an eviction + for SenderQueue, vs an immediate reject-with-exception for + queue.Queue. This is the main new cost the redesign introduces. + 3. A single put() that has to walk past a large backlog of already-EXPIRED + entries at the front (SenderQueue's opportunistic-cleanup loop has no + upper bound tied to "just free one slot" -- it clears every consecutive + stale entry it finds). Reports cost as a function of backlog size, to + surface whether this can spike a calling thread's latency. + 4. Producer/consumer concurrency: N producer threads hammering put() while + 1 consumer thread drains, measuring achieved producer throughput and + put() latency percentiles under real lock contention. + 5. Per-item memory footprint: PendingPayload wrapper vs a bare str. + +Usage: + python3 tests/performance/test_sender_queue_benchmark.py [--quick] + + --quick shrinks every N so it finishes in a few seconds (CI-friendly); + default sizes are big enough to get low-noise numbers on a quiet + machine. + +This prints numbers and interpretation guidance; it does not hard-fail on +absolute thresholds (those are too hardware/noise dependent to gate CI on +reliably). The one thing it does assert on is the *shape* of the eviction +cost in scenario 3 -- that it's linear in backlog size, not something worse. +Read the printed numbers yourself before/after a change and compare. +""" +import os +import sys +import threading +import time + +try: + import queue as stdlib_queue +except ImportError: + import Queue as stdlib_queue # type: ignore[no-redef] + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..")) + +from datadog.dogstatsd.sender_queue import ( # noqa: E402 + PendingPayload, + SenderQueue, + coalesce_enqueue_time, + monotonic, +) + +QUICK = "--quick" in sys.argv + + +def section(title): + print() + print("=" * 78) + print(title) + print("=" * 78) + + +def note(msg): + print(" . {}".format(msg)) + + +PACKET = "some.metric.name:1|c|#tag1:val1,tag2:val2\n" + + +# -------------------------------------------------------------------------- +# Baseline adapter: reproduces the OLD (pre-SenderQueue) put/get contract on +# top of stdlib queue.Queue, so scenario code can treat both implementations +# uniformly. +# -------------------------------------------------------------------------- +class _OldStyleQueueAdapter(object): + def __init__(self, maxsize): + self._q = stdlib_queue.Queue(maxsize) + self.dropped = 0 + + def put(self, item): + try: + self._q.put_nowait(item) + except stdlib_queue.Full: + self.dropped += 1 + + def get(self): + return self._q.get() + + def task_done(self): + self._q.task_done() + + def qsize(self): + return self._q.qsize() + + +def make_sender_queue(maxsize, expiry_seconds=3600.0): + drops = {"full": 0, "expired": 0} + + def on_full(_item): + drops["full"] += 1 + + def on_expired(_item): + drops["expired"] += 1 + + q = SenderQueue(maxsize, expiry_seconds, on_full, on_expired) + q.drops = drops + return q + + +def percentiles(samples_us): + samples_us = sorted(samples_us) + n = len(samples_us) + + def pct(p): + idx = min(n - 1, int(n * p)) + return samples_us[idx] + + return { + "p50": pct(0.50), + "p90": pct(0.90), + "p99": pct(0.99), + "max": samples_us[-1], + } + + +def time_puts(put_fn, n): + """Time n individual put() calls, returning (total_seconds, [latency_us, ...]).""" + samples = [0.0] * n + t_start = time.perf_counter() + for i in range(n): + t0 = time.perf_counter() + put_fn() + samples[i] = (time.perf_counter() - t0) * 1e6 + total = time.perf_counter() - t_start + return total, samples + + +def report(label, n, total_seconds, samples_us): + p = percentiles(samples_us) + print( + " {:<28s} ops/sec={:>10,.0f} p50={:>7.3f}us p90={:>7.3f}us p99={:>7.3f}us max={:>9.3f}us".format( + label, n / total_seconds, p["p50"], p["p90"], p["p99"], p["max"] + ) + ) + return p + + +# -------------------------------------------------------------------------- +# Scenario 1: unbounded, no contention, no eviction. +# -------------------------------------------------------------------------- +def scenario_1_unbounded_single_threaded(): + section("1. Unbounded put()/get(), single-threaded (best case, no eviction)") + n = 20000 if not QUICK else 2000 + + old = _OldStyleQueueAdapter(maxsize=0) + total, samples = time_puts(lambda: old.put(PACKET), n) + report("queue.Queue (baseline)", n, total, samples) + for _ in range(n): + old.get() + old.task_done() + + new = make_sender_queue(maxsize=0) + total, samples = time_puts(lambda: new.put(PendingPayload(PACKET, coalesce_enqueue_time(), False)), n) + new_p = report("SenderQueue", n, total, samples) + for _ in range(n): + new.get() + new.task_done() + + note("SenderQueue p99 put() latency: {:.3f}us for {:,} plain puts with headroom to spare".format(new_p["p99"], n)) + + +# -------------------------------------------------------------------------- +# Scenario 2: bounded queue, kept permanently full -- every put() evicts. +# -------------------------------------------------------------------------- +def scenario_2_sustained_overflow(): + section("2. Bounded queue kept permanently full: every put() forces eviction (new cost)") + n = 20000 if not QUICK else 2000 + maxsize = 8 + + old = _OldStyleQueueAdapter(maxsize=maxsize) + for _ in range(maxsize): + old.put(PACKET) + total, samples = time_puts(lambda: old.put(PACKET), n) + old_p = report("queue.Queue (baseline)", n, total, samples) + note("queue.Queue just rejects with an exception when full -- O(1), no eviction work at all") + + new = make_sender_queue(maxsize=maxsize) + for _ in range(maxsize): + new.put(PendingPayload(PACKET, coalesce_enqueue_time(), False)) + total, samples = time_puts(lambda: new.put(PendingPayload(PACKET, coalesce_enqueue_time(), False)), n) + new_p = report("SenderQueue", n, total, samples) + + ratio = new_p["p99"] / old_p["p99"] if old_p["p99"] else float("inf") + note("SenderQueue's drop-oldest-and-evict costs {:.1f}x queue.Queue's reject-with-exception at p99".format(ratio)) + note("(each put() here evicts exactly one item -- the mandatory oldest -- since nothing is expired)") + + +# -------------------------------------------------------------------------- +# Scenario 3: one put() that has to walk past a large expired backlog. +# -------------------------------------------------------------------------- +def scenario_3_large_expired_backlog(): + section("3. Cost of ONE put() as a function of an already-expired backlog size") + note("SenderQueue's opportunistic cleanup has no cap tied to 'free just one slot': it clears") + note("every consecutive stale entry at the front. This measures whether that can spike latency.") + + backlog_sizes = [1, 10, 100, 1000, 5000] if not QUICK else [1, 10, 100] + results = [] + for backlog in backlog_sizes: + # expiry_seconds=0 with a backdated enqueued_at makes every backlog + # entry expired the instant it's queued. maxsize=backlog (exactly + # full) so the next put() below is what actually triggers eviction. + q = make_sender_queue(maxsize=backlog, expiry_seconds=0.0) + stale_at = monotonic() - 1000.0 + for _ in range(backlog): + q.put(PendingPayload(PACKET, stale_at, False)) + + t0 = time.perf_counter() + q.put(PendingPayload(PACKET, coalesce_enqueue_time(), False)) + elapsed_us = (time.perf_counter() - t0) * 1e6 + + note("backlog={:>5d} stale entries -> single put() took {:>9.3f}us, evicted {:d}".format( + backlog, elapsed_us, q.drops["expired"] + q.drops["full"] + )) + results.append((backlog, elapsed_us)) + + # Sanity check on the *shape*: cost should scale roughly linearly with + # backlog size, not blow up super-linearly. Compare the per-entry cost + # at the smallest and largest backlog sizes; allow a generous margin for + # fixed overhead and noise, but a large deviation would indicate a real + # algorithmic problem worth investigating. + (small_n, small_us), (large_n, large_us) = results[1], results[-1] + small_per_entry = small_us / small_n + large_per_entry = large_us / large_n + ratio = large_per_entry / small_per_entry if small_per_entry else float("inf") + note( + "per-entry eviction cost: {:.3f}us/entry at backlog={} vs {:.3f}us/entry at backlog={} (ratio={:.2f}x)".format( + small_per_entry, small_n, large_per_entry, large_n, ratio + ) + ) + if ratio > 5.0: + print(" [WARN] per-entry eviction cost grew by {:.1f}x from a small to a large backlog".format(ratio)) + print(" -- that's worse than linear; investigate before shipping.") + else: + print(" [OK] per-entry eviction cost stayed roughly flat as backlog size grew (linear, as expected)") + note("takeaway: a single put() CAN take noticeably longer if a huge stale backlog piles up (e.g. a") + note("long outage with a very large sender_queue_size). Keep sender_queue_size sized to what you're") + note("actually willing to let one put() walk through in the worst case.") + + +# -------------------------------------------------------------------------- +# Scenario 4: producer/consumer concurrency. +# -------------------------------------------------------------------------- +def _run_concurrent(put_fn, get_and_ack_fn, n_producers, n_per_producer, duration_cap=15.0): + latencies = [] + latencies_lock = threading.Lock() + stop = threading.Event() + + def producer(): + local_latencies = [] + for _ in range(n_per_producer): + t0 = time.perf_counter() + put_fn() + local_latencies.append((time.perf_counter() - t0) * 1e6) + with latencies_lock: + latencies.extend(local_latencies) + + def consumer(): + while not stop.is_set(): + get_and_ack_fn() + + consumer_thread = threading.Thread(target=consumer) + consumer_thread.daemon = True + consumer_thread.start() + + producers = [threading.Thread(target=producer) for _ in range(n_producers)] + t0 = time.perf_counter() + for p in producers: + p.start() + for p in producers: + p.join(timeout=duration_cap) + elapsed = time.perf_counter() - t0 + stop.set() + + total_ops = n_producers * n_per_producer + return elapsed, total_ops, latencies + + +def scenario_4_concurrency(): + section("4. Producer/consumer concurrency: N producers hammering put(), 1 consumer draining") + n_producers = 4 + n_per_producer = 5000 if not QUICK else 500 + + old = _OldStyleQueueAdapter(maxsize=1000) + elapsed, total_ops, samples = _run_concurrent( + lambda: old.put(PACKET), + lambda: (old.get(), old.task_done()), + n_producers, + n_per_producer, + ) + old_p = report("queue.Queue (baseline)", total_ops, elapsed, samples) + note("queue.Queue: {} producers x {} puts in {:.3f}s, {} dropped-on-full".format( + n_producers, n_per_producer, elapsed, old.dropped + )) + + new = make_sender_queue(maxsize=1000) + elapsed, total_ops, samples = _run_concurrent( + lambda: new.put(PendingPayload(PACKET, coalesce_enqueue_time(), False)), + lambda: (new.get(), new.task_done()), + n_producers, + n_per_producer, + ) + new_p = report("SenderQueue", total_ops, elapsed, samples) + note("SenderQueue: {} producers x {} puts in {:.3f}s, {} dropped-full, {} dropped-expired".format( + n_producers, n_per_producer, elapsed, new.drops["full"], new.drops["expired"] + )) + + ratio_p99 = new_p["p99"] / old_p["p99"] if old_p["p99"] else float("inf") + note("under real thread contention, SenderQueue's p99 put() latency is {:.2f}x queue.Queue's".format(ratio_p99)) + + +# -------------------------------------------------------------------------- +# Scenario 5: per-item memory footprint. +# -------------------------------------------------------------------------- +def scenario_5_memory_footprint(): + section("5. Per-item memory footprint: PendingPayload wrapper vs a bare str") + note("sys.getsizeof() is shallow: PendingPayload holds a *reference* to the payload string,") + note("not a copy, so its own size doesn't include the string's bytes. The old queue.Queue held") + note("that same string directly with nothing wrapping it, so the real extra cost per item is") + note("the wrapper object itself, plus (for non-replay-safe items) a float object for enqueued_at.") + + payload_str = PACKET + wrapper_size = sys.getsizeof(PendingPayload(payload_str, monotonic(), False)) + + note("payload str (shared either way): {} bytes".format(sys.getsizeof(payload_str))) + note("PendingPayload wrapper itself (__slots__, no __dict__): {} bytes".format(wrapper_size)) + print() + + note("replay_safe=True payloads (gauge_with_timestamp, etc.) never have their enqueued_at read") + note("(SenderQueue._expired() short-circuits on replay_safe first), so base.py passes None") + note("instead of a fresh timestamp -- no float allocation at all for this class of payload:") + replay_safe_wrapped = PendingPayload(payload_str, None, True) + note(" PendingPayload(..., enqueued_at=None, replay_safe=True): {} bytes total, +0 for the timestamp".format( + sys.getsizeof(replay_safe_wrapped) + )) + print() + + note("Non-replay-safe payloads DO need a real enqueued_at, but base.py uses coalesce_enqueue_time()") + note("instead of a bare monotonic() call: many payloads enqueued within the same ~0.1s window share") + note("ONE float object instead of each allocating their own. Demonstrating with {:,} back-to-back".format(2000)) + note("puts (a burst, which is exactly when memory pressure from a growing queue matters most):") + n = 2000 + timestamps = [coalesce_enqueue_time() for _ in range(n)] + distinct = len(set(id(t) for t in timestamps)) + note(" {:,} enqueues -> {} distinct float objects allocated ({:.2f}% of naive per-item allocation)".format( + n, distinct, 100.0 * distinct / n + )) + + worst_case_extra = wrapper_size + sys.getsizeof(monotonic()) + print() + note("Worst case (every timestamp lands in a different coalesce bucket, i.e. low, spread-out") + note("traffic): extra overhead per item vs the old bare-string queue is still just ~{} bytes".format(worst_case_extra)) + for n in (100, 10000, 100000): + note(" at sender_queue_size={:<7d} that's ~{:.1f}KB worst-case additional resident overhead".format( + n, worst_case_extra * n / 1024.0 + )) + + +# -------------------------------------------------------------------------- +def main(): + print("SenderQueue performance microbenchmark") + print("Python {}.{}.{} {}".format(sys.version_info[0], sys.version_info[1], sys.version_info[2], sys.platform)) + if QUICK: + print("(--quick mode: reduced iteration counts)") + + scenario_1_unbounded_single_threaded() + scenario_2_sustained_overflow() + scenario_3_large_expired_backlog() + scenario_4_concurrency() + scenario_5_memory_footprint() + + section("DONE") + print(" This script can't literally run against the pre-SenderQueue commit (SenderQueue") + print(" didn't exist), so the queue.Queue lines above ARE the 'before' baseline: they") + print(" faithfully reproduce the old put_nowait()/get()/task_done() contract SenderQueue") + print(" replaced. Compare SenderQueue's numbers against the queue.Queue numbers *in the") + print(" same run* (same machine, same moment, same load) rather than against an absolute") + print(" number, and re-run a few times to see how much that ratio itself varies with noise.") + print(" For an end-to-end (with real socket I/O) before/after comparison instead, run") + print(" tests/performance/test_statsd_throughput.py with disable_background_sender=False") + print(" against both the current commit and the one before this queue was introduced.") + + +if __name__ == "__main__": + main() diff --git a/tests/unit/dogstatsd/test_statsd.py b/tests/unit/dogstatsd/test_statsd.py index 3045f3286..f6756d2b5 100644 --- a/tests/unit/dogstatsd/test_statsd.py +++ b/tests/unit/dogstatsd/test_statsd.py @@ -30,7 +30,8 @@ # Datadog libraries from datadog import initialize, statsd from datadog import __version__ as version -from datadog.dogstatsd.base import DEFAULT_BUFFERING_FLUSH_INTERVAL, DEFAULT_HOST, DEFAULT_PORT, DogStatsd, MIN_SEND_BUFFER_SIZE, UDP_OPTIMAL_PAYLOAD_LENGTH, UDS_CONNECT_RETRY_INITIAL_BACKOFF, UDS_OPTIMAL_PAYLOAD_LENGTH +from datadog.dogstatsd.base import DEFAULT_BUFFERING_FLUSH_INTERVAL, DEFAULT_HOST, DEFAULT_PORT, DogStatsd, MIN_SEND_BUFFER_SIZE, PendingPayload, SenderQueue, Stop, UDP_OPTIMAL_PAYLOAD_LENGTH, UDS_CONNECT_RETRY_INITIAL_BACKOFF, UDS_OPTIMAL_PAYLOAD_LENGTH +from datadog.dogstatsd.sender_queue import coalesce_enqueue_time, monotonic as sender_queue_clock from datadog.dogstatsd.context import TimedContextManagerDecorator from datadog.util.compat import is_higher_py35, is_p3k from tests.util.contextmanagers import preserve_environment_variable, EnvVars @@ -2621,26 +2622,269 @@ def test_sender_queue_no_timeout(self): statsd = DogStatsd(disable_background_sender=False, sender_queue_timeout=None) def test_bytes_dropped_queue_counts_actual_bytes(self): - # Use a queue of size 1 and a non-blocking timeout so packets are dropped - # when the queue is full, then verify bytes_dropped_queue reflects the real - # byte length of the dropped packet (including the appended newline). + # Use a queue of size 1 so the second packet forces the first (oldest) + # one out, then verify bytes_dropped_queue reflects the real byte + # length of the dropped packet (including the appended newline), and + # that the newer payload is the one that survives in the queue. statsd = DogStatsd( disable_background_sender=False, sender_queue_size=1, - sender_queue_timeout=0, ) statsd.socket = FakeSocket() # Build a packet whose serialised form we know, then compute its length. metric_name = "test.metric" - # Send two packets: the first fills the queue, the second is dropped. - statsd._send_to_server(metric_name) + # Send two packets: the first is evicted (dropped) to make room for the second. statsd._send_to_server(metric_name) + statsd._send_to_server(metric_name + ".second") expected_bytes = len((metric_name + '\n').encode("utf-8")) self.assertEqual(statsd.bytes_dropped_queue, expected_bytes) self.assertEqual(statsd.packets_dropped_queue, 1) + self.assertEqual(statsd.bytes_dropped_expired, 0) + self.assertEqual(statsd.packets_dropped_expired, 0) + + # The surviving (newest) payload is the one the sender thread will send. + statsd.wait_for_pending() + self.assertEqual(statsd.socket.payloads[0].decode("utf-8"), metric_name + ".second\n") + + statsd.stop() + + def test_sender_queue_drops_oldest_and_stale_entries_on_overflow(self): + dropped_queue_full = [] + dropped_expired = [] + + pending_queue = SenderQueue( + maxsize=2, + expiry_seconds=20.0, + on_drop_queue_full=dropped_queue_full.append, + on_drop_expired=dropped_expired.append, + ) + + now = sender_queue_clock() + fresh = PendingPayload("fresh\n", now, False) + stale = PendingPayload("stale\n", now - 100, False) + newest = PendingPayload("newest\n", now, False) + + # Fill the queue: [fresh, stale] (stale is already expired, but that + # doesn't matter until something tries to make room or pull it off). + pending_queue.put(fresh) + pending_queue.put(stale) + self.assertEqual(pending_queue.qsize(), 2) + + # Queue is full: the oldest entry (fresh) is evicted to make room, and + # since the next entry at the front (stale) is also expired, it gets + # opportunistically cleared out too. + pending_queue.put(newest) + + self.assertEqual([p.payload for p in dropped_queue_full], ["fresh\n"]) + self.assertEqual([p.payload for p in dropped_expired], ["stale\n"]) + self.assertEqual(pending_queue.qsize(), 1) + self.assertEqual(pending_queue.get().payload, "newest\n") + + def test_sender_queue_overflow_attributes_stale_oldest_entry_to_expiry(self): + dropped_queue_full = [] + dropped_expired = [] + + pending_queue = SenderQueue( + maxsize=1, + expiry_seconds=20.0, + on_drop_queue_full=dropped_queue_full.append, + on_drop_expired=dropped_expired.append, + ) + + stale = PendingPayload("stale\n", sender_queue_clock() - 100, False) + pending_queue.put(stale) + + # The oldest (and only) entry being evicted is itself already + # expired: that's a staleness drop, not a queue-full drop. + pending_queue.put(PendingPayload("newest\n", sender_queue_clock(), False)) + + self.assertEqual(dropped_queue_full, []) + self.assertEqual([p.payload for p in dropped_expired], ["stale\n"]) + + def test_sender_queue_get_drops_expired_entries(self): + dropped_expired = [] + + pending_queue = SenderQueue( + maxsize=0, + expiry_seconds=20.0, + on_drop_queue_full=lambda item: self.fail("unexpected queue-full drop"), + on_drop_expired=dropped_expired.append, + ) + + now = sender_queue_clock() + pending_queue.put(PendingPayload("stale-1\n", now - 100, False)) + pending_queue.put(PendingPayload("stale-2\n", now - 100, False)) + pending_queue.put(PendingPayload("fresh\n", now, False)) + + # get() lazily drains every stale entry at the front before handing + # back the next payload actually worth sending. + item = pending_queue.get() + self.assertEqual(item.payload, "fresh\n") + self.assertEqual([p.payload for p in dropped_expired], ["stale-1\n", "stale-2\n"]) + + def test_sender_queue_replay_safe_payload_never_expires(self): + pending_queue = SenderQueue( + maxsize=0, + expiry_seconds=20.0, + on_drop_queue_full=lambda item: self.fail("unexpected queue-full drop"), + on_drop_expired=lambda item: self.fail("replay-safe payload should not expire"), + ) + + # Far older than the expiry window, but replay_safe=True: never dropped for staleness. + old_but_replay_safe = PendingPayload("timestamped\n", sender_queue_clock() - 10000, True) + pending_queue.put(old_but_replay_safe) + + self.assertIs(pending_queue.get(), old_but_replay_safe) + + def test_coalesce_enqueue_time_reuses_the_same_object_within_a_window(self): + # Back-to-back calls (well within the coalescing window) should + # return the exact same float object, not just an equal value -- + # that's the whole point: fewer allocations under a burst. + a = coalesce_enqueue_time() + b = coalesce_enqueue_time() + self.assertIs(a, b) + + def test_coalesce_enqueue_time_advances_across_windows(self): + first = coalesce_enqueue_time() + time.sleep(0.15) # comfortably past the 0.1s coalescing granularity + second = coalesce_enqueue_time() + self.assertGreater(second, first) + + def test_sender_queue_requeue_front_when_room_available(self): + pending_queue = SenderQueue( + maxsize=2, + expiry_seconds=20.0, + on_drop_queue_full=lambda item: self.fail("unexpected queue-full drop"), + on_drop_expired=lambda item: self.fail("unexpected expiry drop"), + ) + + in_flight = PendingPayload("in-flight\n", sender_queue_clock(), False) + pending_queue.put(in_flight) + + # Simulate the sender thread picking it up and failing to send it. + got = pending_queue.get() + self.assertIs(got, in_flight) + pending_queue.requeue_front(got) + + # There was room for it: it's retried first, ahead of anything newer. + pending_queue.put(PendingPayload("new\n", sender_queue_clock(), False)) + self.assertEqual(pending_queue.get().payload, "in-flight\n") + self.assertEqual(pending_queue.get().payload, "new\n") + + def test_sender_queue_requeue_front_drops_when_queue_is_full(self): + dropped_queue_full = [] + + pending_queue = SenderQueue( + maxsize=1, + expiry_seconds=20.0, + on_drop_queue_full=dropped_queue_full.append, + on_drop_expired=lambda item: self.fail("unexpected expiry drop"), + ) + + in_flight = PendingPayload("in-flight\n", sender_queue_clock(), False) + pending_queue.put(in_flight) + + # Simulate the sender thread picking it up, failing to send it, and a + # fresh payload filling the now-empty slot in the meantime. + got = pending_queue.get() + self.assertIs(got, in_flight) + pending_queue.put(PendingPayload("new\n", sender_queue_clock(), False)) + + # The queue is already at maxsize: the requeue is dropped rather than + # growing the queue past its limit or evicting the newer entry. + pending_queue.requeue_front(got) + + self.assertEqual([p.payload for p in dropped_queue_full], ["in-flight\n"]) + self.assertEqual(pending_queue.qsize(), 1) + self.assertEqual(pending_queue.get().payload, "new\n") + + def test_sender_queue_requeue_front_drops_when_expired(self): + dropped_expired = [] + + pending_queue = SenderQueue( + maxsize=0, + expiry_seconds=0.01, + on_drop_queue_full=lambda item: self.fail("unexpected queue-full drop"), + on_drop_expired=dropped_expired.append, + ) + + # Simulate the sender thread picking up a payload and failing to + # send it, with enough time passing in between that it's now stale. + # Unbounded queue (so it's never "full") isolates the expiry check. + in_flight = PendingPayload("stale\n", sender_queue_clock(), False) + pending_queue.put(in_flight) + got = pending_queue.get() + time.sleep(0.02) + + pending_queue.requeue_front(got) + + self.assertEqual([p.payload for p in dropped_expired], ["stale\n"]) + self.assertEqual(pending_queue.qsize(), 0) + + def test_replay_safe_flows_through_to_pending_payload(self): + statsd = DogStatsd(disable_background_sender=False) + statsd.socket = FakeSocket() + + captured = [] + original_put = statsd._queue.put + + def capture_put(item): + if item is not Stop: + captured.append(item) + return original_put(item) + + statsd._queue.put = capture_put + + statsd.increment("no.timestamp") + statsd.gauge_with_timestamp("with.timestamp", 1, int(time.time())) + statsd.wait_for_pending() + + self.assertEqual(len(captured), 2) + self.assertFalse(captured[0].replay_safe) + self.assertIsNotNone(captured[0].enqueued_at, "non-replay-safe payloads need a real timestamp to expire against") + self.assertTrue(captured[1].replay_safe) + self.assertIsNone( + captured[1].enqueued_at, + "replay_safe payloads never have enqueued_at read (see SenderQueue._expired()), " + "so it should be skipped entirely rather than allocated for nothing", + ) + + statsd.stop() + + def test_connection_failure_requeues_and_resends_once_reconnected(self): + # A UDS client whose socket is broken, with a small connect budget so + # the internal reconnect-and-retry loop inside _xmit_packet gives up + # quickly and hands off to the sender queue's own retry-by-requeuing. + working_socket = FakeSocket() + attempts = {"count": 0} + + def flaky_get_uds_socket(cls, socket_path, timeout, connect_timeout): + attempts["count"] += 1 + if attempts["count"] < 4: + raise socket.error(errno.ECONNREFUSED, "still refused") + return working_socket + + with mock.patch.object(DogStatsd, "_get_uds_socket", classmethod(flaky_get_uds_socket)): + statsd = DogStatsd( + socket_path="/tmp/dogstatsd-test-requeue.sock", + disable_telemetry=True, + disable_background_sender=False, + ) + statsd.socket_connect_timeout = 0.05 + + statsd.gauge("eventually.sent", 1) + statsd.wait_for_pending() + + # The payload survived every failed reconnect attempt and was sent + # once a working socket was finally available -- it was never + # dropped as a writer failure or expired out of the queue. + self.assertGreaterEqual(attempts["count"], 4) + self.assertEqual(statsd.packets_dropped_writer, 0) + self.assertEqual(statsd.packets_dropped_expired, 0) + self.assertTrue(working_socket.payloads[0].decode("utf-8").startswith("eventually.sent:1|g")) statsd.stop() From cf0e26ef344ec072f6c10ccbca86aae0f02c3f1d Mon Sep 17 00:00:00 2001 From: Stephen Wakely Date: Wed, 9 Sep 2026 13:19:29 +0100 Subject: [PATCH 2/5] Remove coalesce time optimisation. --- datadog/dogstatsd/base.py | 4 +- datadog/dogstatsd/sender_queue.py | 33 ---------------- .../test_sender_queue_benchmark.py | 38 ++++++++----------- tests/unit/dogstatsd/test_statsd.py | 16 +------- 4 files changed, 19 insertions(+), 72 deletions(-) diff --git a/datadog/dogstatsd/base.py b/datadog/dogstatsd/base.py index fc969d953..69424ede9 100644 --- a/datadog/dogstatsd/base.py +++ b/datadog/dogstatsd/base.py @@ -48,7 +48,7 @@ PendingPayload, Stop, PENDING_PAYLOAD_EXPIRY_SECONDS, - coalesce_enqueue_time, + monotonic, ) from datadog.util.compat import text, urlparse from datadog.util.format import normalize_tags, validate_cardinality @@ -1663,7 +1663,7 @@ def _send_to_server(self, packet, replay_safe=False): # replay_safe payloads never have their enqueued_at read # (see SenderQueue._expired()'s short-circuit), so skip # both the clock read and the float allocation for them. - enqueued_at = None if replay_safe else coalesce_enqueue_time() + enqueued_at = None if replay_safe else monotonic() self._queue.put(PendingPayload(packet_with_newline, enqueued_at, replay_safe)) return diff --git a/datadog/dogstatsd/sender_queue.py b/datadog/dogstatsd/sender_queue.py index 8cf202ade..caf3e0205 100644 --- a/datadog/dogstatsd/sender_queue.py +++ b/datadog/dogstatsd/sender_queue.py @@ -23,39 +23,6 @@ # around until they can actually be sent. PENDING_PAYLOAD_EXPIRY_SECONDS = 10.0 -# Granularity for coalesce_enqueue_time() below. Deliberately far below -# PENDING_PAYLOAD_EXPIRY_SECONDS (by two orders of magnitude with the default -# above), so it has no meaningful effect on expiry accuracy, but lets many -# payloads enqueued within the same short window share one float object -# instead of each allocating their own -- which is exactly when it matters: -# under sustained load or a backlog, not when the queue is lightly used. -_TIMESTAMP_COALESCE_SECONDS = 0.1 - -# Bucket + cached value for coalesce_enqueue_time(). Plain module globals, -# not a lock: under a race between threads, the worst outcome is a -# redundant allocation (two threads each compute a fresh reading for the -# same bucket), never an incorrect timestamp. -_coalesce_bucket = None # type: Optional[int] -_coalesce_value = 0.0 # type: float - - -def coalesce_enqueue_time(): - # type: () -> float - """A monotonic() reading coalesced to _TIMESTAMP_COALESCE_SECONDS granularity. - - Only meant for stamping payloads that DO need expiry tracking (see - PendingPayload.enqueued_at). The slop this introduces (at most one - bucket width, 0.1s by default) is negligible next to the multi-second - expiry window it feeds into. - """ - global _coalesce_bucket, _coalesce_value - raw = monotonic() - bucket = int(raw / _TIMESTAMP_COALESCE_SECONDS) - if bucket != _coalesce_bucket: - _coalesce_bucket = bucket - _coalesce_value = raw - return _coalesce_value - class PendingPayload(object): """A single packet queued for the background sender. diff --git a/tests/performance/test_sender_queue_benchmark.py b/tests/performance/test_sender_queue_benchmark.py index b755c569f..c904cceb4 100644 --- a/tests/performance/test_sender_queue_benchmark.py +++ b/tests/performance/test_sender_queue_benchmark.py @@ -58,7 +58,6 @@ from datadog.dogstatsd.sender_queue import ( # noqa: E402 PendingPayload, SenderQueue, - coalesce_enqueue_time, monotonic, ) @@ -172,7 +171,7 @@ def scenario_1_unbounded_single_threaded(): old.task_done() new = make_sender_queue(maxsize=0) - total, samples = time_puts(lambda: new.put(PendingPayload(PACKET, coalesce_enqueue_time(), False)), n) + total, samples = time_puts(lambda: new.put(PendingPayload(PACKET, monotonic(), False)), n) new_p = report("SenderQueue", n, total, samples) for _ in range(n): new.get() @@ -198,8 +197,8 @@ def scenario_2_sustained_overflow(): new = make_sender_queue(maxsize=maxsize) for _ in range(maxsize): - new.put(PendingPayload(PACKET, coalesce_enqueue_time(), False)) - total, samples = time_puts(lambda: new.put(PendingPayload(PACKET, coalesce_enqueue_time(), False)), n) + new.put(PendingPayload(PACKET, monotonic(), False)) + total, samples = time_puts(lambda: new.put(PendingPayload(PACKET, monotonic(), False)), n) new_p = report("SenderQueue", n, total, samples) ratio = new_p["p99"] / old_p["p99"] if old_p["p99"] else float("inf") @@ -227,7 +226,7 @@ def scenario_3_large_expired_backlog(): q.put(PendingPayload(PACKET, stale_at, False)) t0 = time.perf_counter() - q.put(PendingPayload(PACKET, coalesce_enqueue_time(), False)) + q.put(PendingPayload(PACKET, monotonic(), False)) elapsed_us = (time.perf_counter() - t0) * 1e6 note("backlog={:>5d} stale entries -> single put() took {:>9.3f}us, evicted {:d}".format( @@ -316,7 +315,7 @@ def scenario_4_concurrency(): new = make_sender_queue(maxsize=1000) elapsed, total_ops, samples = _run_concurrent( - lambda: new.put(PendingPayload(PACKET, coalesce_enqueue_time(), False)), + lambda: new.put(PendingPayload(PACKET, monotonic(), False)), lambda: (new.get(), new.task_done()), n_producers, n_per_producer, @@ -356,25 +355,20 @@ def scenario_5_memory_footprint(): )) print() - note("Non-replay-safe payloads DO need a real enqueued_at, but base.py uses coalesce_enqueue_time()") - note("instead of a bare monotonic() call: many payloads enqueued within the same ~0.1s window share") - note("ONE float object instead of each allocating their own. Demonstrating with {:,} back-to-back".format(2000)) - note("puts (a burst, which is exactly when memory pressure from a growing queue matters most):") - n = 2000 - timestamps = [coalesce_enqueue_time() for _ in range(n)] - distinct = len(set(id(t) for t in timestamps)) - note(" {:,} enqueues -> {} distinct float objects allocated ({:.2f}% of naive per-item allocation)".format( - n, distinct, 100.0 * distinct / n + non_replay_safe_extra = wrapper_size + sys.getsizeof(monotonic()) + note("Non-replay-safe payloads DO need a real enqueued_at -- one monotonic() reading per item,") + note("same as any other Python object holding a fresh timestamp. Extra overhead per item vs the") + note("old bare-string queue: ~{} bytes ({} wrapper + {} float).".format( + non_replay_safe_extra, wrapper_size, sys.getsizeof(monotonic()) )) - - worst_case_extra = wrapper_size + sys.getsizeof(monotonic()) - print() - note("Worst case (every timestamp lands in a different coalesce bucket, i.e. low, spread-out") - note("traffic): extra overhead per item vs the old bare-string queue is still just ~{} bytes".format(worst_case_extra)) for n in (100, 10000, 100000): - note(" at sender_queue_size={:<7d} that's ~{:.1f}KB worst-case additional resident overhead".format( - n, worst_case_extra * n / 1024.0 + note(" at sender_queue_size={:<7d} that's ~{:.1f}KB of additional resident overhead".format( + n, non_replay_safe_extra * n / 1024.0 )) + note("(An earlier version of this code coalesced timestamps to a shared per-100ms-bucket float") + note("to cut this under bursty load -- best case ~234KB saved at sender_queue_size=10,000, i.e.") + note("~0.09% of a typical 256MB container's RSS. Reverted: not worth the added global mutable") + note("state, cross-instance coupling, and dedicated concurrency tests for savings that small.)") # -------------------------------------------------------------------------- diff --git a/tests/unit/dogstatsd/test_statsd.py b/tests/unit/dogstatsd/test_statsd.py index f6756d2b5..ed4731daf 100644 --- a/tests/unit/dogstatsd/test_statsd.py +++ b/tests/unit/dogstatsd/test_statsd.py @@ -31,7 +31,7 @@ from datadog import initialize, statsd from datadog import __version__ as version from datadog.dogstatsd.base import DEFAULT_BUFFERING_FLUSH_INTERVAL, DEFAULT_HOST, DEFAULT_PORT, DogStatsd, MIN_SEND_BUFFER_SIZE, PendingPayload, SenderQueue, Stop, UDP_OPTIMAL_PAYLOAD_LENGTH, UDS_CONNECT_RETRY_INITIAL_BACKOFF, UDS_OPTIMAL_PAYLOAD_LENGTH -from datadog.dogstatsd.sender_queue import coalesce_enqueue_time, monotonic as sender_queue_clock +from datadog.dogstatsd.sender_queue import monotonic as sender_queue_clock from datadog.dogstatsd.context import TimedContextManagerDecorator from datadog.util.compat import is_higher_py35, is_p3k from tests.util.contextmanagers import preserve_environment_variable, EnvVars @@ -2739,20 +2739,6 @@ def test_sender_queue_replay_safe_payload_never_expires(self): self.assertIs(pending_queue.get(), old_but_replay_safe) - def test_coalesce_enqueue_time_reuses_the_same_object_within_a_window(self): - # Back-to-back calls (well within the coalescing window) should - # return the exact same float object, not just an equal value -- - # that's the whole point: fewer allocations under a burst. - a = coalesce_enqueue_time() - b = coalesce_enqueue_time() - self.assertIs(a, b) - - def test_coalesce_enqueue_time_advances_across_windows(self): - first = coalesce_enqueue_time() - time.sleep(0.15) # comfortably past the 0.1s coalescing granularity - second = coalesce_enqueue_time() - self.assertGreater(second, first) - def test_sender_queue_requeue_front_when_room_available(self): pending_queue = SenderQueue( maxsize=2, From 08cebfc0ca8c7bf7bd6376c3b86a328a153e3a1a Mon Sep 17 00:00:00 2001 From: Stephen Wakely Date: Thu, 10 Sep 2026 09:44:11 +0100 Subject: [PATCH 3/5] Include expired drops --- datadog/dogstatsd/base.py | 19 +++++++-- tests/unit/dogstatsd/test_statsd.py | 61 ++++++++++++++++++++++++++--- 2 files changed, 71 insertions(+), 9 deletions(-) diff --git a/datadog/dogstatsd/base.py b/datadog/dogstatsd/base.py index 69424ede9..d7dd36260 100644 --- a/datadog/dogstatsd/base.py +++ b/datadog/dogstatsd/base.py @@ -1622,6 +1622,17 @@ def _flush_telemetry(self): tags.extend(self.constant_tags) telemetry_tags = ",".join(tags) + # There's no dedicated wire-protocol metric for expired drops (see + # bytes_dropped_expired/packets_dropped_expired for that level of + # detail in-process): they're folded into the *_dropped_queue lines + # reported to the Agent, since both categories share the same root + # cause from the Agent's point of view -- the payload never reached + # a socket write attempt, dropped by the queue itself rather than by + # the writer. Without this, they'd silently vanish even from the + # combined dropped total sent to the Agent. + bytes_dropped_queue = self.bytes_dropped_queue + self.bytes_dropped_expired + packets_dropped_queue = self.packets_dropped_queue + self.packets_dropped_expired + return TELEMETRY_FORMATTING_STR % ( self.metrics_count, telemetry_tags, @@ -1631,17 +1642,17 @@ def _flush_telemetry(self): telemetry_tags, self.bytes_sent, telemetry_tags, - self.bytes_dropped_queue + self.bytes_dropped_writer, + bytes_dropped_queue + self.bytes_dropped_writer, telemetry_tags, - self.bytes_dropped_queue, + bytes_dropped_queue, telemetry_tags, self.bytes_dropped_writer, telemetry_tags, self.packets_sent, telemetry_tags, - self.packets_dropped_queue + self.packets_dropped_writer, + packets_dropped_queue + self.packets_dropped_writer, telemetry_tags, - self.packets_dropped_queue, + packets_dropped_queue, telemetry_tags, self.packets_dropped_writer, telemetry_tags, diff --git a/tests/unit/dogstatsd/test_statsd.py b/tests/unit/dogstatsd/test_statsd.py index ed4731daf..ada17bd10 100644 --- a/tests/unit/dogstatsd/test_statsd.py +++ b/tests/unit/dogstatsd/test_statsd.py @@ -124,20 +124,26 @@ def __init__(self): super(OverflownSocket, self).__init__(errno.EAGAIN) -def telemetry_metrics(metrics=1, events=0, service_checks=0, bytes_sent=0, bytes_dropped_writer=0, packets_sent=1, packets_dropped_writer=0, transport="udp", tags="", bytes_dropped_queue=0, packets_dropped_queue=0): +def telemetry_metrics(metrics=1, events=0, service_checks=0, bytes_sent=0, bytes_dropped_writer=0, packets_sent=1, packets_dropped_writer=0, transport="udp", tags="", bytes_dropped_queue=0, packets_dropped_queue=0, bytes_dropped_expired=0, packets_dropped_expired=0): tags = "," + tags if tags else "" + # Expired drops have no dedicated wire metric: they're folded into the + # *_dropped_queue lines (and totals) reported to the Agent. See + # DogStatsd._flush_telemetry(). + reported_bytes_dropped_queue = bytes_dropped_queue + bytes_dropped_expired + reported_packets_dropped_queue = packets_dropped_queue + packets_dropped_expired + return "\n".join([ "datadog.dogstatsd.client.metrics:{}|c|#client:py,client_version:{},client_transport:{}{}".format(metrics, version, transport, tags), "datadog.dogstatsd.client.events:{}|c|#client:py,client_version:{},client_transport:{}{}".format(events, version, transport, tags), "datadog.dogstatsd.client.service_checks:{}|c|#client:py,client_version:{},client_transport:{}{}".format(service_checks, version, transport, tags), "datadog.dogstatsd.client.bytes_sent:{}|c|#client:py,client_version:{},client_transport:{}{}".format(bytes_sent, version, transport, tags), - "datadog.dogstatsd.client.bytes_dropped:{}|c|#client:py,client_version:{},client_transport:{}{}".format(bytes_dropped_queue + bytes_dropped_writer, version, transport, tags), - "datadog.dogstatsd.client.bytes_dropped_queue:{}|c|#client:py,client_version:{},client_transport:{}{}".format(bytes_dropped_queue, version, transport, tags), + "datadog.dogstatsd.client.bytes_dropped:{}|c|#client:py,client_version:{},client_transport:{}{}".format(reported_bytes_dropped_queue + bytes_dropped_writer, version, transport, tags), + "datadog.dogstatsd.client.bytes_dropped_queue:{}|c|#client:py,client_version:{},client_transport:{}{}".format(reported_bytes_dropped_queue, version, transport, tags), "datadog.dogstatsd.client.bytes_dropped_writer:{}|c|#client:py,client_version:{},client_transport:{}{}".format(bytes_dropped_writer, version, transport, tags), "datadog.dogstatsd.client.packets_sent:{}|c|#client:py,client_version:{},client_transport:{}{}".format(packets_sent, version, transport, tags), - "datadog.dogstatsd.client.packets_dropped:{}|c|#client:py,client_version:{},client_transport:{}{}".format(packets_dropped_queue + packets_dropped_writer, version, transport, tags), - "datadog.dogstatsd.client.packets_dropped_queue:{}|c|#client:py,client_version:{},client_transport:{}{}".format(packets_dropped_queue, version, transport, tags), + "datadog.dogstatsd.client.packets_dropped:{}|c|#client:py,client_version:{},client_transport:{}{}".format(reported_packets_dropped_queue + packets_dropped_writer, version, transport, tags), + "datadog.dogstatsd.client.packets_dropped_queue:{}|c|#client:py,client_version:{},client_transport:{}{}".format(reported_packets_dropped_queue, version, transport, tags), "datadog.dogstatsd.client.packets_dropped_writer:{}|c|#client:py,client_version:{},client_transport:{}{}".format(packets_dropped_writer, version, transport, tags), ]) + "\n" @@ -1955,6 +1961,51 @@ def test_telemetry(self): self.assertEqual(0, self.statsd.bytes_dropped_queue) self.assertEqual(0, self.statsd.packets_dropped_queue) + def test_telemetry_folds_expired_drops_into_dropped_queue(self): + # There's no dedicated wire metric for expired drops: they're + # reported to the Agent as part of *_dropped_queue (and the combined + # *_dropped total), alongside capacity-based queue drops, since both + # never reach a socket write attempt. The distinction is still + # available in-process via bytes_dropped_expired/packets_dropped_expired. + # Avoid any real container-id auto-detected from the host/sandbox + # cgroup leaking into the expected payload below -- this test is + # about the telemetry counters, not the container-id field. + self.statsd._container_id = None + + self.statsd.bytes_dropped_queue = 8 + self.statsd.packets_dropped_queue = 9 + self.statsd.bytes_dropped_expired = 10 + self.statsd.packets_dropped_expired = 11 + self.statsd.bytes_dropped_writer = 5 + self.statsd.packets_dropped_writer = 7 + + self.statsd.open_buffer() + self.statsd.gauge('page.views', 123) + self.statsd.close_buffer() + + payload = 'page.views:123|g\n' + telemetry = telemetry_metrics( + metrics=1, + bytes_sent=len(payload), + packets_sent=1, + bytes_dropped_queue=8, + packets_dropped_queue=9, + bytes_dropped_expired=10, + packets_dropped_expired=11, + bytes_dropped_writer=5, + packets_dropped_writer=7, + ) + + self.assert_equal_telemetry(payload, self.recv(2), telemetry=telemetry) + + # The in-process counters stay separate even after the flush resets + # them -- confirming the fold happens only in the wire output, not + # by merging the underlying attributes. + self.assertEqual(0, self.statsd.bytes_dropped_queue) + self.assertEqual(0, self.statsd.packets_dropped_queue) + self.assertEqual(0, self.statsd.bytes_dropped_expired) + self.assertEqual(0, self.statsd.packets_dropped_expired) + def test_telemetry_flush_interval(self): dogstatsd = DogStatsd(disable_buffering=False) fake_socket = FakeSocket() From c335ae8167be162953056abba3933dd6914708b1 Mon Sep 17 00:00:00 2001 From: Stephen Wakely Date: Thu, 10 Sep 2026 17:09:02 +0100 Subject: [PATCH 4/5] Implement sender queue timeout --- datadog/dogstatsd/base.py | 37 +++++--- datadog/dogstatsd/sender_queue.py | 45 ++++++++-- tests/unit/dogstatsd/test_statsd.py | 130 ++++++++++++++++++++++++++++ 3 files changed, 189 insertions(+), 23 deletions(-) diff --git a/datadog/dogstatsd/base.py b/datadog/dogstatsd/base.py index d7dd36260..e6d4ab5bf 100644 --- a/datadog/dogstatsd/base.py +++ b/datadog/dogstatsd/base.py @@ -490,18 +490,22 @@ def __init__( :type disable_background_sender: boolean :param sender_queue_size: Set the maximum number of packets to queue for the sender. Optional. - Once the queue is full, adding a new packet drops the oldest queued packet (and any additional - expired packets at the front of the queue) to make room, instead of blocking or dropping the new - packet. Packets aren't held indefinitely either: a queued packet that hasn't been sent within - PENDING_PAYLOAD_EXPIRY_SECONDS is dropped when it's pulled off the queue, unless it carries its own - explicit timestamp (e.g. gauge_with_timestamp, or count/service_check/event with an explicit - timestamp), in which case it's kept until it can actually be sent. + Once the queue is full, adding a new packet waits (see sender_queue_timeout) and then, if still + full, drops the oldest queued packet (and any additional expired packets at the front of the + queue) to make room, instead of dropping the new packet. Packets aren't held indefinitely either: + a queued packet that hasn't been sent within PENDING_PAYLOAD_EXPIRY_SECONDS is dropped when it's + pulled off the queue, unless it carries its own explicit timestamp (e.g. gauge_with_timestamp, or + count/service_check/event with an explicit timestamp), in which case it's kept until it can + actually be sent. Default: 0 (unlimited). :type sender_queue_size: integer - :param sender_queue_timeout: Deprecated and ignored. The sender queue no longer blocks: it always - makes room for a new packet by dropping older or expired entries instead. Kept only for backwards - compatibility with existing call sites. + :param sender_queue_timeout: Set how long, in seconds, adding a packet to a full sender queue + will wait for the background sender to free up a slot before falling back to dropping the oldest + queued packet to make room. If set to zero or None (the default), no waiting happens: a full + queue makes room immediately by dropping the oldest packet. Note this blocks the calling thread + (the one emitting the metric), not just the background sender -- pick a value that fits how long + you're willing to let application code stall during a backlog. :type sender_queue_timeout: float :param track_instance: Keep track of this instance and automatically handle cleanup when os.fork() is called, @@ -718,19 +722,23 @@ def enable_background_sender(self, sender_queue_size=0, sender_queue_timeout=0): to os.fork(). :param sender_queue_size: Set the maximum number of packets to queue for the sender. - Once the queue is full, adding a new packet drops the oldest queued packet (and any additional - expired packets at the front of the queue) to make room, instead of blocking or dropping the new - packet. + Once the queue is full, adding a new packet waits (see sender_queue_timeout) and then, if + still full, drops the oldest queued packet (and any additional expired packets at the front + of the queue) to make room, instead of dropping the new packet. Default: 0 (unlimited). :type sender_queue_size: integer, optional - :param sender_queue_timeout: Deprecated and ignored: the sender queue no longer blocks. Kept only - for backwards compatibility with existing call sites. + :param sender_queue_timeout: Set how long, in seconds, adding a packet to a full sender queue + will wait for the background sender to free up a slot before falling back to dropping the + oldest queued packet to make room. If set to zero or None (the default), no waiting happens. + Note this blocks the calling thread (the one emitting the metric), not just the background + sender. :type sender_queue_timeout: float, optional """ with self._config_lock: self._sender_enabled = True self._sender_queue_size = sender_queue_size + self._sender_queue_timeout = sender_queue_timeout self._start_sender_thread() @@ -2131,6 +2139,7 @@ def _start_sender_thread(self): PENDING_PAYLOAD_EXPIRY_SECONDS, self._account_dropped_queue_full, self._account_dropped_expired, + put_timeout=self._sender_queue_timeout, ) log.debug("Starting background sender thread") diff --git a/datadog/dogstatsd/sender_queue.py b/datadog/dogstatsd/sender_queue.py index caf3e0205..8a5922154 100644 --- a/datadog/dogstatsd/sender_queue.py +++ b/datadog/dogstatsd/sender_queue.py @@ -53,10 +53,14 @@ def __init__(self, payload, enqueued_at, replay_safe): class SenderQueue(object): """Bounded hand-off queue between application threads and the background sender thread. - Unlike queue.Queue, put() never blocks and never rejects a payload. When - the queue is already at its maximum size, the oldest entry is dropped to - make room, along with any additional expired entries left at the front, - so a backlog of stale payloads can't shut out fresh metrics indefinitely. + put() never rejects a payload outright. When the queue is already at its + maximum size and put_timeout is falsy (the default), the oldest entry is dropped + immediately to make room, along with any additional expired entries left + at the front. When put_timeout is a positive number, put() + instead blocks the calling thread for up to that many seconds waiting + for the sender thread to drain a slot; only once that wait times out + (or immediately, if put_timeout is falsy) does it fall back to the same + drop-oldest eviction. get() drops expired entries lazily too, from the front, before returning the next payload actually worth handing to the sender. @@ -65,18 +69,21 @@ class SenderQueue(object): handed back with requeue_front() so it's retried first. That still respects both the expiry check and the size limit though: the queue must never grow past maxsize, and a payload that's gone stale while it - was being (re)tried is dropped rather than requeued. + was being (re)tried is dropped rather than requeued. requeue_front() + never blocks on put_timeout. """ - def __init__(self, maxsize, expiry_seconds, on_drop_queue_full, on_drop_expired): - # type: (int, float, Callable[[PendingPayload], None], Callable[[PendingPayload], None]) -> None + def __init__(self, maxsize, expiry_seconds, on_drop_queue_full, on_drop_expired, put_timeout=None): + # type: (int, float, Callable[[PendingPayload], None], Callable[[PendingPayload], None], Optional[float]) -> None self._maxsize = maxsize self._expiry_seconds = expiry_seconds self._on_drop_queue_full = on_drop_queue_full self._on_drop_expired = on_drop_expired + self._put_timeout = put_timeout self._deque = collections.deque() # type: collections.deque self._lock = threading.Lock() self._not_empty = threading.Condition(self._lock) + self._not_full = threading.Condition(self._lock) self._all_tasks_done = threading.Condition(self._lock) # Keep track of the tasks that are being processed. A task pulled from the queue may @@ -126,10 +133,27 @@ def _make_room_locked(self): def put(self, item): # type: (Union[PendingPayload, object]) -> None - """Queue a payload (or the Stop sentinel), evicting old entries if needed.""" + """Queue a payload (or the Stop sentinel). + + If the queue is full: waits for room for up to put_timeout seconds + (if put_timeout is a positive number), then falls back to evicting + the oldest entry (see _make_room_locked()) if the wait timed out + without room opening up -- or immediately, with no wait at all, if + put_timeout is falsy. Either way, put() never rejects the payload + outright. + """ with self._not_empty: if item is not Stop and self._maxsize > 0 and len(self._deque) >= self._maxsize: - self._make_room_locked() + if self._put_timeout: + deadline = monotonic() + self._put_timeout + while len(self._deque) >= self._maxsize: + remaining = deadline - monotonic() + if remaining <= 0: + break + self._not_full.wait(remaining) + + if len(self._deque) >= self._maxsize: + self._make_room_locked() self._deque.append(item) self._unfinished_tasks += 1 @@ -170,6 +194,9 @@ def get(self): while not self._deque: self._not_empty.wait() 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() if item is Stop: return item diff --git a/tests/unit/dogstatsd/test_statsd.py b/tests/unit/dogstatsd/test_statsd.py index ada17bd10..03fb32546 100644 --- a/tests/unit/dogstatsd/test_statsd.py +++ b/tests/unit/dogstatsd/test_statsd.py @@ -2671,6 +2671,33 @@ def test_sender_calls_task_done(self): def test_sender_queue_no_timeout(self): statsd = DogStatsd(disable_background_sender=False, sender_queue_timeout=None) + statsd.stop() + + def test_sender_queue_timeout_blocks_the_calling_thread_through_the_client(self): + # End-to-end: sender_queue_timeout configured on the real client + # actually makes statsd.increment() (the calling/application thread) + # block waiting for room, not just an internal SenderQueue detail. + statsd = DogStatsd( + disable_background_sender=False, + sender_queue_size=1, + sender_queue_timeout=5.0, + ) + # No socket assigned: the sender thread can never drain anything by + # actually sending, so the only way room opens up is via get() + # pulling an item off (which happens immediately, since nothing can + # succeed in sending it -- it gets hard-dropped as a writer failure + # and the sender loop moves on to the next get()). + statsd.socket = FakeSocket() + + statsd.increment("first") + + t0 = time.time() + statsd.increment("second") + elapsed = time.time() - t0 + + self.assertLess(elapsed, 5.0, "should not have waited out the full 5s timeout") + statsd.wait_for_pending() + statsd.stop() def test_bytes_dropped_queue_counts_actual_bytes(self): # Use a queue of size 1 so the second packet forces the first (oldest) @@ -2702,6 +2729,109 @@ def test_bytes_dropped_queue_counts_actual_bytes(self): statsd.stop() + def test_sender_queue_put_timeout_none_evicts_immediately(self): + # Default behaviour (put_timeout falsy): no waiting at all, same as + # before this feature existed. + dropped_queue_full = [] + pending_queue = SenderQueue( + maxsize=1, + expiry_seconds=100.0, + on_drop_queue_full=dropped_queue_full.append, + on_drop_expired=lambda item: self.fail("unexpected expiry drop"), + ) + + pending_queue.put(PendingPayload("first\n", sender_queue_clock(), False)) + + t0 = time.time() + pending_queue.put(PendingPayload("second\n", sender_queue_clock(), False)) + elapsed = time.time() - t0 + + self.assertLess(elapsed, 0.05, "put() should not have waited at all") + self.assertEqual([p.payload for p in dropped_queue_full], ["first\n"]) + self.assertEqual(pending_queue.get().payload, "second\n") + + def test_sender_queue_put_timeout_wakes_up_when_room_opens(self): + # A slot freed by get() (well within put_timeout) should wake a + # blocked put() immediately rather than making it wait out the full + # timeout, and nothing should be dropped. + dropped_queue_full = [] + pending_queue = SenderQueue( + maxsize=1, + expiry_seconds=100.0, + on_drop_queue_full=dropped_queue_full.append, + on_drop_expired=lambda item: self.fail("unexpected expiry drop"), + put_timeout=5.0, + ) + pending_queue.put(PendingPayload("first\n", sender_queue_clock(), False)) + + result = {} + + def blocked_put(): + t0 = time.time() + pending_queue.put(PendingPayload("second\n", sender_queue_clock(), False)) + result["elapsed"] = time.time() - t0 + + t = threading.Thread(target=blocked_put) + t.start() + time.sleep(0.2) + self.assertTrue(t.is_alive(), "put() should still be waiting for room") + + # Drain the one slot: the blocked put() should wake up promptly. + self.assertEqual(pending_queue.get().payload, "first\n") + pending_queue.task_done() + + t.join(timeout=5.0) + self.assertFalse(t.is_alive()) + self.assertLess(result["elapsed"], 5.0, "should have woken up well before the 5s timeout") + self.assertEqual(dropped_queue_full, [], "nothing should have been dropped: room opened up in time") + self.assertEqual(pending_queue.get().payload, "second\n") + + def test_sender_queue_put_timeout_falls_back_to_eviction(self): + # If room never opens up within put_timeout, put() falls back to + # the same drop-oldest eviction as the immediate (no-wait) case. + dropped_queue_full = [] + pending_queue = SenderQueue( + maxsize=1, + expiry_seconds=100.0, + on_drop_queue_full=dropped_queue_full.append, + on_drop_expired=lambda item: self.fail("unexpected expiry drop"), + put_timeout=0.2, + ) + pending_queue.put(PendingPayload("first\n", sender_queue_clock(), False)) + + t0 = time.time() + pending_queue.put(PendingPayload("second\n", sender_queue_clock(), False)) + elapsed = time.time() - t0 + + self.assertGreaterEqual(elapsed, 0.2) + self.assertEqual([p.payload for p in dropped_queue_full], ["first\n"]) + self.assertEqual(pending_queue.get().payload, "second\n") + + def test_sender_queue_requeue_front_never_blocks_on_put_timeout(self): + # requeue_front() runs on the background sender thread; it must + # never wait on put_timeout, or one stuck retry would stall every + # other queued payload behind it. + dropped_queue_full = [] + pending_queue = SenderQueue( + maxsize=1, + expiry_seconds=100.0, + on_drop_queue_full=dropped_queue_full.append, + on_drop_expired=lambda item: self.fail("unexpected expiry drop"), + put_timeout=5.0, + ) + in_flight = PendingPayload("in-flight\n", sender_queue_clock(), False) + pending_queue.put(in_flight) + got = pending_queue.get() + pending_queue.put(PendingPayload("new\n", sender_queue_clock(), False)) # fills the one slot again + + t0 = time.time() + pending_queue.requeue_front(got) + elapsed = time.time() - t0 + + self.assertLess(elapsed, 0.05, "requeue_front() must not block on put_timeout") + self.assertEqual([p.payload for p in dropped_queue_full], ["in-flight\n"]) + self.assertEqual(pending_queue.get().payload, "new\n") + def test_sender_queue_drops_oldest_and_stale_entries_on_overflow(self): dropped_queue_full = [] dropped_expired = [] From 0515f198e8bcba605061372aadd204731a5dd749 Mon Sep 17 00:00:00 2001 From: Stephen Wakely Date: Fri, 11 Sep 2026 16:04:41 +0100 Subject: [PATCH 5/5] Allow waiting indefinitely --- datadog/dogstatsd/base.py | 29 ++++++------ datadog/dogstatsd/sender_queue.py | 47 ++++++++++++------- tests/unit/dogstatsd/test_statsd.py | 73 +++++++++++++++++++++++++++-- 3 files changed, 113 insertions(+), 36 deletions(-) diff --git a/datadog/dogstatsd/base.py b/datadog/dogstatsd/base.py index e6d4ab5bf..e8cb61aff 100644 --- a/datadog/dogstatsd/base.py +++ b/datadog/dogstatsd/base.py @@ -490,22 +490,18 @@ def __init__( :type disable_background_sender: boolean :param sender_queue_size: Set the maximum number of packets to queue for the sender. Optional. - Once the queue is full, adding a new packet waits (see sender_queue_timeout) and then, if still - full, drops the oldest queued packet (and any additional expired packets at the front of the - queue) to make room, instead of dropping the new packet. Packets aren't held indefinitely either: - a queued packet that hasn't been sent within PENDING_PAYLOAD_EXPIRY_SECONDS is dropped when it's - pulled off the queue, unless it carries its own explicit timestamp (e.g. gauge_with_timestamp, or - count/service_check/event with an explicit timestamp), in which case it's kept until it can - actually be sent. + How many packets to queue before blocking or dropping the packet if the packet queue is already full. Default: 0 (unlimited). :type sender_queue_size: integer :param sender_queue_timeout: Set how long, in seconds, adding a packet to a full sender queue will wait for the background sender to free up a slot before falling back to dropping the oldest - queued packet to make room. If set to zero or None (the default), no waiting happens: a full - queue makes room immediately by dropping the oldest packet. Note this blocks the calling thread - (the one emitting the metric), not just the background sender -- pick a value that fits how long - you're willing to let application code stall during a backlog. + queued packet to make room. Default: 0, meaning no waiting happens at all: a full queue makes + room immediately by dropping the oldest packet. If set to None, waits forever for room instead + of ever falling back to dropping the oldest packet -- an explicit opt-in to unbounded + backpressure; nothing bounds how long this can block if the sender can never catch up. Note this + blocks the calling thread (the one emitting the metric), not just the background sender -- pick a + value that fits how long you're willing to let application code stall during a backlog. :type sender_queue_timeout: float :param track_instance: Keep track of this instance and automatically handle cleanup when os.fork() is called, @@ -729,9 +725,10 @@ def enable_background_sender(self, sender_queue_size=0, sender_queue_timeout=0): :type sender_queue_size: integer, optional :param sender_queue_timeout: Set how long, in seconds, adding a packet to a full sender queue will wait for the background sender to free up a slot before falling back to dropping the - oldest queued packet to make room. If set to zero or None (the default), no waiting happens. - Note this blocks the calling thread (the one emitting the metric), not just the background - sender. + oldest queued packet to make room. Default: 0, meaning no waiting happens at all. If set to + None, waits forever for room instead of ever falling back to dropping the oldest packet -- + an explicit opt-in to unbounded backpressure. Note this blocks the calling thread (the one + emitting the metric), not just the background sender. :type sender_queue_timeout: float, optional """ @@ -2176,7 +2173,9 @@ def _sender_main_loop(self, pending_queue): # next line has type ignore because the type checker cannot # know that 'if item is Stop' is the only case where item is # of object type. - sent = self._xmit_packet_with_telemetry(item.payload, queue_mode=True) # type: ignore[attr-defined] # noqa: F821 + sent = self._xmit_packet_with_telemetry( + item.payload, queue_mode=True # type: ignore[attr-defined] + ) if sent is None: # Connection trouble: keep the payload for the next attempt diff --git a/datadog/dogstatsd/sender_queue.py b/datadog/dogstatsd/sender_queue.py index 8a5922154..f7db9a181 100644 --- a/datadog/dogstatsd/sender_queue.py +++ b/datadog/dogstatsd/sender_queue.py @@ -13,7 +13,7 @@ from typing import Callable, Optional, Union # noqa: F401 -# Sentinel telling the background sender thread to shut down. +# Sentinel telling the background sender thread to shut down. Stop = object() # How long (in seconds) a non-replay-safe payload may sit in the background @@ -54,13 +54,18 @@ class SenderQueue(object): """Bounded hand-off queue between application threads and the background sender thread. put() never rejects a payload outright. When the queue is already at its - maximum size and put_timeout is falsy (the default), the oldest entry is dropped - immediately to make room, along with any additional expired entries left - at the front. When put_timeout is a positive number, put() - instead blocks the calling thread for up to that many seconds waiting - for the sender thread to drain a slot; only once that wait times out - (or immediately, if put_timeout is falsy) does it fall back to the same - drop-oldest eviction. + maximum size, what happens depends on put_timeout: + - 0 (the default): no waiting at all -- the oldest entry is dropped + immediately to make room, along with any additional expired entries + left at the front. + - None: put() blocks the calling thread indefinitely, waiting for the + sender thread to drain a slot. It will wait forever if nothing ever + does -- this is an explicit opt-in to unbounded backpressure on the + calling thread. + - a positive number: put() blocks the calling thread for up to that + many seconds waiting for a slot; if the wait times out without one + opening up, it falls back to the same drop-oldest eviction as the + 0 case. get() drops expired entries lazily too, from the front, before returning the next payload actually worth handing to the sender. @@ -73,8 +78,8 @@ class SenderQueue(object): never blocks on put_timeout. """ - def __init__(self, maxsize, expiry_seconds, on_drop_queue_full, on_drop_expired, put_timeout=None): - # type: (int, float, Callable[[PendingPayload], None], Callable[[PendingPayload], None], Optional[float]) -> None + def __init__(self, maxsize, expiry_seconds, on_drop_queue_full, on_drop_expired, put_timeout=0): + # type: (int, float, Callable[[PendingPayload], None], Callable[[PendingPayload], None], Optional[float]) -> None # noqa: E501 self._maxsize = maxsize self._expiry_seconds = expiry_seconds self._on_drop_queue_full = on_drop_queue_full @@ -135,22 +140,29 @@ def put(self, item): # type: (Union[PendingPayload, object]) -> None """Queue a payload (or the Stop sentinel). - If the queue is full: waits for room for up to put_timeout seconds - (if put_timeout is a positive number), then falls back to evicting - the oldest entry (see _make_room_locked()) if the wait timed out - without room opening up -- or immediately, with no wait at all, if - put_timeout is falsy. Either way, put() never rejects the payload - outright. + If the queue is full: waits for room according to put_timeout -- + forever if it's None, up to put_timeout seconds if it's a positive + number, or not at all if it's 0 (the default) -- then falls back to + evicting the oldest entry (see _make_room_locked()) if the queue is + still full once the wait is over. Either way, put() never rejects + the payload outright. """ with self._not_empty: if item is not Stop and self._maxsize > 0 and len(self._deque) >= self._maxsize: - if self._put_timeout: + if self._put_timeout is None: + # Wait forever: an explicit opt-in to unbounded + # backpressure on the calling thread. + while len(self._deque) >= self._maxsize: + self._not_full.wait() + elif self._put_timeout > 0: deadline = monotonic() + self._put_timeout while len(self._deque) >= self._maxsize: remaining = deadline - monotonic() if remaining <= 0: break self._not_full.wait(remaining) + # else: put_timeout is 0 (or negative) -- no wait at all, + # straight to eviction below. if len(self._deque) >= self._maxsize: self._make_room_locked() @@ -238,4 +250,3 @@ def empty(self): # type: () -> bool with self._lock: return not self._deque - diff --git a/tests/unit/dogstatsd/test_statsd.py b/tests/unit/dogstatsd/test_statsd.py index 03fb32546..5fac674cc 100644 --- a/tests/unit/dogstatsd/test_statsd.py +++ b/tests/unit/dogstatsd/test_statsd.py @@ -2729,9 +2729,13 @@ def test_bytes_dropped_queue_counts_actual_bytes(self): statsd.stop() - def test_sender_queue_put_timeout_none_evicts_immediately(self): - # Default behaviour (put_timeout falsy): no waiting at all, same as - # before this feature existed. + def test_sender_queue_put_timeout_default_evicts_immediately(self): + # Default put_timeout (0, whether omitted or explicit): no waiting + # at all, same as before this feature existed. Deliberately omits + # put_timeout here to prove the *default* -- not just 0 -- means + # "don't wait", since None means something very different (wait + # forever) and must not be the implicit default for anyone who + # constructs a SenderQueue without thinking about put_timeout at all. dropped_queue_full = [] pending_queue = SenderQueue( maxsize=1, @@ -2750,6 +2754,69 @@ def test_sender_queue_put_timeout_none_evicts_immediately(self): self.assertEqual([p.payload for p in dropped_queue_full], ["first\n"]) self.assertEqual(pending_queue.get().payload, "second\n") + def test_sender_queue_put_timeout_zero_evicts_immediately(self): + # Same as the default, but with put_timeout=0 passed explicitly. + dropped_queue_full = [] + pending_queue = SenderQueue( + maxsize=1, + expiry_seconds=100.0, + on_drop_queue_full=dropped_queue_full.append, + on_drop_expired=lambda item: self.fail("unexpected expiry drop"), + put_timeout=0, + ) + + pending_queue.put(PendingPayload("first\n", sender_queue_clock(), False)) + + t0 = time.time() + pending_queue.put(PendingPayload("second\n", sender_queue_clock(), False)) + elapsed = time.time() - t0 + + self.assertLess(elapsed, 0.05, "put() should not have waited at all") + self.assertEqual([p.payload for p in dropped_queue_full], ["first\n"]) + self.assertEqual(pending_queue.get().payload, "second\n") + + def test_sender_queue_put_timeout_none_waits_forever_and_never_evicts(self): + # put_timeout=None is an explicit opt-in to unbounded blocking: put() + # must keep waiting indefinitely -- not fall back to eviction after + # some internal default -- until room actually opens up. + dropped_queue_full = [] + pending_queue = SenderQueue( + maxsize=1, + expiry_seconds=100.0, + on_drop_queue_full=lambda item: dropped_queue_full.append(item), + on_drop_expired=lambda item: self.fail("unexpected expiry drop"), + put_timeout=None, + ) + pending_queue.put(PendingPayload("first\n", sender_queue_clock(), False)) + + result = {} + + def blocked_put(): + t0 = time.time() + pending_queue.put(PendingPayload("second\n", sender_queue_clock(), False)) + result["elapsed"] = time.time() - t0 + + t = threading.Thread(target=blocked_put) + t.start() + try: + # Nothing is draining the queue: with a real timeout this would + # have already fired and evicted "first" well before 1s. With + # None it must still be waiting. + time.sleep(1.0) + self.assertTrue(t.is_alive(), "put(timeout=None) must keep waiting, never fall back to eviction on its own") + self.assertEqual(dropped_queue_full, []) + + # Now free up room: the blocked put() should wake up and + # succeed without ever having dropped anything. + self.assertEqual(pending_queue.get().payload, "first\n") + pending_queue.task_done() + finally: + t.join(timeout=5.0) + + self.assertFalse(t.is_alive()) + self.assertEqual(dropped_queue_full, [], "put_timeout=None must never fall back to eviction") + self.assertEqual(pending_queue.get().payload, "second\n") + def test_sender_queue_put_timeout_wakes_up_when_room_opens(self): # A slot freed by get() (well within put_timeout) should wake a # blocked put() immediately rather than making it wait out the full