diff --git a/CHANGELOG.md b/CHANGELOG.md index 2cb79c55..db478dc2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,36 @@ # Changelog +## Version 0.26.0 + +### ⚠️ Breaking changes + +#### Worker stats reworked into configurable consumers ([377](https://github.com/closeio/tasktiger/pull/377)) + +* The worker no longer logs the periodic `stats` event by default, and the + `STATS_INTERVAL` config option was removed (a leftover value is silently + ignored). To keep the log line: + + ```python + from tasktiger.stats import StatsThread + + tiger = TaskTiger(connection=conn, config={ + "STATS_CONSUMERS": [StatsThread(log, interval=60)], + }) + ``` +* `StatsThread` takes a logger and an interval instead of the worker, and the + worker no longer instantiates `worker.StatsThread` — monkey-patching that + name silently stops working. Subclass `tasktiger.stats.StatsConsumer` and + register instances via `STATS_CONSUMERS` instead. +* `Worker.stats_thread` was replaced by `Worker.stats`, the list of active + consumers. + +### Other changes + +* The `stats` event now includes `time_idle` (blocking waits for new work), + `time_overhead` (everything else), and `occupancy` + (`100 * time_busy / (time_busy + time_idle)`), better suited than + `utilization` for load/autoscaling decisions. + ## Version 0.25.0 * Added `Task.scheduled_at` property signifying when the task is/was supposed to run. diff --git a/tasktiger/stats.py b/tasktiger/stats.py index 0f455f26..606bc75a 100644 --- a/tasktiger/stats.py +++ b/tasktiger/stats.py @@ -1,71 +1,134 @@ import threading import time -from typing import TYPE_CHECKING, Optional +from typing import Any, Optional from ._internal import g_fork_lock -if TYPE_CHECKING: - from .worker import Worker +class StatsConsumer: + """Receives worker measurements; configure via STATS_CONSUMERS. -class StatsThread(threading.Thread): - def __init__(self, tiger: "Worker") -> None: + The worker calls start()/stop() around its run loop and the paired + on_*_start()/on_*_end() hooks around each task and idle wait. Hooks + must not raise — that disrupts task processing. One instance is reused + across every worker and run of a TaskTiger. + """ + + def start(self) -> None: + pass + + def stop(self) -> None: + pass + + def on_task_start(self) -> None: + pass + + def on_task_end(self) -> None: + pass + + def on_idle_start(self) -> None: + pass + + def on_idle_end(self) -> None: + pass + + +class StatsThread(threading.Thread, StatsConsumer): + """Periodically logs a "stats" event with worker time accounting. + + Construct with a logger and interval (seconds) and add to + STATS_CONSUMERS; the worker starts and stops it. As a thread it can + only start once — use a fresh instance per worker run. + """ + + def __init__(self, log: Any, interval: float) -> None: super(StatsThread, self).__init__() - self.tiger = tiger + self.log = log + self.interval = interval self._stop_event = threading.Event() self._task_running = False self._time_start = time.monotonic() self._time_busy: float = 0.0 self._task_start_time: Optional[float] = None + self._time_idle: float = 0.0 + self._idle_start_time: Optional[float] = None self.daemon = True # Exit process if main thread exits unexpectedly - # Lock that protects stats computations from interleaving. For example, - # we don't want report_task_start() to run at the same time as - # compute_stats(), as it might result in an inconsistent state. + # Serializes stats computations: the on_*() hooks must not + # interleave with compute_stats(), or state goes inconsistent. + # Timestamps are read under the lock so a span can't straddle a + # window boundary. self._computation_lock = threading.Lock() - def report_task_start(self) -> None: - now = time.monotonic() + def on_task_start(self) -> None: with self._computation_lock: - self._task_start_time = now + assert self._task_start_time is None + self._task_start_time = time.monotonic() self._task_running = True - def report_task_end(self) -> None: - now = time.monotonic() + def on_task_end(self) -> None: with self._computation_lock: assert self._task_start_time is not None - self._time_busy += now - self._task_start_time + self._time_busy += time.monotonic() - self._task_start_time self._task_running = False self._task_start_time = None - def compute_stats(self) -> None: - now = time.monotonic() + def on_idle_start(self) -> None: + with self._computation_lock: + assert self._idle_start_time is None + self._idle_start_time = time.monotonic() + + def on_idle_end(self) -> None: + with self._computation_lock: + assert self._idle_start_time is not None + self._time_idle += time.monotonic() - self._idle_start_time + self._idle_start_time = None + def compute_stats(self) -> None: with self._computation_lock: + now = time.monotonic() time_total = now - self._time_start time_busy = self._time_busy + time_idle = self._time_idle self._time_start = now self._time_busy = 0 + self._time_idle = 0 if self._task_running: assert self._task_start_time is not None time_busy += now - self._task_start_time self._task_start_time = now else: self._task_start_time = None + if self._idle_start_time is not None: + time_idle += now - self._idle_start_time + self._idle_start_time = now if time_total: + # busy: in task code. idle: blocking waits for work. + # overhead: the rest (dequeue, scan, locks, maintenance). + time_overhead = time_total - time_busy - time_idle + # occupancy = busy share of busy + idle, ignoring overhead; + # unlike utilization it stays ~100 when saturated with short + # tasks. 0 with no tasks. utilization = 100.0 / time_total * time_busy + time_attributable = time_busy + time_idle + occupancy = ( + 100.0 * time_busy / time_attributable if time_attributable else 0.0 + ) with g_fork_lock: - self.tiger.log.info( + self.log.info( "stats", time_total=time_total, time_busy=time_busy, + time_idle=time_idle, + time_overhead=time_overhead, utilization=utilization, + occupancy=occupancy, ) def run(self) -> None: - while not self._stop_event.wait(self.tiger.config["STATS_INTERVAL"]): + while not self._stop_event.wait(self.interval): self.compute_stats() def stop(self) -> None: diff --git a/tasktiger/tasktiger.py b/tasktiger/tasktiger.py index aeb611b9..b194ad58 100644 --- a/tasktiger/tasktiger.py +++ b/tasktiger/tasktiger.py @@ -191,8 +191,9 @@ def init( # subqueues will be automatically treated as batch queues, and the # batch value of the most specific subqueue name takes precedence. "BATCH_QUEUES": {}, - # How often to print stats. - "STATS_INTERVAL": 60, + # StatsConsumer instances receiving worker task/idle + # measurements, shared across all workers of this TaskTiger. + "STATS_CONSUMERS": [], # Single worker queues can reduce redis activity in some use cases # by locking at the queue level instead of just at the task or task # group level. These queues will only allow a single worker to diff --git a/tasktiger/worker.py b/tasktiger/worker.py index 38d4fd1c..d6f4b0f8 100644 --- a/tasktiger/worker.py +++ b/tasktiger/worker.py @@ -8,10 +8,12 @@ import time import uuid from collections import OrderedDict +from contextlib import ExitStack, contextmanager from typing import ( TYPE_CHECKING, Any, Collection, + Iterator, Dict, List, Literal, @@ -41,7 +43,7 @@ from .executor import Executor, ForkExecutor from .redis_semaphore import Semaphore from .runner import get_runner_class -from .stats import StatsThread +from .stats import StatsConsumer from .task import Task from .timeouts import JobTimeoutException from .utils import redis_glob_escape @@ -80,7 +82,7 @@ def __init__( self._key = tiger._key self._did_work = True self._last_task_check = 0.0 - self.stats_thread: Optional[StatsThread] = None + self.stats: List[StatsConsumer] = [] self.id = str(uuid.uuid4()) if executor_class is None: @@ -214,6 +216,22 @@ def _worker_queue_scheduled_tasks(self) -> None: self.connection.publish(self._key("activity"), queue) self._did_work = True + @contextmanager + def _measure_idle(self) -> Iterator[None]: + with ExitStack() as stack: + for consumer in self.stats: + consumer.on_idle_start() + stack.callback(consumer.on_idle_end) + yield + + @contextmanager + def _measure_task(self) -> Iterator[None]: + with ExitStack() as stack: + for consumer in self.stats: + consumer.on_task_start() + stack.callback(consumer.on_task_end) + yield + def _poll_for_queues(self) -> None: """ Refresh list of queues. @@ -223,7 +241,8 @@ def _poll_for_queues(self) -> None: This is only used when using polling to get queues with queued tasks. """ if not self._did_work: - time.sleep(self.config["POLL_TASK_QUEUES_INTERVAL"]) + with self._measure_idle(): + time.sleep(self.config["POLL_TASK_QUEUES_INTERVAL"]) self._refresh_queue_set() def _pubsub_for_queues(self, timeout: float = 0, batch_timeout: float = 0) -> None: @@ -250,9 +269,10 @@ def _pubsub_for_queues(self, timeout: float = 0, batch_timeout: float = 0) -> No pubsub_sleep = batch_exit - time.time() else: pubsub_sleep = start_time + timeout - time.time() - message = self._pubsub.get_message( - timeout=0 if pubsub_sleep < 0 or self._did_work else pubsub_sleep - ) + with self._measure_idle(): + message = self._pubsub.get_message( + timeout=0 if pubsub_sleep < 0 or self._did_work else pubsub_sleep + ) # Pull remaining messages off of channel while message: @@ -670,14 +690,10 @@ def _execute_task_group( if not ready_tasks: return True, [] - if self.stats_thread: - self.stats_thread.report_task_start() + with self._measure_task(): + self._prepare_execution(ready_tasks) - self._prepare_execution(ready_tasks) - - success = self.executor.execute(queue, ready_tasks, log, locks, queue_lock) - if self.stats_thread: - self.stats_thread.report_task_end() + success = self.executor.execute(queue, ready_tasks, log, locks, queue_lock) for lock in locks: try: @@ -974,11 +990,6 @@ def run( # executing pipelines. self.log.warning("using old Redis version") - if self.config["STATS_INTERVAL"]: - stats_thread = StatsThread(self) - self.stats_thread = stats_thread - stats_thread.start() - # Queue any periodic tasks that are not queued yet. self._queue_periodic_tasks() @@ -995,7 +1006,11 @@ def run( self._refresh_queue_set() + self.stats = list(self.config["STATS_CONSUMERS"]) try: + for consumer in self.stats: + consumer.start() + while True: # Update the queue set on every iteration so we don't get stuck # on processing a specific queue. @@ -1028,9 +1043,9 @@ def run( raise finally: - if self.stats_thread: - self.stats_thread.stop() - self.stats_thread = None + for consumer in self.stats: + consumer.stop() + self.stats = [] # Free up Redis connection if self._pubsub: diff --git a/tests/test_stats.py b/tests/test_stats.py index 41cbcc2c..60a0c35f 100644 --- a/tests/test_stats.py +++ b/tests/test_stats.py @@ -1,22 +1,11 @@ import time from unittest import mock -import pytest +from tasktiger.stats import StatsConsumer, StatsThread -from tasktiger.stats import StatsThread -from tests.utils import get_tiger - - -@pytest.fixture -def tiger(): - t = get_tiger() - t.config["STATS_INTERVAL"] = 0.07 - return t - - -def test_start_and_stop(tiger): - stats = StatsThread(tiger) +def test_start_and_stop(): + stats = StatsThread(mock.Mock(), interval=0.07) stats.compute_stats = mock.Mock() stats.start() @@ -28,3 +17,144 @@ def test_start_and_stop(tiger): # Stats are no longer being collected time.sleep(0.22) assert len(stats.compute_stats.mock_calls) == 3 + + +def test_utilization_and_occupancy(): + log = mock.Mock() + with mock.patch( + "tasktiger.stats.time.monotonic", + autospec=True, + side_effect=[1000.0, 1003.0, 1005.0, 1012.0, 1018.0, 1032.0], + ): + stats = StatsThread(log, interval=60) + stats.on_idle_start() + stats.on_idle_end() + stats.on_task_start() + stats.on_task_end() + + stats.compute_stats() + + assert log.info.mock_calls == [ + mock.call( + "stats", + time_total=32.0, + time_busy=6.0, + time_idle=2.0, + time_overhead=24.0, + utilization=18.75, + occupancy=75.0, + ) + ] + + +def test_in_progress_idle_time_spans_stats_windows(): + log = mock.Mock() + with mock.patch( + "tasktiger.stats.time.monotonic", + autospec=True, + side_effect=[500.0, 506.0, 520.0, 524.0, 540.0], + ): + stats = StatsThread(log, interval=60) + stats.on_idle_start() + + # Window ends mid-idle: partial idle attributed here, the rest + # to the next window. + stats.compute_stats() + + stats.on_idle_end() + stats.compute_stats() + + assert log.info.mock_calls == [ + mock.call( + "stats", + time_total=20.0, + time_busy=0.0, + time_idle=14.0, + time_overhead=6.0, + utilization=0.0, + occupancy=0.0, + ), + mock.call( + "stats", + time_total=20.0, + time_busy=0.0, + time_idle=4.0, + time_overhead=16.0, + utilization=0.0, + occupancy=0.0, + ), + ] + + +def test_in_progress_task_time_spans_stats_windows(): + log = mock.Mock() + with mock.patch( + "tasktiger.stats.time.monotonic", + autospec=True, + side_effect=[800.0, 809.0, 830.0, 838.0, 850.0], + ): + stats = StatsThread(log, interval=60) + stats.on_task_start() + + # Window ends mid-task: partial busy attributed here, the rest + # to the next window. + stats.compute_stats() + + stats.on_task_end() + stats.compute_stats() + + assert log.info.mock_calls == [ + mock.call( + "stats", + time_total=30.0, + time_busy=21.0, + time_idle=0.0, + time_overhead=9.0, + utilization=70.0, + occupancy=100.0, + ), + mock.call( + "stats", + time_total=20.0, + time_busy=8.0, + time_idle=0.0, + time_overhead=12.0, + utilization=40.0, + occupancy=100.0, + ), + ] + + +def test_occupancy_is_zero_without_tasks_or_waits(): + log = mock.Mock() + with mock.patch( + "tasktiger.stats.time.monotonic", + autospec=True, + side_effect=[500.0, 520.0], + ): + stats = StatsThread(log, interval=60) + + stats.compute_stats() + + assert log.info.mock_calls == [ + mock.call( + "stats", + time_total=20.0, + time_busy=0.0, + time_idle=0.0, + time_overhead=20.0, + utilization=0.0, + occupancy=0.0, + ) + ] + + +def test_stats_consumer_defaults_are_noops(): + consumer = StatsConsumer() + + consumer.start() + consumer.on_task_start() + consumer.on_task_end() + consumer.on_idle_start() + consumer.on_idle_end() + consumer.stop() diff --git a/tests/test_workers.py b/tests/test_workers.py index a4c895c9..17793b17 100644 --- a/tests/test_workers.py +++ b/tests/test_workers.py @@ -3,6 +3,7 @@ import datetime import time from multiprocessing import Process +from unittest import mock import pytest from freezefrog import FreezeTime @@ -10,6 +11,7 @@ from tasktiger import Task, Worker from tasktiger._internal import ACTIVE from tasktiger.executor import SyncExecutor +from tasktiger.stats import StatsConsumer, StatsThread from tasktiger.worker import LOCK_REDIS_KEY from .config import DELAY @@ -258,3 +260,86 @@ def test_stop_heartbeat_thread_on_unhandled_exception(self, tiger, ensure_queues # handled by the executor, the task is still active until it times out # and gets requeued by another worker. ensure_queues(active={"default": 1}) + + +def test_worker_runs_without_stats_consumers(tiger): + worker = Worker(tiger) + + worker.run(once=True, force_once=True) + + assert worker.stats == [] + + +def test_poll_for_queues_measures_idle_time(tiger): + tiger.config["POLL_TASK_QUEUES_INTERVAL"] = 0.01 + worker = Worker(tiger) + consumer = mock.create_autospec(StatsConsumer, instance=True) + worker.stats = [consumer] + worker._did_work = False + + worker._poll_for_queues() + + assert consumer.on_idle_start.mock_calls == [mock.call()] + assert consumer.on_idle_end.mock_calls == [mock.call()] + + +def test_poll_for_queues_without_stats_consumers(tiger): + tiger.config["POLL_TASK_QUEUES_INTERVAL"] = 0.01 + worker = Worker(tiger) + worker._did_work = False + assert worker.stats == [] + + worker._poll_for_queues() + + +def test_worker_measures_idle_time(tiger): + consumer = mock.create_autospec(StatsConsumer, instance=True) + tiger.config["STATS_CONSUMERS"] = [consumer] + + Worker(tiger).run(once=True, force_once=True) + + assert consumer.on_idle_start.mock_calls == [mock.call()] + assert consumer.on_idle_end.mock_calls == [mock.call()] + + +def test_worker_reports_to_configured_stats_consumers(tiger): + consumer = mock.create_autospec(StatsConsumer, instance=True) + tiger.config["STATS_CONSUMERS"] = [consumer] + Task(tiger, simple_task).delay() + + Worker(tiger, executor_class=SyncExecutor).run(once=True, force_once=True) + + assert consumer.mock_calls == [ + mock.call.start(), + mock.call.on_idle_start(), + mock.call.on_idle_end(), + mock.call.on_task_start(), + mock.call.on_task_end(), + mock.call.stop(), + ] + + +def test_measure_task_propagates_exception_and_still_ends(tiger): + worker = Worker(tiger) + consumer = mock.create_autospec(StatsConsumer, instance=True) + worker.stats = [consumer] + + class Boom(Exception): + pass + + with pytest.raises(Boom): + with worker._measure_task(): + raise Boom("task blew up") + + assert consumer.on_task_start.mock_calls == [mock.call()] + assert consumer.on_task_end.mock_calls == [mock.call()] + + +def test_worker_manages_stats_thread_lifecycle(tiger): + stats_thread = StatsThread(mock.Mock(), interval=60) + tiger.config["STATS_CONSUMERS"] = [stats_thread] + + Worker(tiger).run(once=True, force_once=True) + + stats_thread.join(timeout=5) + assert not stats_thread.is_alive()