API only allow MAX_CONCURRENT_JOB*NODE_COUNT concurrent jobs

bitmakerla · Sep 16, 2022 · 21b01a5 · 21b01a5
1 parent 1e99d85
commit 21b01a5
Show file tree

Hide file tree

Showing 6 changed files with 26 additions and 5 deletions.
diff --git a/estela-api/config/settings/local.py b/estela-api/config/settings/local.py
@@ -2,7 +2,7 @@
 
 DEBUG = True
 
-RUN_JOBS_PER_LOT = 10
 CHECK_JOB_ERRORS_BATCH_SIZE = 10
+MAX_CONCURRENT_JOBS = 10
 
 SPIDERDATA_DB_PRODUCTION = False
diff --git a/estela-api/config/settings/prod.py b/estela-api/config/settings/prod.py
@@ -2,8 +2,8 @@
 
 DEBUG = False
 
-RUN_JOBS_PER_LOT = 1000
 CHECK_JOB_ERRORS_BATCH_SIZE = 1000
+MAX_CONCURRENT_JOBS = 40
 
 MULTI_NODE_MODE = True
 

diff --git a/estela-api/config/settings/test.py b/estela-api/config/settings/test.py
@@ -2,7 +2,7 @@
 
 DEBUG = True
 
-RUN_JOBS_PER_LOT = 10
 CHECK_JOB_ERRORS_BATCH_SIZE = 10
+MAX_CONCURRENT_JOBS = 10
 
 MONGO_PRODUCTION = False
diff --git a/estela-api/core/tasks.py b/estela-api/core/tasks.py
@@ -22,10 +22,16 @@ def get_default_token(job):
 
 @celery_app.task
 def run_spider_jobs():
+    running_jobs_count = SpiderJob.objects.filter(
+        status=SpiderJob.RUNNING_STATUS
+    ).count()
+    number_of_jobs_to_run = (
+        settings.MAX_CONCURRENT_JOBS * job_manager.get_scale_size()
+    ) - running_jobs_count
+
     jobs = SpiderJob.objects.filter(status=SpiderJob.IN_QUEUE_STATUS)[
-        : settings.RUN_JOBS_PER_LOT
+        :number_of_jobs_to_run
     ]
-
     for job in jobs:
         job_args = {arg.name: arg.value for arg in job.args.all()}
         job_env_vars = {env_var.name: env_var.value for env_var in job.env_vars.all()}

diff --git a/estela-api/engines/example_engine.py b/estela-api/engines/example_engine.py
@@ -24,3 +24,6 @@ def read_job(self, job_name):
 
     def read_job_status(self, job_name):
         return self.Status()
+
+    def get_scale_size(self):
+        pass
diff --git a/estela-api/engines/kubernetes.py b/estela-api/engines/kubernetes.py
@@ -188,3 +188,15 @@ def read_job_status(self, name, namespace="default", api_instance=None):
             return None
 
         return self.Status(api_response.status)
+
+    def get_scale_size(self):
+        if api_instance is None:
+            api_instance = self.get_api_instance()
+
+        try:
+            api_response = api_instance.list_node()
+        except ApiException:
+            return None
+
+        number_of_nodes = len(api_response.items)
+        return number_of_nodes