From ce7d292830d56f05a631050d0f8b33752a1a1e02 Mon Sep 17 00:00:00 2001 From: Carlos Feria <2582866+carlosthe19916@users.noreply.github.com> Date: Tue, 4 Aug 2026 21:44:51 +0200 Subject: [PATCH] fix: improve RedisWorker auto-scaling with resource-aware task metrics Signed-off-by: Carlos Feria <2582866+carlosthe19916@users.noreply.github.com> --- CHANGES/7938.bugfix | 1 + pulpcore/tasking/redis_worker.py | 66 +++- pulpcore/tests/functional/api/test_tasking.py | 292 ++++++++++++++++++ 3 files changed, 347 insertions(+), 12 deletions(-) create mode 100644 CHANGES/7938.bugfix diff --git a/CHANGES/7938.bugfix b/CHANGES/7938.bugfix new file mode 100644 index 0000000000..53f8e096bb --- /dev/null +++ b/CHANGES/7938.bugfix @@ -0,0 +1 @@ +Fixed the `waiting_tasks` metric to count only tasks that can run in parallel under resource locks, matching worker FIFO scheduling. diff --git a/pulpcore/tasking/redis_worker.py b/pulpcore/tasking/redis_worker.py index 5ad6356821..afde0a4f29 100644 --- a/pulpcore/tasking/redis_worker.py +++ b/pulpcore/tasking/redis_worker.py @@ -93,6 +93,57 @@ def _f(self, *args, **kwargs): return _decorator +def count_waiting_tasks_for_metric(): + """ + Count WAITING/RUNNING tasks older than five seconds that can run at the same + time given exclusive/shared resource reservations (FIFO-aware). + + Returns: + int: Parallelizable unfinished-task count. Callers subtract online workers + when publishing the OpenTelemetry `waiting_tasks` gauge. + """ + cutoff_time = timezone.now() - timedelta(seconds=5) + + incomplete_tasks = ( + Task.objects.filter( + state__in=[TASK_STATES.RUNNING, TASK_STATES.WAITING], + pulp_created__lt=cutoff_time, + ) + .order_by("pulp_created") + .only("reserved_resources_record") + ) + + taken_exclusive = set() + taken_shared = set() + parallel_count = 0 + + for task in incomplete_tasks: + exclusive_resources, shared_resources = extract_task_resources(task) + conflicts = False + + for resource in exclusive_resources: + if resource in taken_exclusive or resource in taken_shared: + conflicts = True + break + + if not conflicts: + for resource in shared_resources: + if resource in taken_exclusive: + conflicts = True + break + + # Always reserve (even on conflict) so FIFO can't be bypassed; see fetch_task. + # e.g. T1(A,B), T2(B,C), T3(C): without reserving C for T2, T3 counts as +1. + taken_exclusive.update(exclusive_resources) + taken_shared.update(shared_resources) + if conflicts: + continue + + parallel_count += 1 + + return parallel_count + + class RedisWorker: """ Worker implementation using Redis distributed lock-based resource acquisition. @@ -331,19 +382,10 @@ def record_waiting_tasks_metric(self): """ Record metrics for waiting tasks in the queue. - This method counts all tasks in RUNNING or WAITING state that are older - than 5 seconds, then subtracts the number of active workers to get the - number of tasks waiting to be picked up by workers. + Publishes `count_waiting_tasks_for_metric() - num_workers` as the + OpenTelemetry `waiting_tasks` gauge. """ - cutoff_time = timezone.now() - timedelta(seconds=5) - - task_count = Task.objects.filter( - state__in=[TASK_STATES.RUNNING, TASK_STATES.WAITING], pulp_created__lt=cutoff_time - ).count() - - waiting_tasks = task_count - self.num_workers - - self.waiting_tasks_meter.set(waiting_tasks) + self.waiting_tasks_meter.set(count_waiting_tasks_for_metric() - self.num_workers) def beat(self): """Periodic worker maintenance tasks (heartbeat, cleanup, etc.).""" diff --git a/pulpcore/tests/functional/api/test_tasking.py b/pulpcore/tests/functional/api/test_tasking.py index d9e95a56a0..c9186563cd 100644 --- a/pulpcore/tests/functional/api/test_tasking.py +++ b/pulpcore/tests/functional/api/test_tasking.py @@ -13,6 +13,7 @@ from pulpcore.client.pulpcore import ApiException from pulpcore.constants import IMMEDIATE_TIMEOUT +from pulpcore.tasking.redis_worker import count_waiting_tasks_for_metric from pulpcore.tests.functional.utils import PulpTaskError, download_file @@ -86,6 +87,297 @@ def test_multi_resource_locking(dispatch_task, monitor_task): assert task1.finished_at < task5.started_at +@pytest.fixture +def _read_waiting_tasks_metric_value(django_db_blocker): + """ + Read count_waiting_tasks_for_metric against the live Pulp database. + + Unblock Django DB access without @pytest.mark.django_db, which would create a + separate test_pulp database that does not contain tasks dispatched via the API. + """ + + def _read(): + with django_db_blocker.unblock(): + return count_waiting_tasks_for_metric() + + return _read + + +def test_waiting_tasks_metric_resource_contention( + dispatch_task, pulpcore_bindings, _read_waiting_tasks_metric_value +): + """Exclusive holder + exclusive/shared waiters on same resource → metric +1. + + Not @pytest.mark.parallel — global counter; baseline deltas race. + """ + exclusive_resource = str(uuid4()) + blocked_exclusive_waiter_count = 5 + blocked_shared_waiter_count = 5 + expected_parallel_capacity = 1 + dispatched_task_hrefs = [] + + waiting_tasks_metric_before = _read_waiting_tasks_metric_value() + + try: + # One long-running task that holds the exclusive lock. + lock_holder_task_href = dispatch_task( + "pulpcore.app.tasks.test.sleep", + args=(60,), + exclusive_resources=[exclusive_resource], + ) + dispatched_task_hrefs.append(lock_holder_task_href) + + # Wait until the task is running. Max 15 seconds (30 * 0.5) + lock_holder_task = None + for _ in range(30): + lock_holder_task = pulpcore_bindings.TasksApi.read(lock_holder_task_href) + if lock_holder_task.state == "running": + break + time.sleep(0.5) + assert lock_holder_task is not None and lock_holder_task.state == "running", ( + f"Lock-holder task did not start running, " + f"state={getattr(lock_holder_task, 'state', None)}" + ) + + # Exclusive waiters on the same resource stay waiting until the holder finishes. + for _ in range(blocked_exclusive_waiter_count): + dispatched_task_hrefs.append( + dispatch_task( + "pulpcore.app.tasks.test.sleep", + args=(1,), + exclusive_resources=[exclusive_resource], + ) + ) + # Shared waiters on the same resource are also blocked by the exclusive holder. + for _ in range(blocked_shared_waiter_count): + dispatched_task_hrefs.append( + dispatch_task( + "pulpcore.app.tasks.test.sleep", + args=(1,), + shared_resources=[exclusive_resource], + ) + ) + + # Metric only counts tasks older than 5 seconds. + time.sleep(6) + + waiting_tasks_metric_increase = ( + _read_waiting_tasks_metric_value() - waiting_tasks_metric_before + ) + naive_unfinished_count = ( + expected_parallel_capacity + + blocked_exclusive_waiter_count + + blocked_shared_waiter_count + ) + assert waiting_tasks_metric_increase == expected_parallel_capacity, ( + f"Expected waiting_tasks metric to rise by {expected_parallel_capacity} " + f"(one runnable lane on exclusive resource {exclusive_resource!r}), " + f"got +{waiting_tasks_metric_increase} " + f"(naive unfinished count would add +{naive_unfinished_count})" + ) + finally: + for task_href in dispatched_task_hrefs: + try: + pulpcore_bindings.TasksApi.tasks_cancel(task_href, {"state": "canceled"}) + except ApiException: + pass + + +def test_waiting_tasks_metric_two_exclusive_lanes( + dispatch_task, pulpcore_bindings, _read_waiting_tasks_metric_value +): + """Two independent exclusive lanes with waiters → metric +2. + + Not @pytest.mark.parallel — global counter; baseline deltas race. + """ + exclusive_resource_a = str(uuid4()) + exclusive_resource_b = str(uuid4()) + blocked_waiting_task_count_per_lane = 5 + expected_parallel_capacity = 2 + dispatched_task_hrefs = [] + + waiting_tasks_metric_before = _read_waiting_tasks_metric_value() + + try: + # Lane A: one running holder, then waiters blocked on the same resource. + lock_holder_a_href = dispatch_task( + "pulpcore.app.tasks.test.sleep", + args=(60,), + exclusive_resources=[exclusive_resource_a], + ) + dispatched_task_hrefs.append(lock_holder_a_href) + + lock_holder_a = None + for _ in range(30): + lock_holder_a = pulpcore_bindings.TasksApi.read(lock_holder_a_href) + if lock_holder_a.state == "running": + break + time.sleep(0.5) + assert lock_holder_a is not None and lock_holder_a.state == "running", ( + f"Lock-holder A did not start running, state={getattr(lock_holder_a, 'state', None)}" + ) + + for _ in range(blocked_waiting_task_count_per_lane): + dispatched_task_hrefs.append( + dispatch_task( + "pulpcore.app.tasks.test.sleep", + args=(1,), + exclusive_resources=[exclusive_resource_a], + ) + ) + + # Lane B: independent exclusive resource. Holder may stay waiting if only + # one worker is free; the metric still counts it as a second parallel lane. + lock_holder_b_href = dispatch_task( + "pulpcore.app.tasks.test.sleep", + args=(60,), + exclusive_resources=[exclusive_resource_b], + ) + dispatched_task_hrefs.append(lock_holder_b_href) + + for _ in range(blocked_waiting_task_count_per_lane): + dispatched_task_hrefs.append( + dispatch_task( + "pulpcore.app.tasks.test.sleep", + args=(1,), + exclusive_resources=[exclusive_resource_b], + ) + ) + + # Metric only counts tasks older than 5 seconds. + time.sleep(6) + + waiting_tasks_metric_increase = ( + _read_waiting_tasks_metric_value() - waiting_tasks_metric_before + ) + unfinished_task_count = expected_parallel_capacity + 2 * blocked_waiting_task_count_per_lane + assert waiting_tasks_metric_increase == expected_parallel_capacity, ( + f"Expected waiting_tasks metric to rise by {expected_parallel_capacity} " + f"(one lane each on {exclusive_resource_a!r} and {exclusive_resource_b!r}), " + f"got +{waiting_tasks_metric_increase} " + f"(naive unfinished count would add +{unfinished_task_count})" + ) + finally: + for task_href in dispatched_task_hrefs: + try: + pulpcore_bindings.TasksApi.tasks_cancel(task_href, {"state": "canceled"}) + except ApiException: + pass + + +def test_waiting_tasks_metric_shared_resources_can_run_together( + dispatch_task, pulpcore_bindings, _read_waiting_tasks_metric_value +): + """N shared-only tasks on same resource → metric +N (not +1). + + Not @pytest.mark.parallel — global counter; baseline deltas race. + """ + shared_resource = str(uuid4()) + shared_task_count = 5 + dispatched_task_hrefs = [] + + waiting_tasks_metric_before = _read_waiting_tasks_metric_value() + + try: + for _ in range(shared_task_count): + dispatched_task_hrefs.append( + dispatch_task( + "pulpcore.app.tasks.test.sleep", + args=(60,), + shared_resources=[shared_resource], + ) + ) + + # Metric only counts tasks older than 5 seconds. + time.sleep(6) + + waiting_tasks_metric_increase = ( + _read_waiting_tasks_metric_value() - waiting_tasks_metric_before + ) + assert waiting_tasks_metric_increase == shared_task_count, ( + f"Expected waiting_tasks metric to rise by {shared_task_count} " + f"(shared holders of {shared_resource!r} can run in parallel), " + f"got +{waiting_tasks_metric_increase}" + ) + finally: + for task_href in dispatched_task_hrefs: + try: + pulpcore_bindings.TasksApi.tasks_cancel(task_href, {"state": "canceled"}) + except ApiException: + pass + + +def test_waiting_tasks_metric_fifo_blocks_indirect_resources( + dispatch_task, pulpcore_bindings, _read_waiting_tasks_metric_value +): + """T1(A,B) running, T2(B,C) and T3(C) waiting → metric +1 (FIFO), not +2. + + Not @pytest.mark.parallel — global counter; baseline deltas race. + """ + resource_a = str(uuid4()) + resource_b = str(uuid4()) + resource_c = str(uuid4()) + expected_parallel_capacity = 1 + dispatched_task_hrefs = [] + + waiting_tasks_metric_before = _read_waiting_tasks_metric_value() + + try: + # T1 holds A and B. + lock_holder_task_href = dispatch_task( + "pulpcore.app.tasks.test.sleep", + args=(60,), + exclusive_resources=[resource_a, resource_b], + ) + dispatched_task_hrefs.append(lock_holder_task_href) + + lock_holder_task = None + for _ in range(30): + lock_holder_task = pulpcore_bindings.TasksApi.read(lock_holder_task_href) + if lock_holder_task.state == "running": + break + time.sleep(0.5) + assert lock_holder_task is not None and lock_holder_task.state == "running", ( + f"Lock-holder task did not start running, " + f"state={getattr(lock_holder_task, 'state', None)}" + ) + + # T2 needs B and C — blocked on B, but still reserves C for FIFO. + dispatched_task_hrefs.append( + dispatch_task( + "pulpcore.app.tasks.test.sleep", + args=(1,), + exclusive_resources=[resource_b, resource_c], + ) + ) + # T3 needs only C — must not count as a second lane ahead of T2. + dispatched_task_hrefs.append( + dispatch_task( + "pulpcore.app.tasks.test.sleep", + args=(1,), + exclusive_resources=[resource_c], + ) + ) + + # Metric only counts tasks older than 5 seconds. + time.sleep(6) + + waiting_tasks_metric_increase = ( + _read_waiting_tasks_metric_value() - waiting_tasks_metric_before + ) + assert waiting_tasks_metric_increase == expected_parallel_capacity, ( + f"Expected waiting_tasks metric to rise by {expected_parallel_capacity} " + f"(FIFO: T2 blocks T3 on {resource_c!r}), " + f"got +{waiting_tasks_metric_increase}" + ) + finally: + for task_href in dispatched_task_hrefs: + try: + pulpcore_bindings.TasksApi.tasks_cancel(task_href, {"state": "canceled"}) + except ApiException: + pass + + @pytest.mark.long_running @pytest.mark.parallel def test_worker_cleanup_on_missing_worker(dispatch_task, monitor_task, pulpcore_bindings):