return idle_workers_info
def get_busy_workers(self) -> list[_WorkerInfo]:
"""
Gets a WorkerInfo object for all busy workers.
This uses Celery's API to poll active queues.
Use the `busy_workers` property to avoid re-polling.
Returns:
list: List of WorkerInfo objects for all idle workers.
"""
# TODO: Get busy workers
# Seems to be a glitch getting busy workers in this version of Celery.
# Fix planned for next release Feb or March.
# This might work, but isn't yet:
i = celery_app.control.inspect(timeout=10)
i.pattern = self.vc_key + "*"
i.limit = 1
logger.debug(f"[magenta]Active: {i.active_queues()}")
busy_workers_info = []
return busy_workers_info
if not busy_workers:
return busy_workers_info
for worker, attributes in busy_workers.items():
worker_vc_key = attributes[0]["routing_key"]
worker_info = _WorkerInfo(
name=worker,
host=str(worker).split("@")[1],
version_constraint_key=worker_vc_key,
compatible=worker_vc_key == self.vc_key,
)
busy_workers_info.append(worker_info)
return busy_workers_info
class AppStatus:
def __init__(self, package_name: str):
Get busy workers
Seems to be a glitch getting busy workers in this version of Celery.
Fix planned for next release Feb or March.
This might work, but isn't yet:
https://github.com/in03/proxima/blob/18431a76c515bb1ac976dbba54691be02d474e6c/src/proxima/app/checks.py#L138
26e24bda85ba6727e5cc66313769425c999e9de8