Quality of Service status refactor

cads_broker/config.py

@@ -28,7 +28,7 @@
 class BrokerConfig(pydantic_settings.BaseSettings):
     broker_priority_algorithm: str = "legacy"
     broker_priority_interval_hours: int = 24
-    broker_get_number_of_workers_cache_time: int = 10
+    broker_get_number_of_workers_cache_time: int = 60
     broker_qos_rules_cache_time: int = 10
     broker_get_tasks_from_scheduler_cache_time: int = 1
     broker_rules_path: str = "/src/rules.qos"
@@ -42,7 +42,12 @@ class BrokerConfig(pydantic_settings.BaseSettings):
     broker_max_dismissed_requests: int = 100
     broker_cancel_stuck_requests_cache_ttl: int = 60
     broker_stuck_requests_limit_minutes: int = 15
-    broker_memory_error_user_visible_log: str = "Worker has been killed due to memory usage."
+    broker_memory_error_user_visible_log: str = (
+        "Worker has been killed due to memory usage."
+    )
+    broker_workers_gap: int = (
+        10  # max discrepancy of workers number before qos rules are reloaded
+    )
 
 
 class SqlalchemySettings(pydantic_settings.BaseSettings):

cads_broker/database.py

@@ -652,21 +652,18 @@ def get_qos_status_from_request(
 def requeue_request(
     request: SystemRequest,
     session: sa.orm.Session,
-) -> SystemRequest | None:
-    if request.status == "running":
-        # ugly implementation because sqlalchemy doesn't allow to directly update JSONB
-        # FIXME: use a specific column for resubmit_number
-        metadata = dict(request.request_metadata)
-        metadata.update(
-            {"resubmit_number": request.request_metadata.get("resubmit_number", 0) + 1}
-        )
-        request.request_metadata = metadata
-        request.status = "accepted"
-        session.commit()
-        logger.info("requeueing request", **logger_kwargs(request=request))
-        return request
-    else:
-        return None
+) -> SystemRequest:
+    # ugly implementation because sqlalchemy doesn't allow to directly update JSONB
+    # FIXME: use a specific column for resubmit_number
+    metadata = dict(request.request_metadata)
+    metadata.update(
+        {"resubmit_number": request.request_metadata.get("resubmit_number", 0) + 1}
+    )
+    request.request_metadata = metadata
+    request.status = "accepted"
+    session.commit()
+    logger.info("requeueing request", **logger_kwargs(request=request))
+    return request
 
 
 def set_request_cache_id(request_uid: str, cache_id: int, session: sa.orm.Session):
@@ -680,11 +677,9 @@ def set_request_cache_id(request_uid: str, cache_id: int, session: sa.orm.Sessio
 def set_successful_request(
     request_uid: str,
     session: sa.orm.Session,
-) -> SystemRequest | None:
+) -> SystemRequest:
     statement = sa.select(SystemRequest).where(SystemRequest.request_uid == request_uid)
     request = session.scalars(statement).one()
-    if request.status == "successful":
-        return None
     request.status = "successful"
     request.finished_at = sa.func.now()
     session.commit()