test: refactor record_waiting_tasks_metric metric

pulpcore/tasking/redis_worker.py

@@ -93,6 +93,63 @@ 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)
+
+    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:
     """
     Worker implementation using Redis distributed lock-based resource acquisition.
@@ -322,19 +379,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.)."""

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,296 @@ 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(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, _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.
+
+    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
+
+
+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.
+
+    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, _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
+    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
+
+
+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):