From 7f5fc5097366c1cd201aeb1f68fc0784e2991bc9 Mon Sep 17 00:00:00 2001 From: Johannes Maron Date: Wed, 8 Jul 2026 11:30:11 +0200 Subject: [PATCH 1/7] Switch ingress/egress telemetry to pub/sub --- tests/backends/test_redis.py | 228 ++++++++---------------- threadmill/backends/base.py | 17 +- threadmill/backends/lua/acknowledge.lua | 6 +- threadmill/backends/lua/reaper.lua | 6 +- threadmill/backends/redis.py | 120 +++++-------- threadmill/inspector/app.py | 67 ++++++- threadmill/inspector/inspector.scss | 13 ++ 7 files changed, 229 insertions(+), 228 deletions(-) diff --git a/tests/backends/test_redis.py b/tests/backends/test_redis.py index aba9206..e304ea2 100644 --- a/tests/backends/test_redis.py +++ b/tests/backends/test_redis.py @@ -61,54 +61,10 @@ def test_error_path__maintain_continues_after_exception(self, caplog): patch.object( broker, "_reap_running_queue", side_effect=RuntimeError("reaper") ), - patch.object( - default_task_backend, - "_trim_telemetry", - side_effect=RuntimeError("trim"), - ), ): broker.main() assert "Mover error for queue" in caplog.text assert "Running reaper error for queue" in caplog.text - assert "Telemetry trim error for queue" in caplog.text - - def test_main__trims_stale_telemetry(self): - """main() trims every telemetry series older than result_ttl.""" - backend = RedisTaskBackend( - "broker_trim_test", - { - "QUEUES": ["default"], - "REDIS_URL": "redis://localhost:6379/0", - "OPTIONS": { - "result_ttl": datetime.timedelta(seconds=60), - }, - }, - ) - try: - task_result = backend.enqueue(echo, args=[1]) - ingress_key = backend.INGRESS_KEY.format( - prefix=backend.key_prefix, queue_name="default" - ) - successful_key = backend._segment_key( - TaskResultStatus.SUCCESSFUL, "default" - ) - failed_key = backend._segment_key(TaskResultStatus.FAILED, "default") - old = (timezone.now() - datetime.timedelta(seconds=120)).timestamp() * 1000 - backend.client.zadd(ingress_key, {task_result.id: old}) - backend.client.zadd(successful_key, {task_result.id: old}) - backend.client.zadd(failed_key, {task_result.id: old}) - assert backend.client.zcard(ingress_key) == 1 - assert backend.client.zcard(successful_key) == 1 - assert backend.client.zcard(failed_key) == 1 - - broker = RedisBroker(backend) - broker.main() - - assert backend.client.zcard(ingress_key) == 0 - assert backend.client.zcard(successful_key) == 0 - assert backend.client.zcard(failed_key) == 0 - finally: - backend.close() class TestRedisTaskBackend: @@ -217,9 +173,10 @@ def test_running_reaper__fails_expired_tasks(self): assert len(result.errors) == 1 assert "AcknowledgementTimeout" in result.errors[0].exception_class_path - # Reaping an expired task must count as egress and failed + # Reaping an expired task records a failed result. Live egress + # now arrives via pub/sub, so backend.telemetry() rates are zero. stats = backend.telemetry().queues["default"] - assert stats.rates.egress == 1 + assert stats.rates.egress == 0 assert stats.counts.failed == 1 assert stats.counts.successful == 0 finally: @@ -286,7 +243,7 @@ def test_telemetry__empty_backend(self): backend.close() def test_telemetry__counts_tasks(self): - """Telemetry reflects windowed ingress/egress rates and per-status counts.""" + """Telemetry reports per-status counts; rates come from pub/sub, not polling.""" backend = RedisTaskBackend( "telemetry_counts_test", { @@ -327,16 +284,14 @@ def test_telemetry__counts_tasks(self): telemetry = backend.telemetry() assert telemetry.queues["default"] == _stats( - ingress=2, - egress=2, successful=1, failed=1, ) finally: backend.close() - def test_telemetry__egress_equals_successful_plus_failed(self): - """Egress is derived as successful + failed over the display window.""" + def test_telemetry__counts_successful_and_failed(self): + """Telemetry counts successful and finished results; polling rates stay zero.""" backend = RedisTaskBackend( "telemetry_egress_test", { @@ -377,8 +332,9 @@ def test_telemetry__egress_equals_successful_plus_failed(self): ) stats = backend.telemetry().queues["default"] - assert stats.rates.egress == stats.counts.successful + stats.counts.failed - assert stats.rates.egress == 3 + # Rates come from the pub/sub buffer, not from Redis polling. + assert stats.rates.ingress == 0 + assert stats.rates.egress == 0 assert stats.counts.successful == 2 assert stats.counts.failed == 1 finally: @@ -437,10 +393,31 @@ def _ack(status: TaskResultStatus) -> str: finally: backend.close() - def test_telemetry__ingress_egress_age_out_of_window(self): - """Ingress and egress older than the display window are excluded from the rates.""" + @staticmethod + def _await_message(pubsub, expected: bytes, *, timeout: float = 2.0): + """Drain pubsub until a user message with the expected payload arrives.""" + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + message = pubsub.get_message(timeout=0.1) + if message and message.get("type") == "message": + if message["data"] == expected: + return message + raise AssertionError(f"no pubsub message {expected!r} received") + + @staticmethod + def _drain_subscription(pubsub, *, timeout: float = 2.0): + """Block until the pubsub connection reports it is subscribed.""" + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if pubsub.get_message(timeout=0.1): + if pubsub.subscribed: + return + raise AssertionError("pubsub never reached subscribed state") + + def test_enqueue__publishes_ingress_telemetry(self): + """enqueue() publishes an ingress event on the telemetry channel.""" backend = RedisTaskBackend( - "telemetry_window_test", + "telemetry_publish_ingress_test", { "QUEUES": ["default"], "REDIS_URL": "redis://localhost:6379/0", @@ -449,132 +426,81 @@ def test_telemetry__ingress_egress_age_out_of_window(self): }, }, ) + pubsub = backend.client.pubsub() try: - task_result = backend.enqueue(echo, args=[42]) - acquired = backend.acquire( - timeout=datetime.timedelta(seconds=1), worker="window-test" - ) - assert acquired is not None - backend.acknowledge( - dataclasses.replace( - acquired, - status=TaskResultStatus.SUCCESSFUL, - finished_at=timezone.now(), - ) - ) - - ingress_key = backend.INGRESS_KEY.format( - prefix=backend.key_prefix, queue_name="default" - ) - successful_results_key = backend._segment_key( - TaskResultStatus.SUCCESSFUL, "default" - ) - old = (timezone.now() - datetime.timedelta(seconds=120)).timestamp() * 1000 - backend.client.zadd(ingress_key, {task_result.id: old}) - backend.client.zadd(successful_results_key, {task_result.id: old}) - - stats = backend.telemetry().queues["default"] - # Older than the 60s display window: not counted as recent throughput. - assert stats.rates.ingress == 0 - assert stats.rates.egress == 0 - # Ingress is evicted by the read path (retention = result_ttl = 60s). - assert backend.client.zcard(ingress_key) == 0 - # The results segment is evicted by acknowledge/reaper, not by reads, - # so the aged result is still present in the segment count. - assert backend.client.zcard(successful_results_key) == 1 - assert stats.counts.successful == 1 + pubsub.subscribe(backend.telemetry_channel) + self._drain_subscription(pubsub) + backend.enqueue(echo, args=[1]) + message = self._await_message(pubsub, b"ingress:default") + assert message["channel"] == backend.telemetry_channel.encode() finally: + pubsub.unsubscribe(backend.telemetry_channel) + pubsub.close() backend.close() - def test_telemetry__interval_shrinks_window(self): - """A custom interval excludes events from the rate but never trims below result_ttl.""" + def test_acknowledge__publishes_egress_telemetry(self): + """acknowledge() publishes an egress event on the telemetry channel.""" backend = RedisTaskBackend( - "telemetry_interval_test", + "telemetry_publish_egress_test", { "QUEUES": ["default"], "REDIS_URL": "redis://localhost:6379/0", "OPTIONS": { + "lease_ttl": datetime.timedelta(hours=1), "result_ttl": datetime.timedelta(seconds=60), }, }, ) + pubsub = backend.client.pubsub() try: - task_result = backend.enqueue(echo, args=[42]) - ingress_key = backend.INGRESS_KEY.format( - prefix=backend.key_prefix, queue_name="default" + backend.enqueue(echo, args=[1]) + acquired = backend.acquire( + timeout=datetime.timedelta(seconds=1), worker="publish-test" ) - recent = (timezone.now() - datetime.timedelta(seconds=5)).timestamp() * 1000 - backend.client.zadd(ingress_key, {task_result.id: recent}) - - # The 30s-old event is inside the default 60s window but outside a 10s one. - stats = backend.telemetry(interval=datetime.timedelta(seconds=10)).queues[ - "default" - ] - assert stats.rates.ingress == 1 - - stale = (timezone.now() - datetime.timedelta(seconds=30)).timestamp() * 1000 - backend.client.zadd(ingress_key, {task_result.id: stale}) - stats = backend.telemetry(interval=datetime.timedelta(seconds=10)).queues[ - "default" - ] - # Outside the 10s count window, so not displayed... - assert stats.rates.ingress == 0 - # ...but still retained for result_ttl (60s): a narrower display - # window must not clobber the retention policy. - assert backend.client.zcard(ingress_key) == 1 + pubsub.subscribe(backend.telemetry_channel) + self._drain_subscription(pubsub) + backend.acknowledge( + dataclasses.replace( + acquired, + status=TaskResultStatus.SUCCESSFUL, + finished_at=timezone.now(), + ) + ) + message = self._await_message(pubsub, b"egress:default") + assert message["channel"] == backend.telemetry_channel.encode() finally: + pubsub.unsubscribe(backend.telemetry_channel) + pubsub.close() backend.close() - def test_telemetry__egress_window_follows_display_interval(self): - """Egress rates are windowed by the display interval, not by result_ttl.""" + def test_reaper__publishes_egress_telemetry(self): + """reaper.lua publishes an egress event per reaped task.""" backend = RedisTaskBackend( - "telemetry_egress_window_test", + "telemetry_publish_reaper_test", { "QUEUES": ["default"], "REDIS_URL": "redis://localhost:6379/0", "OPTIONS": { + "lease_ttl": datetime.timedelta(seconds=1), "result_ttl": datetime.timedelta(seconds=60), }, }, ) + pubsub = backend.client.pubsub() try: backend.enqueue(echo, args=[1]) - acquired = backend.acquire( - timeout=datetime.timedelta(seconds=1), worker="egress-window-test" - ) - assert acquired is not None - backend.acknowledge( - dataclasses.replace( - acquired, - status=TaskResultStatus.SUCCESSFUL, - finished_at=timezone.now(), - ) + backend.acquire( + timeout=datetime.timedelta(seconds=1), worker="reaper-publish-test" ) - successful_key = backend._segment_key( - TaskResultStatus.SUCCESSFUL, "default" - ) - - # Inside result_ttl (60s) but outside a 10s display window. - recent = ( - timezone.now() - datetime.timedelta(seconds=30) - ).timestamp() * 1000 - backend.client.zadd(successful_key, {acquired.id: recent}) - - # A narrow window excludes the 30s-old result from the egress rate... - narrow = backend.telemetry(interval=datetime.timedelta(seconds=10)).queues[ - "default" - ] - assert narrow.rates.egress == 0 - # ...but it is still a present segment member (ZCARD is time-independent). - assert narrow.counts.successful == 1 - - # A wide window (60s) includes the 30s-old result in the egress rate. - wide = backend.telemetry(interval=datetime.timedelta(seconds=60)).queues[ - "default" - ] - assert wide.rates.egress == 1 - assert wide.counts.successful == 1 + pubsub.subscribe(backend.telemetry_channel) + self._drain_subscription(pubsub) + time.sleep(1.1) + RedisBroker(backend).main() + message = self._await_message(pubsub, b"egress:default") + assert message["channel"] == backend.telemetry_channel.encode() finally: + pubsub.unsubscribe(backend.telemetry_channel) + pubsub.close() backend.close() def _acknowledge(self, status: TaskResultStatus) -> str: diff --git a/threadmill/backends/base.py b/threadmill/backends/base.py index 313323a..ba7eda5 100644 --- a/threadmill/backends/base.py +++ b/threadmill/backends/base.py @@ -245,13 +245,26 @@ def peek( def telemetry( self, *, interval: datetime.timedelta = datetime.timedelta(seconds=60) ) -> BackendTelemetry: - """Return a snapshot of stats for all configured queues. + """Return a snapshot of counts for all configured queues. + + Rates are no longer populated by polling; live throughput arrives via + :meth:`subscribe_telemetry` for backends that support it. Args: - interval: The time window for rolling rates. + interval: The rolling window the caller intends to display. """ raise NotImplementedError + def subscribe_telemetry(self) -> collections.abc.AsyncIterator[str] | None: + """Yield telemetry events as ``"{direction}:{queue_name}"`` strings. + + Backends without pub/sub support return ``None`` so the inspector can + fall back to counts-only telemetry. The returned async iterator is + consumed on the inspector's worker thread and must clean up its + subscription on close. + """ + return None + def dequeue(self, task_result: TaskResult) -> None: """Delete a single task from its current status segment.""" raise NotImplementedError diff --git a/threadmill/backends/lua/acknowledge.lua b/threadmill/backends/lua/acknowledge.lua index 9395947..896f02f 100644 --- a/threadmill/backends/lua/acknowledge.lua +++ b/threadmill/backends/lua/acknowledge.lua @@ -3,7 +3,8 @@ -- per-status results history set. Evicts results whose finish score falls -- outside the retention window (result_ttl). The per-status history sets are -- the time series the inspector counts over for telemetry, so no separate --- egress window or status counters are needed here. +-- egress window or status counters are needed here. A pub/sub event is +-- published so the inspector can track live egress throughput. -- -- KEYS[1] -- running set (ZSET) -- KEYS[2] -- result key (STRING, stores serialized TaskResult) @@ -15,6 +16,8 @@ -- ARGV[3] -- result TTL in seconds -- ARGV[4] -- finish timestamp in milliseconds (score for the history set) -- ARGV[5] -- status (SUCCESSFUL or FAILED) +-- ARGV[6] -- telemetry pub/sub channel +-- ARGV[7] -- queue name -- Returns: 1 on success, 0 if task was not in the running set local removed = redis.call('ZREM', KEYS[1], ARGV[1]) @@ -31,4 +34,5 @@ if ARGV[5] ~= 'SUCCESSFUL' then end redis.call('ZADD', results_key, finish, ARGV[1]) redis.call('ZREMRANGEBYSCORE', results_key, 0, cutoff) +redis.call('PUBLISH', ARGV[6], 'egress:' .. ARGV[7]) return 1 diff --git a/threadmill/backends/lua/reaper.lua b/threadmill/backends/lua/reaper.lua index 31f11f0..6bd266a 100644 --- a/threadmill/backends/lua/reaper.lua +++ b/threadmill/backends/lua/reaper.lua @@ -1,6 +1,7 @@ -- Fail tasks whose processing lease has expired. Reaped tasks are recorded -- in the failed results history (the time series the inspector counts over), --- then evicted once older than result_ttl. +-- then evicted once older than result_ttl. A pub/sub event is published per +-- reaped task so the inspector can track live egress throughput. -- -- KEYS[1] -- running set (ZSET) -- KEYS[2] -- failed results history (ZSET, scored by finish time) @@ -10,6 +11,8 @@ -- ARGV[4] -- batch size -- ARGV[5] -- result TTL in seconds -- ARGV[6] -- finished_at as ISO format string +-- ARGV[7] -- telemetry pub/sub channel +-- ARGV[8] -- queue name -- Returns: number of tasks failed local stale = redis.call('ZRANGEBYSCORE', KEYS[1], 0, ARGV[1], 'LIMIT', 0, tonumber(ARGV[4])) @@ -32,6 +35,7 @@ for _, task_id in ipairs(stale) do redis.call('SET', ARGV[3] .. task_id, failed_data, 'EX', ARGV[5]) redis.call('DEL', ARGV[2] .. task_id) redis.call('ZADD', KEYS[2], tonumber(ARGV[1]), task_id) + redis.call('PUBLISH', ARGV[7], 'egress:' .. ARGV[8]) end end end diff --git a/threadmill/backends/redis.py b/threadmill/backends/redis.py index 8029e30..0a9ce33 100644 --- a/threadmill/backends/redis.py +++ b/threadmill/backends/redis.py @@ -1,5 +1,6 @@ """Redis-backed durable priority queue backend for Django's task framework.""" +import collections.abc import dataclasses import datetime import logging @@ -10,6 +11,7 @@ from pathlib import Path import redis +import redis.asyncio from django.tasks import DEFAULT_TASK_QUEUE_NAME, TaskResult, TaskResultStatus from django.tasks.exceptions import TaskResultDoesNotExist from django.tasks.signals import task_enqueued @@ -83,12 +85,13 @@ def _reap_running_queue(self, queue_name: str) -> None: str(self.backend.batch_size), str(int(self.backend.result_ttl.total_seconds())), finished_at_iso, + self.backend.telemetry_channel, + queue_name, ], ) def main(self) -> None: """Run mover and running reaper passes for all queues.""" - now_ms = time.time() * 1000 for queue_name in self.backend.queues: try: self._move_queue(queue_name) @@ -100,11 +103,6 @@ def main(self) -> None: except Exception: # noqa: BLE001 logger.exception("Running reaper error for queue %r", queue_name) - try: - self.backend._trim_telemetry(queue_name, now_ms=now_ms) - except Exception: # noqa: BLE001 - logger.exception("Telemetry trim error for queue %r", queue_name) - class RedisTaskBackend(ThreadmillTaskBackend): """Redis-backed durable priority queue backend. @@ -126,7 +124,7 @@ class RedisTaskBackend(ThreadmillTaskBackend): SEGMENT_KEY = "{prefix}:{queue_name}:{status}" DEFERRED_KEY = "{prefix}:{queue_name}:deferred" - INGRESS_KEY = "{prefix}:{queue_name}:ingress:events" + TELEMETRY_CHANNEL = "{prefix}:telemetry" ACQUIRE_SCRIPT = _load_lua("acquire") """Pop the next task from a priority queue and move it directly to the running set.""" @@ -150,7 +148,9 @@ def __init__(self, alias: str, params: dict) -> None: f"REDIS_URL must be specified in your settings for the {type(self).__name__}." ) from e self.client = redis.from_url(redis_url) + self.redis_url = redis_url self.key_prefix = f"threadmill:{{{alias}}}" + self.telemetry_channel = self.TELEMETRY_CHANNEL.format(prefix=self.key_prefix) self.lease_ttl = self.options.get("lease_ttl", datetime.timedelta(hours=1)) self.result_ttl = self.options.get("result_ttl", datetime.timedelta(days=1)) self.batch_size = self.options.get("batch_size", 100) @@ -196,15 +196,11 @@ def enqueue( ) score = self._compute_score(task.priority, enqueued_at) - enqueued_at_ms = enqueued_at.timestamp() * 1000 serialized = self.serialize_task_result(task_result) task_key = self.TASK_KEY.format(prefix=self.key_prefix, task_id=task_result.id) task_data_ttl = int( self.lease_ttl.total_seconds() * 3 + self.result_ttl.total_seconds() ) - ingress_key = self.INGRESS_KEY.format( - prefix=self.key_prefix, queue_name=task.queue_name - ) pipe = self.client.pipeline() pipe.hset( @@ -216,7 +212,7 @@ def enqueue( }, ) pipe.expire(task_key, task_data_ttl) - pipe.zadd(ingress_key, {task_result.id: enqueued_at_ms}) + pipe.publish(self.telemetry_channel, f"ingress:{task.queue_name}") if task.run_after is not None: deferred_key = self.DEFERRED_KEY.format( @@ -310,6 +306,8 @@ def acknowledge(self, task_result: TaskResult) -> None: str(int(self.result_ttl.total_seconds())), str(finish_score), task_result.status.name, + self.telemetry_channel, + task_result.task.queue_name, ], ) @@ -416,33 +414,14 @@ def get_result(self, result_id: str) -> TaskResult: return self.deserialize_task_result(data) raise TaskResultDoesNotExist(f"Task result {result_id!r} does not exist.") - def _trim_telemetry(self, queue_name: str, *, now_ms: float) -> None: - cutoff = now_ms - self.result_ttl.total_seconds() * 1000 - pipe = self.client.pipeline() - pipe.zremrangebyscore( - self.INGRESS_KEY.format(prefix=self.key_prefix, queue_name=queue_name), - 0, - cutoff, - ) - pipe.zremrangebyscore( - self._segment_key(TaskResultStatus.SUCCESSFUL, queue_name), - 0, - cutoff, - ) - pipe.zremrangebyscore( - self._segment_key(TaskResultStatus.FAILED, queue_name), - 0, - cutoff, - ) - pipe.execute() - def telemetry( self, *, interval: datetime.timedelta = datetime.timedelta(seconds=60) ) -> BackendTelemetry: - now_ms = time.time() * 1000 - window_start_ms = now_ms - interval.total_seconds() * 1000 - exclusive_start = f"({window_start_ms}" - retention_start_ms = now_ms - self.result_ttl.total_seconds() * 1000 + """Return point-in-time counts for all configured queues. + + Rates are zero here; live throughput arrives via pub/sub on + :meth:`subscribe_telemetry` and is assembled by the inspector. + """ pipe = self.client.pipeline() for queue_name in self.queues: pipe.zcard(self._segment_key(TaskResultStatus.READY, queue_name)) @@ -450,50 +429,49 @@ def telemetry( pipe.zcard( self.DEFERRED_KEY.format(prefix=self.key_prefix, queue_name=queue_name) ) - successful_results_key = self._segment_key( - TaskResultStatus.SUCCESSFUL, queue_name - ) - failed_results_key = self._segment_key(TaskResultStatus.FAILED, queue_name) - pipe.zcard(successful_results_key) - pipe.zcard(failed_results_key) - ingress_key = self.INGRESS_KEY.format( - prefix=self.key_prefix, queue_name=queue_name - ) - pipe.zcount(ingress_key, exclusive_start, now_ms) - pipe.zcount(successful_results_key, exclusive_start, now_ms) - pipe.zcount(failed_results_key, exclusive_start, now_ms) - pipe.zremrangebyscore(ingress_key, 0, retention_start_ms) + pipe.zcard(self._segment_key(TaskResultStatus.SUCCESSFUL, queue_name)) + pipe.zcard(self._segment_key(TaskResultStatus.FAILED, queue_name)) results = pipe.execute() + zero_rates = QueueRates(interval=interval, ingress=0, egress=0) queues: dict[str, QueueStats] = {} for index, queue_name in enumerate(self.queues): - base = index * 9 - ( - ready, - running, - deferred, - successful, - failed, - ingress, - successful_rate, - failed_rate, - _, - ) = (int(c or 0) for c in results[base : base + 9]) + base = index * 5 queues[queue_name] = QueueStats( counts=QueueCounts( - ready=ready, - running=running, - deferred=deferred, - successful=successful, - failed=failed, - ), - rates=QueueRates( - interval=interval, - ingress=ingress, - egress=successful_rate + failed_rate, + ready=int(results[base] or 0), + running=int(results[base + 1] or 0), + deferred=int(results[base + 2] or 0), + successful=int(results[base + 3] or 0), + failed=int(results[base + 4] or 0), ), + rates=zero_rates, ) return BackendTelemetry(queues=queues) + def subscribe_telemetry(self) -> collections.abc.AsyncIterator[str] | None: + """Subscribe to the backend's telemetry pub/sub channel. + + Yields decoded ``"{direction}:{queue_name}"`` payloads. The connection + is opened with the asyncio client so the inspector can consume it off + the UI thread. + """ + return self._telemetry_stream() + + async def _telemetry_stream(self) -> collections.abc.AsyncIterator[str]: + client = redis.asyncio.Redis.from_url(self.redis_url) + pubsub = client.pubsub() + await pubsub.subscribe(self.telemetry_channel) + try: + while message := await pubsub.get_message( + ignore_subscribe_messages=True, timeout=None + ): + if (data := message.get("data")) is not None: + yield data.decode() if isinstance(data, bytes) else data + finally: + await pubsub.unsubscribe(self.telemetry_channel) + await pubsub.aclose() + await client.aclose() + def close(self) -> None: """Close the Redis connection.""" self.client.close() diff --git a/threadmill/inspector/app.py b/threadmill/inspector/app.py index 3b27ad9..998945e 100644 --- a/threadmill/inspector/app.py +++ b/threadmill/inspector/app.py @@ -1,8 +1,11 @@ """Textual app for the inspector TUI.""" +import asyncio +import dataclasses import datetime import logging import math +import time import typing from typing import Any @@ -13,6 +16,7 @@ TaskResultStatus, task_backends, ) +from textual import work from textual.app import App, ComposeResult from textual.binding import Binding from textual.containers import Horizontal, Vertical @@ -24,6 +28,7 @@ ListItem, ListView, Select, + Sparkline, Static, TabbedContent, TabPane, @@ -31,12 +36,16 @@ from ..backends.base import BackendTelemetry, ThreadmillTaskBackend from .screens import ConfirmScreen, PurgeScreen +from .telemetry import TelemetryBuffer logger = logging.getLogger(__name__) TELEMETRY_INTERVAL_SECONDS = 2.0 """Seconds between automatic queue-stat refreshes; the task list stays manual.""" +SPARKLINE_INTERVAL_SECONDS = 1.0 +"""Seconds between sparkline refreshes from the rolling telemetry buffer.""" + TAB_STATUSES: list[tuple[str, TaskResultStatus | None]] = [ ("Running", TaskResultStatus.RUNNING), ("Ready", TaskResultStatus.READY), @@ -460,7 +469,9 @@ def __init__( self._task_detail: TaskDetail | None = None self._options_static: Static | None = None self._telemetry_timer = None + self._sparkline_timer = None self._auto_refresh = auto_refresh + self.telemetry_buffer = TelemetryBuffer() self.set_reactive(InspectorApp.backend, backend) def compose(self) -> ComposeResult: @@ -481,6 +492,15 @@ def compose(self) -> ComposeResult: telemetry=InspectorApp.telemetry ) with Vertical(id="right-pane"): + with Horizontal(id="sparkline-row"): + yield Sparkline( + id="ingress-sparkline", + max_color="ansi_green", + ) + yield Sparkline( + id="egress-sparkline", + max_color="ansi_red", + ) yield TaskList(id="task-list").data_bind( backend=InspectorApp.backend, telemetry=InspectorApp.telemetry, @@ -497,12 +517,18 @@ def on_mount(self) -> None: self._options_static = self.query_one("#backend-options", Static) self._refresh_options() self._refresh_telemetry() + self._listen_telemetry() if self._auto_refresh: self._telemetry_timer = self.set_interval( TELEMETRY_INTERVAL_SECONDS, self._refresh_telemetry, name="telemetry-refresh", ) + self._sparkline_timer = self.set_interval( + SPARKLINE_INTERVAL_SECONDS, + self._refresh_sparklines, + name="sparkline-refresh", + ) self._queue_list.focus() def action_quit(self) -> None: @@ -522,6 +548,7 @@ def action_switch_tab(self, tab_id: str) -> None: def watch_backend(self) -> None: self._refresh_options() self._refresh_telemetry() + self._listen_telemetry() def on_select_changed(self, event: Select.Changed) -> None: """Switch backend when the user selects a different alias.""" @@ -549,8 +576,44 @@ def _refresh_options(self) -> None: self._options_static.update(" ".join(parts) or "No options") def _refresh_telemetry(self) -> None: - """Poll the backend for fresh telemetry.""" + """Poll the backend for counts and merge live pub/sub rates for the selected queue.""" try: - self.telemetry = self.backend.telemetry() + telemetry = self.backend.telemetry() except Exception: # noqa: BLE001 logger.exception("Failed to refresh telemetry") + return + queue_name = self._task_list.queue_name + if queue_name and queue_name in telemetry.queues: + live_rates = self.telemetry_buffer.rates_for(queue_name) + stats = telemetry.queues[queue_name] + telemetry.queues[queue_name] = dataclasses.replace(stats, rates=live_rates) + self.telemetry = telemetry + + def _refresh_sparklines(self) -> None: + """Push the latest 60-second rolling series into the ingress/egress sparklines.""" + queue_name = self._task_list.queue_name + ingress = self.query_one("#ingress-sparkline", Sparkline) + egress = self.query_one("#egress-sparkline", Sparkline) + now = time.monotonic() + ingress.data = ( + self.telemetry_buffer.series(queue_name, "ingress", now=now) + if queue_name + else [0.0] * 60 + ) + egress.data = ( + self.telemetry_buffer.series(queue_name, "egress", now=now) + if queue_name + else [0.0] * 60 + ) + + @work(exclusive=True, group="telemetry") + async def _listen_telemetry(self) -> None: + """Subscribe to the backend's pub/sub telemetry stream and feed the buffer.""" + stream = self.backend.subscribe_telemetry() + if stream is None: + return + try: + async for payload in stream: + self.telemetry_buffer.record(payload, now=time.monotonic()) + except asyncio.CancelledError: + raise diff --git a/threadmill/inspector/inspector.scss b/threadmill/inspector/inspector.scss index 067f694..e1256a2 100644 --- a/threadmill/inspector/inspector.scss +++ b/threadmill/inspector/inspector.scss @@ -42,6 +42,19 @@ Select { width: 42; } +#sparkline-row { + height: 2; + margin-bottom: 0; +} + +#ingress-sparkline, +#egress-sparkline { + width: 1fr; + height: 1; + padding: 0; + margin: 0; +} + #backend-row { height: 3; } From a82a04f2b26482f0fe4663c2b3074dd9248ebb19 Mon Sep 17 00:00:00 2001 From: Johannes Maron Date: Wed, 8 Jul 2026 13:44:20 +0200 Subject: [PATCH 2/7] wip --- tests/backends/test_base.py | 6 +-- tests/backends/test_redis.py | 35 +++++++------ tests/test_inspector.py | 18 +++---- threadmill/backends/base.py | 35 ++++++++----- threadmill/backends/lua/reaper.lua | 6 +-- threadmill/backends/redis.py | 28 +++++++--- threadmill/inspector/app.py | 79 +++++++++++++++------------- threadmill/inspector/telemetry.py | 82 ++++++++++++++++++++++++++++++ 8 files changed, 202 insertions(+), 87 deletions(-) create mode 100644 threadmill/inspector/telemetry.py diff --git a/tests/backends/test_base.py b/tests/backends/test_base.py index b72deda..b076338 100644 --- a/tests/backends/test_base.py +++ b/tests/backends/test_base.py @@ -81,10 +81,10 @@ def test_peek__raise_not_implemented_error(self) -> None: ) ) - def test_telemetry__raise_not_implemented_error(self) -> None: - """Raise NotImplementedError for backend telemetry API.""" + def test_queue_stats__raise_not_implemented_error(self) -> None: + """Raise NotImplementedError for backend queue_stats API.""" with pytest.raises(NotImplementedError): - BackendDouble(alias="default", params={}).telemetry() + BackendDouble(alias="default", params={}).queue_stats() def test_dequeue__raise_not_implemented_error(self) -> None: """Raise NotImplementedError for backend dequeue API.""" diff --git a/tests/backends/test_redis.py b/tests/backends/test_redis.py index e304ea2..b538df7 100644 --- a/tests/backends/test_redis.py +++ b/tests/backends/test_redis.py @@ -174,8 +174,8 @@ def test_running_reaper__fails_expired_tasks(self): assert "AcknowledgementTimeout" in result.errors[0].exception_class_path # Reaping an expired task records a failed result. Live egress - # now arrives via pub/sub, so backend.telemetry() rates are zero. - stats = backend.telemetry().queues["default"] + # now arrives via pub/sub, so backend.queue_stats() rates are zero. + stats = backend.queue_stats().queues["default"] assert stats.rates.egress == 0 assert stats.counts.failed == 1 assert stats.counts.successful == 0 @@ -237,7 +237,7 @@ def test_telemetry__empty_backend(self): }, ) try: - telemetry = backend.telemetry() + telemetry = backend.queue_stats() assert telemetry == BackendTelemetry(queues={"default": _stats()}) finally: backend.close() @@ -282,7 +282,7 @@ def test_telemetry__counts_tasks(self): ) ) - telemetry = backend.telemetry() + telemetry = backend.queue_stats() assert telemetry.queues["default"] == _stats( successful=1, failed=1, @@ -331,7 +331,7 @@ def test_telemetry__counts_successful_and_failed(self): ) ) - stats = backend.telemetry().queues["default"] + stats = backend.queue_stats().queues["default"] # Rates come from the pub/sub buffer, not from Redis polling. assert stats.rates.ingress == 0 assert stats.rates.egress == 0 @@ -387,7 +387,7 @@ def _ack(status: TaskResultStatus) -> str: assert backend.client.zcard(successful_key) == 1 assert backend.client.zcard(failed_key) == 1 - stats = backend.telemetry().queues["default"] + stats = backend.queue_stats().queues["default"] assert stats.counts.successful == 1 assert stats.counts.failed == 1 finally: @@ -473,15 +473,15 @@ def test_acknowledge__publishes_egress_telemetry(self): pubsub.close() backend.close() - def test_reaper__publishes_egress_telemetry(self): - """reaper.lua publishes an egress event per reaped task.""" + def test_requeue__publishes_ingress_telemetry(self): + """requeue() publishes an ingress event for the re-queued task.""" backend = RedisTaskBackend( - "telemetry_publish_reaper_test", + "telemetry_publish_requeue_test", { "QUEUES": ["default"], "REDIS_URL": "redis://localhost:6379/0", "OPTIONS": { - "lease_ttl": datetime.timedelta(seconds=1), + "lease_ttl": datetime.timedelta(hours=1), "result_ttl": datetime.timedelta(seconds=60), }, }, @@ -489,14 +489,19 @@ def test_reaper__publishes_egress_telemetry(self): pubsub = backend.client.pubsub() try: backend.enqueue(echo, args=[1]) - backend.acquire( - timeout=datetime.timedelta(seconds=1), worker="reaper-publish-test" + acquired = backend.acquire( + timeout=datetime.timedelta(seconds=1), worker="requeue-publish-test" + ) + assert acquired is not None + failed = dataclasses.replace( + acquired, + status=TaskResultStatus.FAILED, + finished_at=timezone.now(), ) pubsub.subscribe(backend.telemetry_channel) self._drain_subscription(pubsub) - time.sleep(1.1) - RedisBroker(backend).main() - message = self._await_message(pubsub, b"egress:default") + backend.requeue(failed, timezone.now() + datetime.timedelta(seconds=10)) + message = self._await_message(pubsub, b"ingress:default") assert message["channel"] == backend.telemetry_channel.encode() finally: pubsub.unsubscribe(backend.telemetry_channel) diff --git a/tests/test_inspector.py b/tests/test_inspector.py index 0b006d9..847a1e4 100644 --- a/tests/test_inspector.py +++ b/tests/test_inspector.py @@ -37,7 +37,7 @@ class FailingBackend(ThreadmillTaskBackend): - """Backend double whose peek and telemetry raise for error-path tests.""" + """Backend double whose peek and queue_stats raise for error-path tests.""" def enqueue(self, task, args, kwargs): raise NotImplementedError @@ -45,8 +45,8 @@ def enqueue(self, task, args, kwargs): def peek(self, *args, **kwargs): raise RuntimeError("peek failed") - def telemetry(self, *, interval=None): - raise RuntimeError("telemetry failed") + def queue_stats(self, *, interval=None): + raise RuntimeError("queue_stats failed") class ErroringBackend(RedisTaskBackend): @@ -250,7 +250,7 @@ async def test_telemetry_refresh_updates_and_prunes_queues(self): app = InspectorApp(backend=default_task_backend, auto_refresh=False) async with app.run_test() as pilot: await pilot.pause() - app._refresh_telemetry() + app.telemetry = app.backend.queue_stats() await pilot.pause() queue_list = app.query_one("#queue-list", QueueList) assert set(queue_list._items) == set(default_task_backend.queues) @@ -457,17 +457,17 @@ async def test_successful_tab_lists_finished_task(self): assert table.row_count >= 1 async def test_auto_refresh_timer_armed(self): - """With auto-refresh on, a telemetry timer is armed on mount.""" + """With auto-refresh on, the sparkline timer is armed on mount.""" app = InspectorApp(backend=default_task_backend) async with app.run_test() as pilot: await pilot.pause() await pilot.pause() - assert app._telemetry_timer is not None - assert app._telemetry_timer.name == "telemetry-refresh" + assert app._sparkline_timer is not None + assert app._sparkline_timer.name == "sparkline-refresh" app = InspectorApp(backend=default_task_backend, auto_refresh=False) async with app.run_test() as pilot: await pilot.pause() - assert app._telemetry_timer is None + assert app._sparkline_timer is None async def test_initial_focus_on_queue_list(self): """The app opens with focus on the queue list, not the backend selector.""" @@ -495,7 +495,7 @@ async def test_telemetry_refresh_does_not_re_peek_task_list(self): default_task_backend.acquire( timeout=datetime.timedelta(seconds=1), worker="stale-test" ) - app.telemetry = app.backend.telemetry() + app.telemetry = app.backend.queue_stats() await pilot.pause() await pilot.pause() assert table.row_count == before diff --git a/threadmill/backends/base.py b/threadmill/backends/base.py index ba7eda5..2f22a09 100644 --- a/threadmill/backends/base.py +++ b/threadmill/backends/base.py @@ -1,6 +1,7 @@ import collections.abc import dataclasses import datetime +import enum import json import threading import typing @@ -95,6 +96,21 @@ class BackendTelemetry: queues: dict[str, QueueStats] +class TelemetryDirection(enum.Enum): + """Direction of a telemetry event: a task entering or leaving a queue.""" + + INGRESS = "ingress" + EGRESS = "egress" + + +@dataclasses.dataclass(frozen=True, slots=True) +class TelemetryEvent: + """A single ingress/egress event published by a worker via pub/sub.""" + + direction: TelemetryDirection + queue_name: str + + class Broker(threading.Thread): """Backend maintenance thread launched by the task executor.""" @@ -242,28 +258,21 @@ def peek( """ raise NotImplementedError - def telemetry( + def queue_stats( self, *, interval: datetime.timedelta = datetime.timedelta(seconds=60) ) -> BackendTelemetry: """Return a snapshot of counts for all configured queues. - Rates are no longer populated by polling; live throughput arrives via - :meth:`subscribe_telemetry` for backends that support it. - Args: interval: The rolling window the caller intends to display. """ raise NotImplementedError - def subscribe_telemetry(self) -> collections.abc.AsyncIterator[str] | None: - """Yield telemetry events as ``"{direction}:{queue_name}"`` strings. - - Backends without pub/sub support return ``None`` so the inspector can - fall back to counts-only telemetry. The returned async iterator is - consumed on the inspector's worker thread and must clean up its - subscription on close. - """ - return None + def worker_telemetry( + self, + ) -> collections.abc.AsyncIterator[TelemetryEvent]: + """Yield structured telemetry events from the backend's pub/sub stream.""" + raise NotImplementedError def dequeue(self, task_result: TaskResult) -> None: """Delete a single task from its current status segment.""" diff --git a/threadmill/backends/lua/reaper.lua b/threadmill/backends/lua/reaper.lua index 6bd266a..31f11f0 100644 --- a/threadmill/backends/lua/reaper.lua +++ b/threadmill/backends/lua/reaper.lua @@ -1,7 +1,6 @@ -- Fail tasks whose processing lease has expired. Reaped tasks are recorded -- in the failed results history (the time series the inspector counts over), --- then evicted once older than result_ttl. A pub/sub event is published per --- reaped task so the inspector can track live egress throughput. +-- then evicted once older than result_ttl. -- -- KEYS[1] -- running set (ZSET) -- KEYS[2] -- failed results history (ZSET, scored by finish time) @@ -11,8 +10,6 @@ -- ARGV[4] -- batch size -- ARGV[5] -- result TTL in seconds -- ARGV[6] -- finished_at as ISO format string --- ARGV[7] -- telemetry pub/sub channel --- ARGV[8] -- queue name -- Returns: number of tasks failed local stale = redis.call('ZRANGEBYSCORE', KEYS[1], 0, ARGV[1], 'LIMIT', 0, tonumber(ARGV[4])) @@ -35,7 +32,6 @@ for _, task_id in ipairs(stale) do redis.call('SET', ARGV[3] .. task_id, failed_data, 'EX', ARGV[5]) redis.call('DEL', ARGV[2] .. task_id) redis.call('ZADD', KEYS[2], tonumber(ARGV[1]), task_id) - redis.call('PUBLISH', ARGV[7], 'egress:' .. ARGV[8]) end end end diff --git a/threadmill/backends/redis.py b/threadmill/backends/redis.py index 0a9ce33..dd1c526 100644 --- a/threadmill/backends/redis.py +++ b/threadmill/backends/redis.py @@ -23,6 +23,8 @@ QueueCounts, QueueRates, QueueStats, + TelemetryDirection, + TelemetryEvent, ThreadmillTaskBackend, ) @@ -85,8 +87,6 @@ def _reap_running_queue(self, queue_name: str) -> None: str(self.backend.batch_size), str(int(self.backend.result_ttl.total_seconds())), finished_at_iso, - self.backend.telemetry_channel, - queue_name, ], ) @@ -345,6 +345,7 @@ def requeue(self, task_result: TaskResult, run_after: datetime.datetime) -> None pipe.hset(task_key, mapping={"data": serialized, "score": str(score)}) pipe.expire(task_key, task_data_ttl) pipe.zadd(deferred_key, {task_result.id: run_after_ms}) + pipe.publish(self.telemetry_channel, f"ingress:{task_result.task.queue_name}") pipe.execute() def dequeue(self, task_result: TaskResult) -> None: @@ -414,13 +415,13 @@ def get_result(self, result_id: str) -> TaskResult: return self.deserialize_task_result(data) raise TaskResultDoesNotExist(f"Task result {result_id!r} does not exist.") - def telemetry( + def queue_stats( self, *, interval: datetime.timedelta = datetime.timedelta(seconds=60) ) -> BackendTelemetry: """Return point-in-time counts for all configured queues. Rates are zero here; live throughput arrives via pub/sub on - :meth:`subscribe_telemetry` and is assembled by the inspector. + :meth:`subscribe_telemetry_events` and is assembled by the inspector. """ pipe = self.client.pipeline() for queue_name in self.queues: @@ -448,16 +449,18 @@ def telemetry( ) return BackendTelemetry(queues=queues) - def subscribe_telemetry(self) -> collections.abc.AsyncIterator[str] | None: + def worker_telemetry( + self, + ) -> collections.abc.AsyncIterator[TelemetryEvent] | None: """Subscribe to the backend's telemetry pub/sub channel. - Yields decoded ``"{direction}:{queue_name}"`` payloads. The connection + Yields structured :class:`TelemetryEvent` instances. The connection is opened with the asyncio client so the inspector can consume it off the UI thread. """ return self._telemetry_stream() - async def _telemetry_stream(self) -> collections.abc.AsyncIterator[str]: + async def _telemetry_stream(self) -> collections.abc.AsyncIterator[TelemetryEvent]: client = redis.asyncio.Redis.from_url(self.redis_url) pubsub = client.pubsub() await pubsub.subscribe(self.telemetry_channel) @@ -466,12 +469,21 @@ async def _telemetry_stream(self) -> collections.abc.AsyncIterator[str]: ignore_subscribe_messages=True, timeout=None ): if (data := message.get("data")) is not None: - yield data.decode() if isinstance(data, bytes) else data + payload = data.decode() if isinstance(data, bytes) else data + yield self._parse_telemetry_payload(payload) finally: await pubsub.unsubscribe(self.telemetry_channel) await pubsub.aclose() await client.aclose() + @staticmethod + def _parse_telemetry_payload(payload: str) -> TelemetryEvent: + """Parse a ``"{direction}:{queue_name}"`` pub/sub payload.""" + direction, _, queue_name = payload.partition(":") + return TelemetryEvent( + direction=TelemetryDirection(direction), queue_name=queue_name + ) + def close(self) -> None: """Close the Redis connection.""" self.client.close() diff --git a/threadmill/inspector/app.py b/threadmill/inspector/app.py index 998945e..b4e4627 100644 --- a/threadmill/inspector/app.py +++ b/threadmill/inspector/app.py @@ -34,17 +34,21 @@ TabPane, ) -from ..backends.base import BackendTelemetry, ThreadmillTaskBackend +from ..backends.base import ( + BackendTelemetry, + TelemetryDirection, + ThreadmillTaskBackend, +) from .screens import ConfirmScreen, PurgeScreen from .telemetry import TelemetryBuffer logger = logging.getLogger(__name__) -TELEMETRY_INTERVAL_SECONDS = 2.0 -"""Seconds between automatic queue-stat refreshes; the task list stays manual.""" +TELEMETRY_INTERVAL = datetime.timedelta(seconds=2) +"""Between automatic queue-stat refreshes; the task list stays manual.""" -SPARKLINE_INTERVAL_SECONDS = 1.0 -"""Seconds between sparkline refreshes from the rolling telemetry buffer.""" +SPARKLINE_INTERVAL = datetime.timedelta(seconds=1) +"""Between sparkline refreshes from the rolling telemetry buffer.""" TAB_STATUSES: list[tuple[str, TaskResultStatus | None]] = [ ("Running", TaskResultStatus.RUNNING), @@ -239,7 +243,7 @@ def switch_tab(self, tab_id: str) -> None: def action_refresh(self) -> None: """Refresh telemetry and re-fetch the task list.""" - self.app._refresh_telemetry() + self.app._poll_queue_stats() self.refresh_tasks() def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None: @@ -468,7 +472,6 @@ def __init__( self._task_list: TaskList | None = None self._task_detail: TaskDetail | None = None self._options_static: Static | None = None - self._telemetry_timer = None self._sparkline_timer = None self._auto_refresh = auto_refresh self.telemetry_buffer = TelemetryBuffer() @@ -516,16 +519,11 @@ def on_mount(self) -> None: self._task_detail = self.query_one("#task-detail", TaskDetail) self._options_static = self.query_one("#backend-options", Static) self._refresh_options() - self._refresh_telemetry() + self._poll_queue_stats() self._listen_telemetry() if self._auto_refresh: - self._telemetry_timer = self.set_interval( - TELEMETRY_INTERVAL_SECONDS, - self._refresh_telemetry, - name="telemetry-refresh", - ) self._sparkline_timer = self.set_interval( - SPARKLINE_INTERVAL_SECONDS, + SPARKLINE_INTERVAL.total_seconds(), self._refresh_sparklines, name="sparkline-refresh", ) @@ -538,7 +536,7 @@ def action_quit(self) -> None: def _after_action(self, message: str) -> None: """Refresh telemetry and tasks after a mutating queue action.""" self.notify(message) - self._refresh_telemetry() + self._poll_queue_stats() self._task_list.refresh_tasks() def action_switch_tab(self, tab_id: str) -> None: @@ -547,7 +545,7 @@ def action_switch_tab(self, tab_id: str) -> None: def watch_backend(self) -> None: self._refresh_options() - self._refresh_telemetry() + self._poll_queue_stats() self._listen_telemetry() def on_select_changed(self, event: Select.Changed) -> None: @@ -575,19 +573,30 @@ def _refresh_options(self) -> None: ] self._options_static.update(" ".join(parts) or "No options") - def _refresh_telemetry(self) -> None: - """Poll the backend for counts and merge live pub/sub rates for the selected queue.""" - try: - telemetry = self.backend.telemetry() - except Exception: # noqa: BLE001 - logger.exception("Failed to refresh telemetry") - return - queue_name = self._task_list.queue_name - if queue_name and queue_name in telemetry.queues: - live_rates = self.telemetry_buffer.rates_for(queue_name) - stats = telemetry.queues[queue_name] - telemetry.queues[queue_name] = dataclasses.replace(stats, rates=live_rates) - self.telemetry = telemetry + @work(exclusive=True, group="queue-stats") + async def _poll_queue_stats(self) -> None: + """Poll the backend for counts and merge live pub/sub rates. + + Runs as a textual worker so the synchronous ``queue_stats`` call is + offloaded via :func:`asyncio.to_thread`, keeping the UI thread free. + The worker loops until cancelled, sleeping ``TELEMETRY_INTERVAL`` + between refreshes. + """ + while True: + try: + telemetry = await asyncio.to_thread(self.backend.queue_stats) + except Exception: # noqa: BLE001 + logger.exception("Failed to refresh queue stats") + else: + queue_name = self._task_list.queue_name + if queue_name and queue_name in telemetry.queues: + live_rates = self.telemetry_buffer.rates_for(queue_name) + stats = telemetry.queues[queue_name] + telemetry.queues[queue_name] = dataclasses.replace( + stats, rates=live_rates + ) + self.telemetry = telemetry + await asyncio.sleep(TELEMETRY_INTERVAL.total_seconds()) def _refresh_sparklines(self) -> None: """Push the latest 60-second rolling series into the ingress/egress sparklines.""" @@ -596,12 +605,14 @@ def _refresh_sparklines(self) -> None: egress = self.query_one("#egress-sparkline", Sparkline) now = time.monotonic() ingress.data = ( - self.telemetry_buffer.series(queue_name, "ingress", now=now) + self.telemetry_buffer.series( + queue_name, TelemetryDirection.INGRESS, now=now + ) if queue_name else [0.0] * 60 ) egress.data = ( - self.telemetry_buffer.series(queue_name, "egress", now=now) + self.telemetry_buffer.series(queue_name, TelemetryDirection.EGRESS, now=now) if queue_name else [0.0] * 60 ) @@ -609,11 +620,11 @@ def _refresh_sparklines(self) -> None: @work(exclusive=True, group="telemetry") async def _listen_telemetry(self) -> None: """Subscribe to the backend's pub/sub telemetry stream and feed the buffer.""" - stream = self.backend.subscribe_telemetry() + stream = self.backend.worker_telemetry() if stream is None: return try: - async for payload in stream: - self.telemetry_buffer.record(payload, now=time.monotonic()) + async for event in stream: + self.telemetry_buffer.record(event, now=time.monotonic()) except asyncio.CancelledError: raise diff --git a/threadmill/inspector/telemetry.py b/threadmill/inspector/telemetry.py new file mode 100644 index 0000000..396a7c7 --- /dev/null +++ b/threadmill/inspector/telemetry.py @@ -0,0 +1,82 @@ +"""In-memory rolling telemetry buffer fed by backend pub/sub events.""" + +import collections +import datetime +import time + +from threadmill.backends.base import QueueRates, TelemetryDirection, TelemetryEvent + +WINDOW_SECONDS = 60 +"""Rolling window the inspector displays for ingress/egress sparklines.""" + +_DirectionBuckets = dict[int, int] +"""second-since-epoch -> count, per (queue, direction).""" + + +class TelemetryBuffer: + """Per-(queue, direction) rolling 60-second count buffer. + + Events arrive as :class:`TelemetryEvent` instances from the backend's + pub/sub stream and are bucketed by receipt time so worker clock skew + can't distort the series. + """ + + def __init__(self, *, window: int = WINDOW_SECONDS) -> None: + self._window = window + self._counts: dict[tuple[str, TelemetryDirection], _DirectionBuckets] = ( + collections.defaultdict(lambda: collections.defaultdict(int)) + ) + + def record(self, event: TelemetryEvent, *, now: float | None = None) -> None: + """Increment the current-second bucket for the event's queue+direction.""" + second = int(now if now is not None else time.monotonic()) + self._counts[event.queue_name, event.direction][second] += 1 + self._prune(second) + + def series( + self, + queue_name: str, + direction: TelemetryDirection, + *, + now: float | None = None, + ) -> list[float]: + """Return ``window`` per-second counts, oldest to newest, for the pair.""" + second = int(now if now is not None else time.monotonic()) + self._prune(second) + buckets = self._counts.get((queue_name, direction), {}) + return [ + float(buckets.get(second - offset, 0)) + for offset in reversed(range(self._window)) + ] + + def rates_for( + self, + queue_name: str, + *, + interval: datetime.timedelta = datetime.timedelta(seconds=WINDOW_SECONDS), + now: float | None = None, + ) -> QueueRates: + """Sum the last ``interval`` of events into per-second ingress/egress rates.""" + second = int(now if now is not None else time.monotonic()) + self._prune(second) + span = min(int(interval.total_seconds()), self._window) + ingress = sum( + self._counts.get((queue_name, TelemetryDirection.INGRESS), {}).get( + second - offset, 0 + ) + for offset in range(span) + ) + egress = sum( + self._counts.get((queue_name, TelemetryDirection.EGRESS), {}).get( + second - offset, 0 + ) + for offset in range(span) + ) + return QueueRates(interval=interval, ingress=ingress, egress=egress) + + def _prune(self, now_second: int) -> None: + """Drop buckets older than the window to bound memory.""" + cutoff = now_second - self._window + for buckets in self._counts.values(): + for stale in [s for s in buckets if s <= cutoff]: + del buckets[stale] From b2259ce709b4a44f6633cc10f9c6b88b7a362ac5 Mon Sep 17 00:00:00 2001 From: Johannes Maron Date: Wed, 8 Jul 2026 21:54:29 +0200 Subject: [PATCH 3/7] wip --- tests/backends/test_redis.py | 14 +++---- tests/test_inspector.py | 6 +-- threadmill/backends/redis.py | 7 ++-- threadmill/inspector/app.py | 53 ++++++++++++------------- threadmill/inspector/inspector.scss | 18 +++++++-- threadmill/inspector/telemetry.py | 60 +++++++++++++++-------------- 6 files changed, 86 insertions(+), 72 deletions(-) diff --git a/tests/backends/test_redis.py b/tests/backends/test_redis.py index b538df7..f04bd6e 100644 --- a/tests/backends/test_redis.py +++ b/tests/backends/test_redis.py @@ -224,8 +224,8 @@ def test_stale_acknowledge__is_noop(self): finally: backend.close() - def test_telemetry__empty_backend(self): - """Telemetry returns zero counts for an empty backend.""" + def test_queue_stats__empty_backend(self): + """queue_stats returns zero counts for an empty backend.""" backend = RedisTaskBackend( "telemetry_empty_test", { @@ -242,8 +242,8 @@ def test_telemetry__empty_backend(self): finally: backend.close() - def test_telemetry__counts_tasks(self): - """Telemetry reports per-status counts; rates come from pub/sub, not polling.""" + def test_queue_stats__counts_tasks(self): + """queue_stats reports per-status counts; rates come from pub/sub, not polling.""" backend = RedisTaskBackend( "telemetry_counts_test", { @@ -290,8 +290,8 @@ def test_telemetry__counts_tasks(self): finally: backend.close() - def test_telemetry__counts_successful_and_failed(self): - """Telemetry counts successful and finished results; polling rates stay zero.""" + def test_queue_stats__counts_successful_and_failed(self): + """queue_stats counts successful and finished results; polling rates stay zero.""" backend = RedisTaskBackend( "telemetry_egress_test", { @@ -340,7 +340,7 @@ def test_telemetry__counts_successful_and_failed(self): finally: backend.close() - def test_telemetry__successful_failed_evicted_by_result_ttl(self): + def test_queue_stats__successful_failed_evicted_by_result_ttl(self): """successful/failed segment counts drop when results age out of result_ttl.""" backend = RedisTaskBackend( "telemetry_eviction_test", diff --git a/tests/test_inspector.py b/tests/test_inspector.py index 847a1e4..f6d3555 100644 --- a/tests/test_inspector.py +++ b/tests/test_inspector.py @@ -245,7 +245,7 @@ async def test_list_view_selected_sets_queue(self): task_list = app.query_one("#task-list", TaskList) assert task_list.queue_name == item.queue_name - async def test_telemetry_refresh_updates_and_prunes_queues(self): + async def test_queue_stats_refresh_updates_and_prunes_queues(self): """Telemetry refresh updates existing queue labels and drops stale queues.""" app = InspectorApp(backend=default_task_backend, auto_refresh=False) async with app.run_test() as pilot: @@ -258,7 +258,7 @@ async def test_telemetry_refresh_updates_and_prunes_queues(self): await pilot.pause() assert list(queue_list._items) == ["default"] - async def test_telemetry_counts_are_scoped_to_selected_queue(self): + async def test_queue_stats_counts_are_scoped_to_selected_queue(self): """Tab counts reflect the selected queue, not backend-wide totals.""" default_task_backend.enqueue(echo, args=[1]) stats = _stats(ready=1) @@ -478,7 +478,7 @@ async def test_initial_focus_on_queue_list(self): assert app.focused is not None assert app.focused.id == "queue-list" - async def test_telemetry_refresh_does_not_re_peek_task_list(self): + async def test_queue_stats_refresh_does_not_re_peek_task_list(self): """A telemetry refresh updates counts but leaves task rows stale until manual refresh.""" default_task_backend.enqueue(echo, args=[1]) default_task_backend.enqueue(echo, args=[2]) diff --git a/threadmill/backends/redis.py b/threadmill/backends/redis.py index dd1c526..9383498 100644 --- a/threadmill/backends/redis.py +++ b/threadmill/backends/redis.py @@ -421,7 +421,7 @@ def queue_stats( """Return point-in-time counts for all configured queues. Rates are zero here; live throughput arrives via pub/sub on - :meth:`subscribe_telemetry_events` and is assembled by the inspector. + :meth:`worker_telemetry` and is assembled by the inspector. """ pipe = self.client.pipeline() for queue_name in self.queues: @@ -464,10 +464,9 @@ async def _telemetry_stream(self) -> collections.abc.AsyncIterator[TelemetryEven client = redis.asyncio.Redis.from_url(self.redis_url) pubsub = client.pubsub() await pubsub.subscribe(self.telemetry_channel) + pubsub.ignore_subscribe_messages = True try: - while message := await pubsub.get_message( - ignore_subscribe_messages=True, timeout=None - ): + async for message in pubsub.listen(): if (data := message.get("data")) is not None: payload = data.decode() if isinstance(data, bytes) else data yield self._parse_telemetry_payload(payload) diff --git a/threadmill/inspector/app.py b/threadmill/inspector/app.py index b4e4627..c31f110 100644 --- a/threadmill/inspector/app.py +++ b/threadmill/inspector/app.py @@ -473,6 +473,7 @@ def __init__( self._task_detail: TaskDetail | None = None self._options_static: Static | None = None self._sparkline_timer = None + self._telemetry_timer = None self._auto_refresh = auto_refresh self.telemetry_buffer = TelemetryBuffer() self.set_reactive(InspectorApp.backend, backend) @@ -496,14 +497,8 @@ def compose(self) -> ComposeResult: ) with Vertical(id="right-pane"): with Horizontal(id="sparkline-row"): - yield Sparkline( - id="ingress-sparkline", - max_color="ansi_green", - ) - yield Sparkline( - id="egress-sparkline", - max_color="ansi_red", - ) + yield Sparkline(id="ingress-sparkline") + yield Sparkline(id="egress-sparkline") yield TaskList(id="task-list").data_bind( backend=InspectorApp.backend, telemetry=InspectorApp.telemetry, @@ -519,9 +514,16 @@ def on_mount(self) -> None: self._task_detail = self.query_one("#task-detail", TaskDetail) self._options_static = self.query_one("#backend-options", Static) self._refresh_options() + self.query_one("#ingress-sparkline", Sparkline).border_title = "Ingress" + self.query_one("#egress-sparkline", Sparkline).border_title = "Egress" self._poll_queue_stats() self._listen_telemetry() if self._auto_refresh: + self._telemetry_timer = self.set_interval( + TELEMETRY_INTERVAL.total_seconds(), + self._poll_queue_stats, + name="telemetry-refresh", + ) self._sparkline_timer = self.set_interval( SPARKLINE_INTERVAL.total_seconds(), self._refresh_sparklines, @@ -579,24 +581,20 @@ async def _poll_queue_stats(self) -> None: Runs as a textual worker so the synchronous ``queue_stats`` call is offloaded via :func:`asyncio.to_thread`, keeping the UI thread free. - The worker loops until cancelled, sleeping ``TELEMETRY_INTERVAL`` - between refreshes. + Called periodically by a :meth:`~textual.app.App.set_interval` timer + and immediately after mutating queue actions. """ - while True: - try: - telemetry = await asyncio.to_thread(self.backend.queue_stats) - except Exception: # noqa: BLE001 - logger.exception("Failed to refresh queue stats") - else: - queue_name = self._task_list.queue_name - if queue_name and queue_name in telemetry.queues: - live_rates = self.telemetry_buffer.rates_for(queue_name) - stats = telemetry.queues[queue_name] - telemetry.queues[queue_name] = dataclasses.replace( - stats, rates=live_rates - ) - self.telemetry = telemetry - await asyncio.sleep(TELEMETRY_INTERVAL.total_seconds()) + try: + telemetry = await asyncio.to_thread(self.backend.queue_stats) + except Exception: # noqa: BLE001 + logger.exception("Failed to refresh queue stats") + return + queue_name = self._task_list.queue_name + if queue_name and queue_name in telemetry.queues: + live_rates = self.telemetry_buffer.rates_for(queue_name) + stats = telemetry.queues[queue_name] + telemetry.queues[queue_name] = dataclasses.replace(stats, rates=live_rates) + self.telemetry = telemetry def _refresh_sparklines(self) -> None: """Push the latest 60-second rolling series into the ingress/egress sparklines.""" @@ -620,7 +618,10 @@ def _refresh_sparklines(self) -> None: @work(exclusive=True, group="telemetry") async def _listen_telemetry(self) -> None: """Subscribe to the backend's pub/sub telemetry stream and feed the buffer.""" - stream = self.backend.worker_telemetry() + try: + stream = self.backend.worker_telemetry() + except NotImplementedError: + return if stream is None: return try: diff --git a/threadmill/inspector/inspector.scss b/threadmill/inspector/inspector.scss index e1256a2..330a806 100644 --- a/threadmill/inspector/inspector.scss +++ b/threadmill/inspector/inspector.scss @@ -43,16 +43,26 @@ Select { } #sparkline-row { - height: 2; + height: 5; margin-bottom: 0; } #ingress-sparkline, #egress-sparkline { width: 1fr; - height: 1; + height: 1fr; padding: 0; margin: 0; + border: round $accent 40%; + border-title-color: $primary; +} + +#ingress-sparkline > .sparkline--max-color { + color: $success; +} + +#egress-sparkline > .sparkline--max-color { + color: $error; } #backend-row { @@ -89,7 +99,7 @@ QueueList ListItem { } TaskList { - height: 60%; + height: 1fr; TabbedContent, ContentSwitcher, @@ -117,7 +127,7 @@ DataTable { } TaskDetail { - height: 40%; + height: 1fr; width: 100%; overflow: scroll; padding: 0 1; diff --git a/threadmill/inspector/telemetry.py b/threadmill/inspector/telemetry.py index 396a7c7..c2d73f5 100644 --- a/threadmill/inspector/telemetry.py +++ b/threadmill/inspector/telemetry.py @@ -9,29 +9,29 @@ WINDOW_SECONDS = 60 """Rolling window the inspector displays for ingress/egress sparklines.""" -_DirectionBuckets = dict[int, int] -"""second-since-epoch -> count, per (queue, direction).""" - class TelemetryBuffer: """Per-(queue, direction) rolling 60-second count buffer. Events arrive as :class:`TelemetryEvent` instances from the backend's pub/sub stream and are bucketed by receipt time so worker clock skew - can't distort the series. + can't distort the series. Each direction's time buckets are stored in + an :class:`~collections.OrderedDict` keyed by integer second; stale + buckets are evicted LRU-style from the front in O(1). """ def __init__(self, *, window: int = WINDOW_SECONDS) -> None: self._window = window - self._counts: dict[tuple[str, TelemetryDirection], _DirectionBuckets] = ( - collections.defaultdict(lambda: collections.defaultdict(int)) - ) + self._counts: dict[ + tuple[str, TelemetryDirection], collections.OrderedDict[int, int] + ] = collections.defaultdict(collections.OrderedDict) def record(self, event: TelemetryEvent, *, now: float | None = None) -> None: """Increment the current-second bucket for the event's queue+direction.""" second = int(now if now is not None else time.monotonic()) - self._counts[event.queue_name, event.direction][second] += 1 - self._prune(second) + buckets = self._counts[event.queue_name, event.direction] + buckets[second] = buckets.get(second, 0) + 1 + self._evict(buckets, second) def series( self, @@ -42,10 +42,11 @@ def series( ) -> list[float]: """Return ``window`` per-second counts, oldest to newest, for the pair.""" second = int(now if now is not None else time.monotonic()) - self._prune(second) - buckets = self._counts.get((queue_name, direction), {}) + buckets = self._counts.get((queue_name, direction)) + if buckets is not None: + self._evict(buckets, second) return [ - float(buckets.get(second - offset, 0)) + float(buckets.get(second - offset, 0)) if buckets else 0.0 for offset in reversed(range(self._window)) ] @@ -58,25 +59,28 @@ def rates_for( ) -> QueueRates: """Sum the last ``interval`` of events into per-second ingress/egress rates.""" second = int(now if now is not None else time.monotonic()) - self._prune(second) span = min(int(interval.total_seconds()), self._window) - ingress = sum( - self._counts.get((queue_name, TelemetryDirection.INGRESS), {}).get( - second - offset, 0 - ) - for offset in range(span) + ingress = self._sum_direction( + queue_name, TelemetryDirection.INGRESS, second, span ) - egress = sum( - self._counts.get((queue_name, TelemetryDirection.EGRESS), {}).get( - second - offset, 0 - ) - for offset in range(span) + egress = self._sum_direction( + queue_name, TelemetryDirection.EGRESS, second, span ) return QueueRates(interval=interval, ingress=ingress, egress=egress) - def _prune(self, now_second: int) -> None: - """Drop buckets older than the window to bound memory.""" + def _sum_direction( + self, queue_name: str, direction: TelemetryDirection, second: int, span: int + ) -> int: + buckets = self._counts.get((queue_name, direction)) + if buckets is None: + return 0 + self._evict(buckets, second) + return sum(buckets.get(second - offset, 0) for offset in range(span)) + + def _evict( + self, buckets: collections.OrderedDict[int, int], now_second: int + ) -> None: + """Drop buckets older than the window (LRU pop from front).""" cutoff = now_second - self._window - for buckets in self._counts.values(): - for stale in [s for s in buckets if s <= cutoff]: - del buckets[stale] + while buckets and next(iter(buckets)) <= cutoff: + buckets.popitem(last=False) From 8fc780ea2e7e984b516c8c4f634dac19671f0a84 Mon Sep 17 00:00:00 2001 From: Johannes Maron Date: Wed, 8 Jul 2026 22:06:43 +0200 Subject: [PATCH 4/7] wip --- AGENTS.md | 1 + tests/backends/test_base.py | 4 ++-- tests/backends/test_redis.py | 20 +++++++++--------- tests/test_inspector.py | 6 +++--- threadmill/backends/base.py | 12 ++++------- threadmill/backends/redis.py | 39 ++++++++++-------------------------- threadmill/inspector/app.py | 14 +++++-------- 7 files changed, 36 insertions(+), 60 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index aa12e04..f62f14c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -37,6 +37,7 @@ Run a single test by node ID, e.g. `uv run pytest tests/test_command.py::TestCom ```bash curl -sSL https://raw.githubusercontent.com/codingjoe/naming-things/refs/heads/main/README.md ``` +- Use Google-style docstrings for mkdocs. NEVER use RST directives (`:param:`, `:returns:`, `:meth:`). - NEVER format code, tests, or docs manually. MUST USE `prek run --all-files`. ## Coverage diff --git a/tests/backends/test_base.py b/tests/backends/test_base.py index b076338..950e08a 100644 --- a/tests/backends/test_base.py +++ b/tests/backends/test_base.py @@ -81,10 +81,10 @@ def test_peek__raise_not_implemented_error(self) -> None: ) ) - def test_queue_stats__raise_not_implemented_error(self) -> None: + async def test_queue_stats__raise_not_implemented_error(self) -> None: """Raise NotImplementedError for backend queue_stats API.""" with pytest.raises(NotImplementedError): - BackendDouble(alias="default", params={}).queue_stats() + await BackendDouble(alias="default", params={}).queue_stats() def test_dequeue__raise_not_implemented_error(self) -> None: """Raise NotImplementedError for backend dequeue API.""" diff --git a/tests/backends/test_redis.py b/tests/backends/test_redis.py index f04bd6e..76c2fbf 100644 --- a/tests/backends/test_redis.py +++ b/tests/backends/test_redis.py @@ -140,7 +140,7 @@ def test_acquire__sets_last_attempted_at(self): finally: backend.close() - def test_running_reaper__fails_expired_tasks(self): + async def test_running_reaper__fails_expired_tasks(self): """Running reaper creates FAILED results for tasks with expired lease.""" backend = RedisTaskBackend( "running_reaper_test", @@ -175,7 +175,7 @@ def test_running_reaper__fails_expired_tasks(self): # Reaping an expired task records a failed result. Live egress # now arrives via pub/sub, so backend.queue_stats() rates are zero. - stats = backend.queue_stats().queues["default"] + stats = (await backend.queue_stats()).queues["default"] assert stats.rates.egress == 0 assert stats.counts.failed == 1 assert stats.counts.successful == 0 @@ -224,7 +224,7 @@ def test_stale_acknowledge__is_noop(self): finally: backend.close() - def test_queue_stats__empty_backend(self): + async def test_queue_stats__empty_backend(self): """queue_stats returns zero counts for an empty backend.""" backend = RedisTaskBackend( "telemetry_empty_test", @@ -237,12 +237,12 @@ def test_queue_stats__empty_backend(self): }, ) try: - telemetry = backend.queue_stats() + telemetry = await backend.queue_stats() assert telemetry == BackendTelemetry(queues={"default": _stats()}) finally: backend.close() - def test_queue_stats__counts_tasks(self): + async def test_queue_stats__counts_tasks(self): """queue_stats reports per-status counts; rates come from pub/sub, not polling.""" backend = RedisTaskBackend( "telemetry_counts_test", @@ -282,7 +282,7 @@ def test_queue_stats__counts_tasks(self): ) ) - telemetry = backend.queue_stats() + telemetry = await backend.queue_stats() assert telemetry.queues["default"] == _stats( successful=1, failed=1, @@ -290,7 +290,7 @@ def test_queue_stats__counts_tasks(self): finally: backend.close() - def test_queue_stats__counts_successful_and_failed(self): + async def test_queue_stats__counts_successful_and_failed(self): """queue_stats counts successful and finished results; polling rates stay zero.""" backend = RedisTaskBackend( "telemetry_egress_test", @@ -331,7 +331,7 @@ def test_queue_stats__counts_successful_and_failed(self): ) ) - stats = backend.queue_stats().queues["default"] + stats = (await backend.queue_stats()).queues["default"] # Rates come from the pub/sub buffer, not from Redis polling. assert stats.rates.ingress == 0 assert stats.rates.egress == 0 @@ -340,7 +340,7 @@ def test_queue_stats__counts_successful_and_failed(self): finally: backend.close() - def test_queue_stats__successful_failed_evicted_by_result_ttl(self): + async def test_queue_stats__successful_failed_evicted_by_result_ttl(self): """successful/failed segment counts drop when results age out of result_ttl.""" backend = RedisTaskBackend( "telemetry_eviction_test", @@ -387,7 +387,7 @@ def _ack(status: TaskResultStatus) -> str: assert backend.client.zcard(successful_key) == 1 assert backend.client.zcard(failed_key) == 1 - stats = backend.queue_stats().queues["default"] + stats = (await backend.queue_stats()).queues["default"] assert stats.counts.successful == 1 assert stats.counts.failed == 1 finally: diff --git a/tests/test_inspector.py b/tests/test_inspector.py index f6d3555..0c3b7cc 100644 --- a/tests/test_inspector.py +++ b/tests/test_inspector.py @@ -45,7 +45,7 @@ def enqueue(self, task, args, kwargs): def peek(self, *args, **kwargs): raise RuntimeError("peek failed") - def queue_stats(self, *, interval=None): + async def queue_stats(self, *, interval=None): raise RuntimeError("queue_stats failed") @@ -250,7 +250,7 @@ async def test_queue_stats_refresh_updates_and_prunes_queues(self): app = InspectorApp(backend=default_task_backend, auto_refresh=False) async with app.run_test() as pilot: await pilot.pause() - app.telemetry = app.backend.queue_stats() + app.telemetry = await app.backend.queue_stats() await pilot.pause() queue_list = app.query_one("#queue-list", QueueList) assert set(queue_list._items) == set(default_task_backend.queues) @@ -495,7 +495,7 @@ async def test_queue_stats_refresh_does_not_re_peek_task_list(self): default_task_backend.acquire( timeout=datetime.timedelta(seconds=1), worker="stale-test" ) - app.telemetry = app.backend.queue_stats() + app.telemetry = await app.backend.queue_stats() await pilot.pause() await pilot.pause() assert table.row_count == before diff --git a/threadmill/backends/base.py b/threadmill/backends/base.py index 2f22a09..fbab995 100644 --- a/threadmill/backends/base.py +++ b/threadmill/backends/base.py @@ -258,20 +258,16 @@ def peek( """ raise NotImplementedError - def queue_stats( + async def queue_stats( self, *, interval: datetime.timedelta = datetime.timedelta(seconds=60) ) -> BackendTelemetry: - """Return a snapshot of counts for all configured queues. - - Args: - interval: The rolling window the caller intends to display. - """ + """Return per-queue task counts for all configured queues.""" raise NotImplementedError - def worker_telemetry( + async def worker_telemetry( self, ) -> collections.abc.AsyncIterator[TelemetryEvent]: - """Yield structured telemetry events from the backend's pub/sub stream.""" + """Yield ingress/egress telemetry events from the backend's pub/sub stream.""" raise NotImplementedError def dequeue(self, task_result: TaskResult) -> None: diff --git a/threadmill/backends/redis.py b/threadmill/backends/redis.py index 9383498..a630c57 100644 --- a/threadmill/backends/redis.py +++ b/threadmill/backends/redis.py @@ -415,15 +415,11 @@ def get_result(self, result_id: str) -> TaskResult: return self.deserialize_task_result(data) raise TaskResultDoesNotExist(f"Task result {result_id!r} does not exist.") - def queue_stats( + async def queue_stats( self, *, interval: datetime.timedelta = datetime.timedelta(seconds=60) ) -> BackendTelemetry: - """Return point-in-time counts for all configured queues. - - Rates are zero here; live throughput arrives via pub/sub on - :meth:`worker_telemetry` and is assembled by the inspector. - """ - pipe = self.client.pipeline() + client = redis.asyncio.Redis.from_url(self.redis_url) + pipe = client.pipeline() for queue_name in self.queues: pipe.zcard(self._segment_key(TaskResultStatus.READY, queue_name)) pipe.zcard(self._segment_key(TaskResultStatus.RUNNING, queue_name)) @@ -432,7 +428,8 @@ def queue_stats( ) pipe.zcard(self._segment_key(TaskResultStatus.SUCCESSFUL, queue_name)) pipe.zcard(self._segment_key(TaskResultStatus.FAILED, queue_name)) - results = pipe.execute() + results = await pipe.execute() + await client.aclose() zero_rates = QueueRates(interval=interval, ingress=0, egress=0) queues: dict[str, QueueStats] = {} for index, queue_name in enumerate(self.queues): @@ -449,18 +446,9 @@ def queue_stats( ) return BackendTelemetry(queues=queues) - def worker_telemetry( + async def worker_telemetry( self, - ) -> collections.abc.AsyncIterator[TelemetryEvent] | None: - """Subscribe to the backend's telemetry pub/sub channel. - - Yields structured :class:`TelemetryEvent` instances. The connection - is opened with the asyncio client so the inspector can consume it off - the UI thread. - """ - return self._telemetry_stream() - - async def _telemetry_stream(self) -> collections.abc.AsyncIterator[TelemetryEvent]: + ) -> collections.abc.AsyncGenerator[TelemetryEvent]: client = redis.asyncio.Redis.from_url(self.redis_url) pubsub = client.pubsub() await pubsub.subscribe(self.telemetry_channel) @@ -469,20 +457,15 @@ async def _telemetry_stream(self) -> collections.abc.AsyncIterator[TelemetryEven async for message in pubsub.listen(): if (data := message.get("data")) is not None: payload = data.decode() if isinstance(data, bytes) else data - yield self._parse_telemetry_payload(payload) + direction, _, queue_name = payload.partition(":") + yield TelemetryEvent( + direction=TelemetryDirection(direction), queue_name=queue_name + ) finally: await pubsub.unsubscribe(self.telemetry_channel) await pubsub.aclose() await client.aclose() - @staticmethod - def _parse_telemetry_payload(payload: str) -> TelemetryEvent: - """Parse a ``"{direction}:{queue_name}"`` pub/sub payload.""" - direction, _, queue_name = payload.partition(":") - return TelemetryEvent( - direction=TelemetryDirection(direction), queue_name=queue_name - ) - def close(self) -> None: """Close the Redis connection.""" self.client.close() diff --git a/threadmill/inspector/app.py b/threadmill/inspector/app.py index c31f110..7cee7e8 100644 --- a/threadmill/inspector/app.py +++ b/threadmill/inspector/app.py @@ -577,15 +577,9 @@ def _refresh_options(self) -> None: @work(exclusive=True, group="queue-stats") async def _poll_queue_stats(self) -> None: - """Poll the backend for counts and merge live pub/sub rates. - - Runs as a textual worker so the synchronous ``queue_stats`` call is - offloaded via :func:`asyncio.to_thread`, keeping the UI thread free. - Called periodically by a :meth:`~textual.app.App.set_interval` timer - and immediately after mutating queue actions. - """ + """Fetch queue counts and merge live pub/sub rates into the telemetry.""" try: - telemetry = await asyncio.to_thread(self.backend.queue_stats) + telemetry = await self.backend.queue_stats() except Exception: # noqa: BLE001 logger.exception("Failed to refresh queue stats") return @@ -617,9 +611,11 @@ def _refresh_sparklines(self) -> None: @work(exclusive=True, group="telemetry") async def _listen_telemetry(self) -> None: - """Subscribe to the backend's pub/sub telemetry stream and feed the buffer.""" + """Feed the telemetry buffer from the backend's pub/sub stream.""" try: stream = self.backend.worker_telemetry() + if asyncio.iscoroutine(stream): + stream = await stream except NotImplementedError: return if stream is None: From f5441f73a5af952e89334bec5ada0512b63e2546 Mon Sep 17 00:00:00 2001 From: Johannes Maron Date: Wed, 8 Jul 2026 22:36:23 +0200 Subject: [PATCH 5/7] wip --- threadmill/inspector/app.py | 96 ++++++++++++++++++------------------- 1 file changed, 47 insertions(+), 49 deletions(-) diff --git a/threadmill/inspector/app.py b/threadmill/inspector/app.py index 7cee7e8..f6579df 100644 --- a/threadmill/inspector/app.py +++ b/threadmill/inspector/app.py @@ -288,15 +288,14 @@ def action_requeue(self) -> None: ) def _do_requeue(self, confirmed: bool, task: TaskResult) -> None: - if not confirmed: - return - try: - self.backend.requeue(task, datetime.datetime.now(datetime.UTC)) - except Exception: # noqa: BLE001 - logger.exception("Requeue failed for task %s", task.id) - self.app.notify("Requeue failed.", severity="error") - return - self.app._after_action("Task requeued.") + if confirmed: + try: + self.backend.requeue(task, datetime.datetime.now(datetime.UTC)) + except Exception: # noqa: BLE001 + logger.exception("Requeue failed for task %s", task.id) + self.app.notify("Requeue failed.", severity="error") + else: + self.app._after_action("Task requeued.") def action_dequeue(self) -> None: """Remove the selected task from its current segment after confirmation.""" @@ -311,15 +310,14 @@ def action_dequeue(self) -> None: ) def _do_dequeue(self, confirmed: bool, task: TaskResult) -> None: - if not confirmed: - return - try: - self.backend.dequeue(task) - except Exception: # noqa: BLE001 - logger.exception("Delete failed for task %s", task.id) - self.app.notify("Delete failed.", severity="error") - return - self.app._after_action("Task deleted.") + if confirmed: + try: + self.backend.dequeue(task) + except Exception: # noqa: BLE001 + logger.exception("Delete failed for task %s", task.id) + self.app.notify("Delete failed.", severity="error") + else: + self.app._after_action("Task deleted.") def _refresh_data(self) -> None: """Fetch and display tasks for the current queue and active tab.""" @@ -426,24 +424,22 @@ def _notify_selection(self, queue_name: str) -> None: def action_purge(self) -> None: item = self.highlighted_child - if not isinstance(item, QueueItem): - return - queue_name = item.queue_name - self.app.push_screen( - PurgeScreen(queue_name), - lambda confirmed: self._do_purge(confirmed, queue_name), - ) + if isinstance(item, QueueItem): + queue_name = item.queue_name + self.app.push_screen( + PurgeScreen(queue_name), + lambda confirmed: self._do_purge(confirmed, queue_name), + ) def _do_purge(self, confirmed: bool, queue_name: str) -> None: - if not confirmed: - return - try: - self.app.backend.purge(queue_name) - except Exception: # noqa: BLE001 - logger.exception("Purge failed for %r", queue_name) - self.app.notify("Purge failed.", severity="error") - return - self.app._after_action(f"Purged queue {queue_name}.") + if confirmed: + try: + self.app.backend.purge(queue_name) + except Exception: # noqa: BLE001 + logger.exception("Purge failed for %r", queue_name) + self.app.notify("Purge failed.", severity="error") + else: + self.app._after_action(f"Purged queue {queue_name}.") class InspectorApp(App): @@ -582,13 +578,15 @@ async def _poll_queue_stats(self) -> None: telemetry = await self.backend.queue_stats() except Exception: # noqa: BLE001 logger.exception("Failed to refresh queue stats") - return - queue_name = self._task_list.queue_name - if queue_name and queue_name in telemetry.queues: - live_rates = self.telemetry_buffer.rates_for(queue_name) - stats = telemetry.queues[queue_name] - telemetry.queues[queue_name] = dataclasses.replace(stats, rates=live_rates) - self.telemetry = telemetry + else: + queue_name = self._task_list.queue_name + if queue_name and queue_name in telemetry.queues: + live_rates = self.telemetry_buffer.rates_for(queue_name) + stats = telemetry.queues[queue_name] + telemetry.queues[queue_name] = dataclasses.replace( + stats, rates=live_rates + ) + self.telemetry = telemetry def _refresh_sparklines(self) -> None: """Push the latest 60-second rolling series into the ingress/egress sparklines.""" @@ -617,11 +615,11 @@ async def _listen_telemetry(self) -> None: if asyncio.iscoroutine(stream): stream = await stream except NotImplementedError: - return - if stream is None: - return - try: - async for event in stream: - self.telemetry_buffer.record(event, now=time.monotonic()) - except asyncio.CancelledError: - raise + pass + else: + if stream is not None: + try: + async for event in stream: + self.telemetry_buffer.record(event, now=time.monotonic()) + except asyncio.CancelledError: + raise From 814cda9998afd8f0b956b6fad8f8c4c6bacbf6b5 Mon Sep 17 00:00:00 2001 From: Johannes Maron Date: Wed, 8 Jul 2026 22:38:25 +0200 Subject: [PATCH 6/7] wip --- threadmill/inspector/telemetry.py | 9 +-------- 1 file changed, 1 insertion(+), 8 deletions(-) diff --git a/threadmill/inspector/telemetry.py b/threadmill/inspector/telemetry.py index c2d73f5..b721e5e 100644 --- a/threadmill/inspector/telemetry.py +++ b/threadmill/inspector/telemetry.py @@ -11,14 +11,7 @@ class TelemetryBuffer: - """Per-(queue, direction) rolling 60-second count buffer. - - Events arrive as :class:`TelemetryEvent` instances from the backend's - pub/sub stream and are bucketed by receipt time so worker clock skew - can't distort the series. Each direction's time buckets are stored in - an :class:`~collections.OrderedDict` keyed by integer second; stale - buckets are evicted LRU-style from the front in O(1). - """ + """Per-(queue, direction) rolling 60-second count buffer.""" def __init__(self, *, window: int = WINDOW_SECONDS) -> None: self._window = window From be442693365c4deec7f91d1fab614262058a08b2 Mon Sep 17 00:00:00 2001 From: Johannes Maron Date: Wed, 8 Jul 2026 23:38:40 +0200 Subject: [PATCH 7/7] wip --- tests/conftest.py | 4 +++- threadmill/backends/redis.py | 28 ++++++++++++++++++++-------- threadmill/inspector/app.py | 5 ++--- threadmill/inspector/telemetry.py | 4 +++- 4 files changed, 28 insertions(+), 13 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index c6a49eb..698c782 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -11,7 +11,9 @@ def _flush_keys(backend) -> None: @pytest.fixture(autouse=True) def flush_default_backend(): - """Flush all threadmill keys before and after each test.""" + """Flush all threadmill keys and reset async client before and after each test.""" _flush_keys(default_task_backend) + default_task_backend._async_client = None yield _flush_keys(default_task_backend) + default_task_backend._async_client = None diff --git a/threadmill/backends/redis.py b/threadmill/backends/redis.py index a630c57..f67989b 100644 --- a/threadmill/backends/redis.py +++ b/threadmill/backends/redis.py @@ -148,6 +148,7 @@ def __init__(self, alias: str, params: dict) -> None: f"REDIS_URL must be specified in your settings for the {type(self).__name__}." ) from e self.client = redis.from_url(redis_url) + self._async_client: redis.asyncio.Redis | None = None self.redis_url = redis_url self.key_prefix = f"threadmill:{{{alias}}}" self.telemetry_channel = self.TELEMETRY_CHANNEL.format(prefix=self.key_prefix) @@ -157,6 +158,13 @@ def __init__(self, alias: str, params: dict) -> None: self._acquire_script = self.client.register_script(self.ACQUIRE_SCRIPT) self._acknowledge_script = self.client.register_script(self.ACKNOWLEDGE_SCRIPT) + @property + def async_client(self) -> redis.asyncio.Redis: + """Lazily-created async Redis client, reused across calls.""" + if self._async_client is None: + self._async_client = redis.asyncio.Redis.from_url(self.redis_url) + return self._async_client + def _compute_score(self, priority: int, enqueued_at: datetime.datetime) -> float: """Compute a ZSET score for priority-ordered FIFO queueing. @@ -418,7 +426,7 @@ def get_result(self, result_id: str) -> TaskResult: async def queue_stats( self, *, interval: datetime.timedelta = datetime.timedelta(seconds=60) ) -> BackendTelemetry: - client = redis.asyncio.Redis.from_url(self.redis_url) + client = self.async_client pipe = client.pipeline() for queue_name in self.queues: pipe.zcard(self._segment_key(TaskResultStatus.READY, queue_name)) @@ -429,7 +437,6 @@ async def queue_stats( pipe.zcard(self._segment_key(TaskResultStatus.SUCCESSFUL, queue_name)) pipe.zcard(self._segment_key(TaskResultStatus.FAILED, queue_name)) results = await pipe.execute() - await client.aclose() zero_rates = QueueRates(interval=interval, ingress=0, egress=0) queues: dict[str, QueueStats] = {} for index, queue_name in enumerate(self.queues): @@ -449,7 +456,7 @@ async def queue_stats( async def worker_telemetry( self, ) -> collections.abc.AsyncGenerator[TelemetryEvent]: - client = redis.asyncio.Redis.from_url(self.redis_url) + client = self.async_client pubsub = client.pubsub() await pubsub.subscribe(self.telemetry_channel) pubsub.ignore_subscribe_messages = True @@ -458,14 +465,19 @@ async def worker_telemetry( if (data := message.get("data")) is not None: payload = data.decode() if isinstance(data, bytes) else data direction, _, queue_name = payload.partition(":") - yield TelemetryEvent( - direction=TelemetryDirection(direction), queue_name=queue_name - ) + try: + event = TelemetryEvent( + direction=TelemetryDirection(direction), + queue_name=queue_name, + ) + except ValueError: + continue + yield event finally: await pubsub.unsubscribe(self.telemetry_channel) await pubsub.aclose() - await client.aclose() def close(self) -> None: - """Close the Redis connection.""" + """Close the Redis connections.""" self.client.close() + self._async_client = None diff --git a/threadmill/inspector/app.py b/threadmill/inspector/app.py index f6579df..4a7070a 100644 --- a/threadmill/inspector/app.py +++ b/threadmill/inspector/app.py @@ -543,6 +543,7 @@ def action_switch_tab(self, tab_id: str) -> None: def watch_backend(self) -> None: self._refresh_options() + self.telemetry_buffer = TelemetryBuffer() self._poll_queue_stats() self._listen_telemetry() @@ -579,10 +580,8 @@ async def _poll_queue_stats(self) -> None: except Exception: # noqa: BLE001 logger.exception("Failed to refresh queue stats") else: - queue_name = self._task_list.queue_name - if queue_name and queue_name in telemetry.queues: + for queue_name, stats in telemetry.queues.items(): live_rates = self.telemetry_buffer.rates_for(queue_name) - stats = telemetry.queues[queue_name] telemetry.queues[queue_name] = dataclasses.replace( stats, rates=live_rates ) diff --git a/threadmill/inspector/telemetry.py b/threadmill/inspector/telemetry.py index b721e5e..efe0ddb 100644 --- a/threadmill/inspector/telemetry.py +++ b/threadmill/inspector/telemetry.py @@ -59,7 +59,9 @@ def rates_for( egress = self._sum_direction( queue_name, TelemetryDirection.EGRESS, second, span ) - return QueueRates(interval=interval, ingress=ingress, egress=egress) + return QueueRates( + interval=datetime.timedelta(seconds=span), ingress=ingress, egress=egress + ) def _sum_direction( self, queue_name: str, direction: TelemetryDirection, second: int, span: int