diff --git a/CHANGES/7986.feature b/CHANGES/7986.feature new file mode 100644 index 0000000000..568a668e19 --- /dev/null +++ b/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. diff --git a/pulpcore/tasking/redis_locks.py b/pulpcore/tasking/redis_locks.py index 7d812945a0..01e65791a4 100644 --- a/pulpcore/tasking/redis_locks.py +++ b/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:") diff --git a/pulpcore/tasking/redis_worker.py b/pulpcore/tasking/redis_worker.py index 805e8bc302..c653ee9819 100644 --- a/pulpcore/tasking/redis_worker.py +++ b/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: