Reduce memory use of the waiting_tasks metric

CHANGES/7986.feature

@@ -0,0 +1 @@
+Reduced the memory overhead of the Redis worker `waiting_tasks` metric by streaming reserved-resource records instead of loading full `Task` objects.

pulpcore/tasking/redis_locks.py

@@ -427,7 +427,23 @@ def extract_task_resources(task):
             exclusive_resources: List of exclusive resource names
             shared_resources: List of shared resource names (with "shared:" prefix stripped)
     """
-    reserved_resources_record = task.reserved_resources_record or []
+    return _split_reserved_resources(task.reserved_resources_record)
+
+
+def _split_reserved_resources(reserved_resources_record):
+    """
+    Split a `reserved_resources_record` array into (exclusive, shared) resource names.
+
+    Shared reservations are stored with a `"shared:"` prefix; exclusive ones are bare.
+    The prefix is stripped from the returned shared names.
+
+    Args:
+        reserved_resources_record (list[str] | None): Raw reservation strings, or None.
+
+    Returns:
+        tuple[list[str], list[str]]: (exclusive_resources, shared_resources).
+    """
+    reserved_resources_record = reserved_resources_record or []
 
     exclusive_resources = [
         resource for resource in reserved_resources_record if not resource.startswith("shared:")

pulpcore/tasking/redis_worker.py

@@ -45,6 +45,7 @@
     IMMEDIATE_OWNER_PREFIX,
     LEGACY_OWNER_SCAN_INTERVAL,
     LEGACY_OWNER_SCAN_KEY,
+    _split_reserved_resources,
     acquire_locks,
     cleanup_locks_for_owner,
     collect_lock_owners,
@@ -110,21 +111,22 @@ def count_waiting_tasks_for_metric():
     """
     cutoff_time = timezone.now() - timedelta(seconds=5)
 
-    incomplete_tasks = (
+    resources_iter = (
         Task.objects.filter(
             state__in=[TASK_STATES.RUNNING, TASK_STATES.WAITING],
             pulp_created__lt=cutoff_time,
         )
         .order_by("pulp_created")
-        .only("reserved_resources_record")
+        .values_list("reserved_resources_record", flat=True)  # Avoids Task object allocation
+        .iterator()
     )
 
     taken_exclusive = set()
     taken_shared = set()
     parallel_count = 0
 
-    for task in incomplete_tasks:
-        exclusive_resources, shared_resources = extract_task_resources(task)
+    for reserved in resources_iter:
+        exclusive_resources, shared_resources = _split_reserved_resources(reserved)
         conflicts = False
 
         for resource in exclusive_resources: