From 9d3cc54ec5769104c8d954a73bcc3e25c886aa9b Mon Sep 17 00:00:00 2001 From: Carlos Feria <2582866+carlosthe19916@users.noreply.github.com> Date: Thu, 30 Jul 2026 10:01:38 +0000 Subject: [PATCH 1/4] test: create test for assesing queue size for tasks Signed-off-by: Carlos Feria <2582866+carlosthe19916@users.noreply.github.com> --- pulpcore/tasking/redis_worker.py | 40 +++++--- pulpcore/tests/functional/api/test_tasking.py | 95 +++++++++++++++++++ 2 files changed, 123 insertions(+), 12 deletions(-) diff --git a/pulpcore/tasking/redis_worker.py b/pulpcore/tasking/redis_worker.py index d9449a885df..eecbc98e148 100644 --- a/pulpcore/tasking/redis_worker.py +++ b/pulpcore/tasking/redis_worker.py @@ -93,6 +93,31 @@ def _f(self, *args, **kwargs): return _decorator +def count_waiting_tasks_for_metric(num_workers): + """ + Compute the waiting_tasks autoscaling metric value. + + Counts how many WAITING/RUNNING tasks older than five seconds can run at the + same time given exclusive/shared resource reservations, then subtracts + ``num_workers``. This value is published as the OpenTelemetry ``waiting_tasks`` + gauge used by KEDA. + + Args: + num_workers (int): Number of online workers to subtract from the + parallelizable unfinished-task count. + + Returns: + int: Metric value (may be negative when there are more workers than + parallelizable tasks). + """ + 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() + return task_count - num_workers + + class RedisWorker: """ Worker implementation using Redis distributed lock-based resource acquisition. @@ -322,19 +347,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`` as the OpenTelemetry + ``waiting_tasks`` gauge used by KEDA for worker autoscaling. """ - 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 d9e95a56a0c..d6dfae01907 100644 --- a/pulpcore/tests/functional/api/test_tasking.py +++ b/pulpcore/tests/functional/api/test_tasking.py @@ -1,6 +1,7 @@ """Tests related to the tasking system.""" import json +import subprocess import time from contextlib import contextmanager from datetime import datetime @@ -86,6 +87,100 @@ def test_multi_resource_locking(dispatch_task, monitor_task): assert task1.finished_at < task5.started_at +def _read_waiting_tasks_metric_value(num_workers=1): + """ + Return the current waiting_tasks metric as an int from the live Pulp server. + + Functional tests cannot call that counter in Python directly, so this runs a + one-line command inside Pulp (pulpcore-manager shell) and returns the number + it prints. + """ + commands = ( + "from pulpcore.tasking.redis_worker import count_waiting_tasks_for_metric;" + f"print(count_waiting_tasks_for_metric({num_workers}))" + ) + process = subprocess.run( + ["pulpcore-manager", "shell", "-c", commands], capture_output=True, check=False + ) + assert process.returncode == 0, process.stderr.decode() + return int(process.stdout.decode().strip()) + + +def test_waiting_tasks_metric_resource_contention(dispatch_task, pulpcore_bindings): + """ + Verify the waiting_tasks autoscaling metric reports how many tasks can run in + parallel under resource locks, not how many unfinished tasks are queued. + + Setup: hold one exclusive resource with a long-running sleep, then enqueue many + more sleeps that need the same exclusive resource (they stay waiting). After the + metric's age cutoff, the published demand for this contention must rise by one + (a single runnable lane), not by one plus every waiter. + + That is the signal KEDA should scale on: workers help only when work is + runnable, not when it is blocked on a held lock. + + Do not mark this test @pytest.mark.parallel — the counter is global, and the + assertion uses a baseline delta that concurrent tests would race. + """ + exclusive_resource = str(uuid4()) + blocked_waiting_task_count = 10 + expected_parallel_capacity = 1 + dispatched_task_hrefs = [] + + waiting_tasks_metric_before = _read_waiting_tasks_metric_value(num_workers=1) + + 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)}" + ) + + # Same exclusive resource as the lock holder → these stay "waiting" until it finishes. + for _ in range(blocked_waiting_task_count): + dispatched_task_hrefs.append( + dispatch_task( + "pulpcore.app.tasks.test.sleep", + args=(1,), + exclusive_resources=[exclusive_resource], + ) + ) + + # Metric only counts tasks older than 5 seconds. + time.sleep(6) + + waiting_tasks_metric_increase = ( + _read_waiting_tasks_metric_value(num_workers=1) - waiting_tasks_metric_before + ) + 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 " + f"+{expected_parallel_capacity + blocked_waiting_task_count})" + ) + 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): From 3d2f16fc78bb48e177370a9044c3e8a4e0f4b63a Mon Sep 17 00:00:00 2001 From: Carlos Feria <2582866+carlosthe19916@users.noreply.github.com> Date: Thu, 30 Jul 2026 10:03:34 +0000 Subject: [PATCH 2/4] fix: task parallel count metric Signed-off-by: Carlos Feria <2582866+carlosthe19916@users.noreply.github.com> --- pulpcore/tasking/redis_worker.py | 40 ++++++++++++++++++++++++++++---- 1 file changed, 36 insertions(+), 4 deletions(-) diff --git a/pulpcore/tasking/redis_worker.py b/pulpcore/tasking/redis_worker.py index eecbc98e148..ee405f13218 100644 --- a/pulpcore/tasking/redis_worker.py +++ b/pulpcore/tasking/redis_worker.py @@ -112,10 +112,42 @@ def count_waiting_tasks_for_metric(num_workers): """ 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() - return task_count - num_workers + 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 + + if conflicts: + continue + + parallel_count += 1 + taken_exclusive.update(exclusive_resources) + taken_shared.update(shared_resources) + + return parallel_count - num_workers class RedisWorker: From be2ef80ddf86f3d5e6f527a2404055fdee533fc9 Mon Sep 17 00:00:00 2001 From: Carlos Feria <2582866+carlosthe19916@users.noreply.github.com> Date: Fri, 31 Jul 2026 08:32:11 +0200 Subject: [PATCH 3/4] fix: add more tests Signed-off-by: Carlos Feria <2582866+carlosthe19916@users.noreply.github.com> --- pulpcore/tests/functional/api/test_tasking.py | 148 ++++++++++++++++++ 1 file changed, 148 insertions(+) diff --git a/pulpcore/tests/functional/api/test_tasking.py b/pulpcore/tests/functional/api/test_tasking.py index d6dfae01907..c473d43ca81 100644 --- a/pulpcore/tests/functional/api/test_tasking.py +++ b/pulpcore/tests/functional/api/test_tasking.py @@ -181,6 +181,154 @@ def test_waiting_tasks_metric_resource_contention(dispatch_task, pulpcore_bindin pass +def test_waiting_tasks_metric_two_exclusive_lanes(dispatch_task, pulpcore_bindings): + """ + Two independent exclusive resources each with a deep waiter queue must count + as two parallel lanes (+2), not as the full unfinished-task pile. + + Do not mark @pytest.mark.parallel — the counter is global; 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(num_workers=1) + + 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(num_workers=1) - 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_exclusive_blocks_shared_waiters(dispatch_task, pulpcore_bindings): + """ + An exclusive holder of resource A blocks later tasks that only need shared + access to A. The metric must rise by 1 (one lane), not by 1 plus every + shared waiter. + + Do not mark @pytest.mark.parallel — the counter is global; baseline deltas race. + """ + exclusive_resource = str(uuid4()) + shared_waiter_count = 5 + expected_parallel_capacity = 1 + dispatched_task_hrefs = [] + + waiting_tasks_metric_before = _read_waiting_tasks_metric_value(num_workers=1) + + try: + # Exclusive holder of A — blocks anyone else using A (exclusive or shared). + 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) + + 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)}" + ) + + # Shared waiters on A — cannot start while the exclusive holder runs. + for _ in range(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(num_workers=1) - waiting_tasks_metric_before + ) + assert waiting_tasks_metric_increase == expected_parallel_capacity, ( + f"Expected waiting_tasks metric to rise by {expected_parallel_capacity} " + f"(exclusive holder of {exclusive_resource!r} blocks shared waiters), " + f"got +{waiting_tasks_metric_increase} " + f"(naive unfinished count would add " + f"+{expected_parallel_capacity + shared_waiter_count})" + ) + 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): From 531319837e56357c7f30242af9e2e3f2b017c361 Mon Sep 17 00:00:00 2001 From: Carlos Feria <2582866+carlosthe19916@users.noreply.github.com> Date: Fri, 31 Jul 2026 15:21:44 +0000 Subject: [PATCH 4/4] fix: remove pulpcore-manager & add test Signed-off-by: Carlos Feria <2582866+carlosthe19916@users.noreply.github.com> --- pulpcore/tests/functional/api/test_tasking.py | 84 +++++++++++++++---- 1 file changed, 66 insertions(+), 18 deletions(-) diff --git a/pulpcore/tests/functional/api/test_tasking.py b/pulpcore/tests/functional/api/test_tasking.py index c473d43ca81..a3a398f7a77 100644 --- a/pulpcore/tests/functional/api/test_tasking.py +++ b/pulpcore/tests/functional/api/test_tasking.py @@ -1,7 +1,6 @@ """Tests related to the tasking system.""" import json -import subprocess import time from contextlib import contextmanager from datetime import datetime @@ -14,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 @@ -87,26 +87,25 @@ def test_multi_resource_locking(dispatch_task, monitor_task): assert task1.finished_at < task5.started_at -def _read_waiting_tasks_metric_value(num_workers=1): +@pytest.fixture +def _read_waiting_tasks_metric_value(django_db_blocker): """ - Return the current waiting_tasks metric as an int from the live Pulp server. + Read count_waiting_tasks_for_metric against the live Pulp database. - Functional tests cannot call that counter in Python directly, so this runs a - one-line command inside Pulp (pulpcore-manager shell) and returns the number - it prints. + 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. """ - commands = ( - "from pulpcore.tasking.redis_worker import count_waiting_tasks_for_metric;" - f"print(count_waiting_tasks_for_metric({num_workers}))" - ) - process = subprocess.run( - ["pulpcore-manager", "shell", "-c", commands], capture_output=True, check=False - ) - assert process.returncode == 0, process.stderr.decode() - return int(process.stdout.decode().strip()) + + def _read(num_workers=1): + with django_db_blocker.unblock(): + return count_waiting_tasks_for_metric(num_workers) + + return _read -def test_waiting_tasks_metric_resource_contention(dispatch_task, pulpcore_bindings): +def test_waiting_tasks_metric_resource_contention( + dispatch_task, pulpcore_bindings, _read_waiting_tasks_metric_value +): """ Verify the waiting_tasks autoscaling metric reports how many tasks can run in parallel under resource locks, not how many unfinished tasks are queued. @@ -181,7 +180,9 @@ def test_waiting_tasks_metric_resource_contention(dispatch_task, pulpcore_bindin pass -def test_waiting_tasks_metric_two_exclusive_lanes(dispatch_task, pulpcore_bindings): +def test_waiting_tasks_metric_two_exclusive_lanes( + dispatch_task, pulpcore_bindings, _read_waiting_tasks_metric_value +): """ Two independent exclusive resources each with a deep waiter queue must count as two parallel lanes (+2), not as the full unfinished-task pile. @@ -263,7 +264,9 @@ def test_waiting_tasks_metric_two_exclusive_lanes(dispatch_task, pulpcore_bindin pass -def test_waiting_tasks_metric_exclusive_blocks_shared_waiters(dispatch_task, pulpcore_bindings): +def test_waiting_tasks_metric_exclusive_blocks_shared_waiters( + dispatch_task, pulpcore_bindings, _read_waiting_tasks_metric_value +): """ An exclusive holder of resource A blocks later tasks that only need shared access to A. The metric must rise by 1 (one lane), not by 1 plus every @@ -329,6 +332,51 @@ def test_waiting_tasks_metric_exclusive_blocks_shared_waiters(dispatch_task, pul pass +def test_waiting_tasks_metric_shared_resources_can_run_together( + dispatch_task, pulpcore_bindings, _read_waiting_tasks_metric_value +): + """ + Tasks that only need shared access to the same resource can run together, so + the metric must rise by N — not collapse to 1 as if the lock were exclusive. + + Use long sleeps so tasks stay unfinished past the metric age cutoff. + Do not mark @pytest.mark.parallel — the counter is global; baseline deltas race. + """ + shared_resource = str(uuid4()) + shared_task_count = 5 + dispatched_task_hrefs = [] + + waiting_tasks_metric_before = _read_waiting_tasks_metric_value(num_workers=1) + + 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(num_workers=1) - 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 + + @pytest.mark.long_running @pytest.mark.parallel def test_worker_cleanup_on_missing_worker(dispatch_task, monitor_task, pulpcore_bindings):