feat: separate jobs processing per individual queues per processor (#231)

Author: quitrk

skynet/constants.py

@@ -2,9 +2,21 @@
 
 response_prefix = 'SkynetResponse'
 
-PENDING_JOBS_KEY = 'jobs:pending'
-RUNNING_JOBS_KEY = 'jobs:running'
-ERROR_JOBS_KEY = 'jobs:error'
+# Processor-specific queue keys
+PENDING_JOBS_OPENAI_KEY = 'jobs:pending:openai'
+PENDING_JOBS_AZURE_KEY = 'jobs:pending:azure'
+PENDING_JOBS_OCI_KEY = 'jobs:pending:oci'
+PENDING_JOBS_LOCAL_KEY = 'jobs:pending:local'
+
+RUNNING_JOBS_OPENAI_KEY = 'jobs:running:openai'
+RUNNING_JOBS_AZURE_KEY = 'jobs:running:azure'
+RUNNING_JOBS_OCI_KEY = 'jobs:running:oci'
+RUNNING_JOBS_LOCAL_KEY = 'jobs:running:local'
+
+ERROR_JOBS_OPENAI_KEY = 'jobs:error:openai'
+ERROR_JOBS_AZURE_KEY = 'jobs:error:azure'
+ERROR_JOBS_OCI_KEY = 'jobs:error:oci'
+ERROR_JOBS_LOCAL_KEY = 'jobs:error:local'
 
 
 class Locale(Enum):

skynet/env.py

@@ -115,6 +115,12 @@ def tobool(val: str | None):
 job_timeout = int(os.environ.get('JOB_TIMEOUT', 60 * 5))  # 5 minutes default
 max_concurrency = int(os.environ.get('MAX_CONCURRENCY', 5))
 
+# per-processor concurrency limits
+max_concurrency_openai = int(os.environ.get('MAX_CONCURRENCY_OPENAI', 5))
+max_concurrency_azure = int(os.environ.get('MAX_CONCURRENCY_AZURE', 5))
+max_concurrency_oci = int(os.environ.get('MAX_CONCURRENCY_OCI', 5))
+max_concurrency_local = int(os.environ.get('MAX_CONCURRENCY_LOCAL', 5))
+
 # summaries
 summary_minimum_payload_length = int(os.environ.get('SUMMARY_MINIMUM_PAYLOAD_LENGTH', 100))
 enable_batching = tobool(os.environ.get('ENABLE_BATCHING', 'true'))

skynet/metrics.py

@@ -1,4 +1,9 @@
-from skynet.constants import PENDING_JOBS_KEY
+from skynet.constants import (
+    PENDING_JOBS_AZURE_KEY,
+    PENDING_JOBS_LOCAL_KEY,
+    PENDING_JOBS_OCI_KEY,
+    PENDING_JOBS_OPENAI_KEY,
+)
 from skynet.env import enable_metrics, modules
 from skynet.logs import get_logger
 from skynet.modules.monitoring import (
@@ -29,8 +34,18 @@ async def autoscaler_metrics():
         Metrics required for the autoscaler.
         '''
 
-        queue_size = await db.llen(PENDING_JOBS_KEY)
-        return {'queueSize': queue_size}
+        # Sum up queue sizes from all processor-specific queues
+        total_queue_size = 0
+        for queue_key in [
+            PENDING_JOBS_OPENAI_KEY,
+            PENDING_JOBS_AZURE_KEY,
+            PENDING_JOBS_OCI_KEY,
+            PENDING_JOBS_LOCAL_KEY,
+        ]:
+            queue_size = await db.llen(queue_key)
+            total_queue_size += queue_size
+
+        return {'queueSize': total_queue_size}
 
     if 'summaries:dispatcher' in modules:
         from skynet.modules.ttt.summaries.app import app as summaries_app