From c109775416ecf981888df721300e2a686f75426e Mon Sep 17 00:00:00 2001 From: Carlos Feria <2582866+carlosthe19916@users.noreply.github.com> Date: Fri, 31 Jul 2026 18:09:48 +0000 Subject: [PATCH] fix: make redis workers to know the current num of workers at startup Signed-off-by: Carlos Feria <2582866+carlosthe19916@users.noreply.github.com> --- CHANGES/+redis-worker-startup-poll.bugfix | 1 + pulpcore/tasking/redis_worker.py | 19 ++++++++++++++----- 2 files changed, 15 insertions(+), 5 deletions(-) create mode 100644 CHANGES/+redis-worker-startup-poll.bugfix diff --git a/CHANGES/+redis-worker-startup-poll.bugfix b/CHANGES/+redis-worker-startup-poll.bugfix new file mode 100644 index 00000000000..fe100f6acb1 --- /dev/null +++ b/CHANGES/+redis-worker-startup-poll.bugfix @@ -0,0 +1 @@ +Fixed Redis workers polling the database at the single-worker rate during startup by learning the online worker count before the first heartbeat. diff --git a/pulpcore/tasking/redis_worker.py b/pulpcore/tasking/redis_worker.py index d9449a885df..5ad63568216 100644 --- a/pulpcore/tasking/redis_worker.py +++ b/pulpcore/tasking/redis_worker.py @@ -133,8 +133,10 @@ def __init__(self): # Metric recording interval self.metric_heartbeat_countdown = METRIC_HEARTBEAT_INTERVAL - # Cache worker count for sleep calculation (updated during beat) - self.num_workers = 1 + # Cache worker count for sleep calculation. Learn the real fleet size at + # startup so new workers do not poll at the single-worker rate until the + # first heartbeat (~WORKER_TTL/3). Refreshed on each heartbeat in beat(). + self.num_workers = max(1, AppStatus.objects.online().filter(app_type="worker").count()) # Redis connection for distributed locks self.redis_conn = get_redis_connection() @@ -157,7 +159,10 @@ def __init__(self): startup_hook() - _logger.info("Initialized RedisWorker with Redis lock-based algorithm") + _logger.info( + "Initialized RedisWorker with Redis lock-based algorithm (online workers=%d)", + self.num_workers, + ) def _init_instrumentation(self): """Initialize OpenTelemetry instrumentation if enabled.""" @@ -210,7 +215,11 @@ def handle_worker_heartbeat(self): self.app_status.save_heartbeat() _logger.debug(msg) except (IntegrityError, DatabaseError): - _logger.error(f"Updating the heartbeat of worker {self.name} failed.") + _logger.error( + "Updating the heartbeat of worker %s failed.", + self.name, + exc_info=True, + ) self.shutdown_requested = True def handle_redis_heartbeat(self): @@ -363,7 +372,7 @@ def beat(self): self.record_waiting_tasks_metric() # Update cached worker count for sleep calculation - self.num_workers = AppStatus.objects.online().filter(app_type="worker").count() + self.num_workers = max(1, AppStatus.objects.online().filter(app_type="worker").count()) def _maybe_release_locks(self, task, mark_released=True): """