From c84023b89f5e54eae904952e469c7b4fb21639ab Mon Sep 17 00:00:00 2001 From: Johannes Maron Date: Sun, 1 Nov 2026 22:25:37 +0100 Subject: [PATCH 1/5] Report zero worker counts on shutdown --- tests/test_executor.py | 17 ++++++++++++++++ threadmill/executor.py | 46 ++++++++++++++++++++++++++++++------------ 2 files changed, 50 insertions(+), 13 deletions(-) diff --git a/tests/test_executor.py b/tests/test_executor.py index d4a42d7..d48df46 100644 --- a/tests/test_executor.py +++ b/tests/test_executor.py @@ -834,6 +834,23 @@ def test_report_metrics__zero_interval_disables_sampling(self, monkeypatch): assert sentry_metrics.gauges == [] + def test_shutdown__reports_zero_worker_counts(self, monkeypatch): + """Report zero workers when the pool shuts down.""" + executor = TaskExecutor( + backend=StubQueueStatsBackend(BackendTelemetry(queues={})), + queues=("default",), + ) + sentry_metrics = StubSentryMetrics() + monkeypatch.setattr(executor_module, "sentry_metrics", sentry_metrics) + executor.pool_maintainer = threading.current_thread() + + executor.shutdown() + + assert sentry_metrics.gauges == [ + ("threadmill.worker.count", 0, {"pool": "process"}), + ("threadmill.worker.count", 0, {"pool": "thread"}), + ] + class TestWorkerProcess: """Tests for the WorkerProcess class.""" diff --git a/threadmill/executor.py b/threadmill/executor.py index 53472a9..415979f 100644 --- a/threadmill/executor.py +++ b/threadmill/executor.py @@ -140,6 +140,9 @@ class TaskExecutor: pool_maintainer: threading.Thread | None = dataclasses.field( default=None, init=False ) + metrics_reporter: threading.Thread | None = dataclasses.field( + default=None, init=False + ) worker_processes: list[WorkerProcess] = dataclasses.field( default_factory=list, init=False ) @@ -191,11 +194,10 @@ def run(self) -> None: ) threads = [self.pool_maintainer] if sentry_metrics is not None: - threads.append( - threading.Thread( - target=self.report_metrics, name="threadmill-metrics", daemon=True - ) + self.metrics_reporter = threading.Thread( + target=self.report_metrics, name="threadmill-metrics", daemon=True ) + threads.append(self.metrics_reporter) if self.backend.broker_class: self.broker = self.backend.broker_class(self.backend) threads.append(self.broker) @@ -209,6 +211,18 @@ def shutdown(self) -> None: """Stop pool maintenance first, then queue consumption and all workers.""" logger.info("Shutting down task executor") self.shutdown_requested.set() + if ( + self.metrics_reporter is not None + and self.metrics_reporter.is_alive() + and self.metrics_reporter is not threading.current_thread() + ): + self.metrics_reporter.join() + if ( + sentry_metrics is not None + and self.sample_interval > datetime.timedelta(0) + and self.pool_maintainer is not None + ): + self.report_worker_counts(0, 0) if ( self.pool_maintainer is not None and self.pool_maintainer.is_alive() @@ -244,19 +258,25 @@ def maintain_worker_pool(self) -> None: return self.shutdown_requested.wait(1) + def report_worker_counts(self, process_count: int, thread_count: int) -> None: + """Record the worker process and thread counts.""" + sentry_metrics.gauge( + "threadmill.worker.count", + process_count, + attributes={"pool": "process"}, + ) + sentry_metrics.gauge( + "threadmill.worker.count", + thread_count, + attributes={"pool": "thread"}, + ) + def report_metrics(self) -> None: """Sample queue depth and worker counts into Sentry until shutdown.""" if self.sample_interval > datetime.timedelta(0): while not self.shutdown_requested.is_set(): - sentry_metrics.gauge( - "threadmill.worker.count", - self.process_count, - attributes={"pool": "process"}, - ) - sentry_metrics.gauge( - "threadmill.worker.count", - self.process_count * self.thread_count, - attributes={"pool": "thread"}, + self.report_worker_counts( + self.process_count, self.process_count * self.thread_count ) telemetry = asyncio.run(self.backend.queue_stats()) for queue_name, stats in telemetry.queues.items(): From c2f8a6c12a2df335e74401eafbc810fb85bc8cf5 Mon Sep 17 00:00:00 2001 From: Johannes Maron Date: Sun, 1 Nov 2026 22:45:40 +0100 Subject: [PATCH 2/5] Report worker counts on pool events --- README.md | 3 +- tests/test_executor.py | 46 +++++++++++--- threadmill/executor.py | 64 ++++++++++---------- threadmill/management/commands/threadmill.py | 2 +- 4 files changed, 75 insertions(+), 40 deletions(-) diff --git a/README.md b/README.md index 427854b..e151c98 100644 --- a/README.md +++ b/README.md @@ -145,9 +145,10 @@ uv add "threadmill[sentry]" | `threadmill.queue.depth` | gauge | `queue`, `status` (ready, running, deferred) | Tasks in a queue with status ready, running, or deferred | | `threadmill.queue.ingress` | counter | `queue` | Tasks that enter a queue | | `threadmill.queue.egress` | counter | `queue` | Tasks that finish an attempt | -| `threadmill.worker.count` | gauge | `pool` (process, thread) | Worker processes and threads in the pool | +| `threadmill.worker.count` | gauge | `pool` (process, thread) | Live worker processes and threads in the pool | | `threadmill.task.duration` | distribution | `task`, `outcome` (successful, failed), `attempt` | Duration of one task attempt in milliseconds | +Worker counts update when the pool starts, replaces a process, or shuts down. A task duration is recorded in the worker that runs the attempt. The SDK drops all metrics when it is not initialized. diff --git a/tests/test_executor.py b/tests/test_executor.py index d48df46..83936fd 100644 --- a/tests/test_executor.py +++ b/tests/test_executor.py @@ -422,6 +422,17 @@ def gauge(self, name, value, *, attributes=None) -> None: self.gauges.append((name, value, attributes)) +class StubWorker: + """Worker process double that reports liveness.""" + + def __init__(self, *, alive: bool) -> None: + self.alive = alive + + def is_alive(self) -> bool: + """Return the configured liveness.""" + return self.alive + + class StubShutdownEvent: """Report a shutdown request after the first loop iteration.""" @@ -779,8 +790,8 @@ def start_after_failed_attempt(process): assert executor.worker_processes[0].is_alive() executor.worker_processes[0].shutdown() - def test_report_metrics__samples_pool_and_queue_counts(self, monkeypatch): - """Emit pool and per-queue counts once per sample interval.""" + def test_report_queue_depth__reports_counts(self, monkeypatch): + """Emit per-queue counts once per sample interval.""" backend = StubQueueStatsBackend( BackendTelemetry( queues={ @@ -809,18 +820,16 @@ def test_report_metrics__samples_pool_and_queue_counts(self, monkeypatch): monkeypatch.setattr(executor_module, "sentry_metrics", sentry_metrics) executor.shutdown_requested = shutdown_requested - executor.report_metrics() + executor.report_queue_depth() assert sentry_metrics.gauges == [ - ("threadmill.worker.count", 2, {"pool": "process"}), - ("threadmill.worker.count", 6, {"pool": "thread"}), ("threadmill.queue.depth", 5, {"queue": "default", "status": "ready"}), ("threadmill.queue.depth", 2, {"queue": "default", "status": "running"}), ("threadmill.queue.depth", 1, {"queue": "default", "status": "deferred"}), ] assert shutdown_requested.timeouts == [7.0] - def test_report_metrics__zero_interval_disables_sampling(self, monkeypatch): + def test_report_queue_depth__zero_interval_disables_sampling(self, monkeypatch): """Skip sampling when the interval is zero.""" executor = TaskExecutor( backend=StubQueueStatsBackend(BackendTelemetry(queues={})), @@ -830,10 +839,33 @@ def test_report_metrics__zero_interval_disables_sampling(self, monkeypatch): sentry_metrics = StubSentryMetrics() monkeypatch.setattr(executor_module, "sentry_metrics", sentry_metrics) - executor.report_metrics() + executor.report_queue_depth() assert sentry_metrics.gauges == [] + def test_report_worker_counts__reports_live_processes(self, monkeypatch): + """Report only live worker processes and their threads.""" + executor = TaskExecutor( + backend=StubQueueStatsBackend(BackendTelemetry(queues={})), + workers=3, + threads=2, + queues=("default",), + ) + sentry_metrics = StubSentryMetrics() + monkeypatch.setattr(executor_module, "sentry_metrics", sentry_metrics) + executor.worker_processes = [ + StubWorker(alive=True), + StubWorker(alive=False), + StubWorker(alive=True), + ] + + executor.report_worker_counts() + + assert sentry_metrics.gauges == [ + ("threadmill.worker.count", 2, {"pool": "process"}), + ("threadmill.worker.count", 4, {"pool": "thread"}), + ] + def test_shutdown__reports_zero_worker_counts(self, monkeypatch): """Report zero workers when the pool shuts down.""" executor = TaskExecutor( diff --git a/threadmill/executor.py b/threadmill/executor.py index 415979f..72ab3a9 100644 --- a/threadmill/executor.py +++ b/threadmill/executor.py @@ -140,7 +140,7 @@ class TaskExecutor: pool_maintainer: threading.Thread | None = dataclasses.field( default=None, init=False ) - metrics_reporter: threading.Thread | None = dataclasses.field( + queue_depth_reporter: threading.Thread | None = dataclasses.field( default=None, init=False ) worker_processes: list[WorkerProcess] = dataclasses.field( @@ -189,15 +189,18 @@ def run(self) -> None: self.worker_processes = [ self.create_worker_process() for _ in range(self.process_count) ] + self.report_worker_counts() self.pool_maintainer = threading.Thread( target=self.maintain_worker_pool, daemon=True ) threads = [self.pool_maintainer] if sentry_metrics is not None: - self.metrics_reporter = threading.Thread( - target=self.report_metrics, name="threadmill-metrics", daemon=True + self.queue_depth_reporter = threading.Thread( + target=self.report_queue_depth, + name="threadmill-queue-depth", + daemon=True, ) - threads.append(self.metrics_reporter) + threads.append(self.queue_depth_reporter) if self.backend.broker_class: self.broker = self.backend.broker_class(self.backend) threads.append(self.broker) @@ -212,17 +215,11 @@ def shutdown(self) -> None: logger.info("Shutting down task executor") self.shutdown_requested.set() if ( - self.metrics_reporter is not None - and self.metrics_reporter.is_alive() - and self.metrics_reporter is not threading.current_thread() + self.queue_depth_reporter is not None + and self.queue_depth_reporter.is_alive() + and self.queue_depth_reporter is not threading.current_thread() ): - self.metrics_reporter.join() - if ( - sentry_metrics is not None - and self.sample_interval > datetime.timedelta(0) - and self.pool_maintainer is not None - ): - self.report_worker_counts(0, 0) + self.queue_depth_reporter.join() if ( self.pool_maintainer is not None and self.pool_maintainer.is_alive() @@ -234,11 +231,14 @@ def shutdown(self) -> None: self.broker.shutdown() with ThreadPoolExecutor(max_workers=self.process_count) as executor: executor.map(lambda worker: worker.shutdown(), self.worker_processes) + if self.pool_maintainer is not None: + self.report_worker_counts() def maintain_worker_pool(self) -> None: """Restart worker processes that have exited, until shutdown is requested.""" while not self.shutdown_requested.is_set(): all_dead = True + replaced = False for index, worker in enumerate(self.worker_processes): if worker.is_alive(): all_dead = False @@ -253,31 +253,33 @@ def maintain_worker_pool(self) -> None: logger.warning( "Unable to start worker-%d. Retrying ...", index, exc_info=error ) + replaced = True + if replaced: + self.report_worker_counts() if all_dead and self.exit_empty: self.shutdown() return self.shutdown_requested.wait(1) - def report_worker_counts(self, process_count: int, thread_count: int) -> None: - """Record the worker process and thread counts.""" - sentry_metrics.gauge( - "threadmill.worker.count", - process_count, - attributes={"pool": "process"}, - ) - sentry_metrics.gauge( - "threadmill.worker.count", - thread_count, - attributes={"pool": "thread"}, - ) + def report_worker_counts(self) -> None: + """Record the live worker process and thread counts.""" + if sentry_metrics is not None: + process_count = sum(worker.is_alive() for worker in self.worker_processes) + sentry_metrics.gauge( + "threadmill.worker.count", + process_count, + attributes={"pool": "process"}, + ) + sentry_metrics.gauge( + "threadmill.worker.count", + process_count * self.thread_count, + attributes={"pool": "thread"}, + ) - def report_metrics(self) -> None: - """Sample queue depth and worker counts into Sentry until shutdown.""" + def report_queue_depth(self) -> None: + """Sample queue depth into Sentry until shutdown.""" if self.sample_interval > datetime.timedelta(0): while not self.shutdown_requested.is_set(): - self.report_worker_counts( - self.process_count, self.process_count * self.thread_count - ) telemetry = asyncio.run(self.backend.queue_stats()) for queue_name, stats in telemetry.queues.items(): for status, depth in ( diff --git a/threadmill/management/commands/threadmill.py b/threadmill/management/commands/threadmill.py index fc6803b..2432fbb 100644 --- a/threadmill/management/commands/threadmill.py +++ b/threadmill/management/commands/threadmill.py @@ -113,7 +113,7 @@ def add_arguments(self, parser): type=self.parse_duration, default=datetime.timedelta(seconds=30), help=( - "Interval between queue depth and worker count samples in seconds." + "Interval between queue depth samples in seconds." " Defaults to 30. Set to 0 to disable sampling." ), ) From a45e1020740bce96e6850cb101b8b639298f28aa Mon Sep 17 00:00:00 2001 From: Johannes Maron Date: Sun, 1 Nov 2026 22:54:08 +0100 Subject: [PATCH 3/5] Drop queue depth reporter reference --- threadmill/executor.py | 20 ++++++-------------- 1 file changed, 6 insertions(+), 14 deletions(-) diff --git a/threadmill/executor.py b/threadmill/executor.py index 72ab3a9..ebdf969 100644 --- a/threadmill/executor.py +++ b/threadmill/executor.py @@ -140,9 +140,6 @@ class TaskExecutor: pool_maintainer: threading.Thread | None = dataclasses.field( default=None, init=False ) - queue_depth_reporter: threading.Thread | None = dataclasses.field( - default=None, init=False - ) worker_processes: list[WorkerProcess] = dataclasses.field( default_factory=list, init=False ) @@ -195,12 +192,13 @@ def run(self) -> None: ) threads = [self.pool_maintainer] if sentry_metrics is not None: - self.queue_depth_reporter = threading.Thread( - target=self.report_queue_depth, - name="threadmill-queue-depth", - daemon=True, + threads.append( + threading.Thread( + target=self.report_queue_depth, + name="threadmill-queue-depth", + daemon=True, + ) ) - threads.append(self.queue_depth_reporter) if self.backend.broker_class: self.broker = self.backend.broker_class(self.backend) threads.append(self.broker) @@ -214,12 +212,6 @@ def shutdown(self) -> None: """Stop pool maintenance first, then queue consumption and all workers.""" logger.info("Shutting down task executor") self.shutdown_requested.set() - if ( - self.queue_depth_reporter is not None - and self.queue_depth_reporter.is_alive() - and self.queue_depth_reporter is not threading.current_thread() - ): - self.queue_depth_reporter.join() if ( self.pool_maintainer is not None and self.pool_maintainer.is_alive() From bfcc68301518c83dc4f5a5faa48cbed932d09e2a Mon Sep 17 00:00:00 2001 From: Johannes Maron Date: Sun, 1 Nov 2026 23:00:29 +0100 Subject: [PATCH 4/5] Trim Sentry metrics prose --- README.md | 4 ---- 1 file changed, 4 deletions(-) diff --git a/README.md b/README.md index e151c98..82ceeb6 100644 --- a/README.md +++ b/README.md @@ -148,10 +148,6 @@ uv add "threadmill[sentry]" | `threadmill.worker.count` | gauge | `pool` (process, thread) | Live worker processes and threads in the pool | | `threadmill.task.duration` | distribution | `task`, `outcome` (successful, failed), `attempt` | Duration of one task attempt in milliseconds | -Worker counts update when the pool starts, replaces a process, or shuts down. -A task duration is recorded in the worker that runs the attempt. -The SDK drops all metrics when it is not initialized. - ### Autoscaling Threadmill exposes queue stats as JSON for KEDA's `metrics-api` scaler. From 2d06bca0c11f7a99363cddbbec11a3a765129317 Mon Sep 17 00:00:00 2001 From: Johannes Maron Date: Sun, 1 Nov 2026 23:02:02 +0100 Subject: [PATCH 5/5] Clarify queue depth sampling help --- threadmill/executor.py | 2 +- threadmill/management/commands/threadmill.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/threadmill/executor.py b/threadmill/executor.py index ebdf969..9673fb9 100644 --- a/threadmill/executor.py +++ b/threadmill/executor.py @@ -186,10 +186,10 @@ def run(self) -> None: self.worker_processes = [ self.create_worker_process() for _ in range(self.process_count) ] - self.report_worker_counts() self.pool_maintainer = threading.Thread( target=self.maintain_worker_pool, daemon=True ) + self.report_worker_counts() threads = [self.pool_maintainer] if sentry_metrics is not None: threads.append( diff --git a/threadmill/management/commands/threadmill.py b/threadmill/management/commands/threadmill.py index 2432fbb..ecb4e5d 100644 --- a/threadmill/management/commands/threadmill.py +++ b/threadmill/management/commands/threadmill.py @@ -114,7 +114,7 @@ def add_arguments(self, parser): default=datetime.timedelta(seconds=30), help=( "Interval between queue depth samples in seconds." - " Defaults to 30. Set to 0 to disable sampling." + " Defaults to 30. Set to 0 to disable queue depth sampling." ), ) parser.add_argument(