SW-720: only receive and start new jobs when enough resources are available

LTA/LTAIngest/LTAIngestServer/LTAIngestTransferServer/lib/ingesttransferserver.py

@@ -86,7 +86,7 @@ class IngestTransferServer:
         self.__running_jobs_log_timestamp = datetime.utcnow()
 
     def start_job(self, job_dict):
-        if not self.__enoughResourcesAvailable():
+        if not self.enoughResourcesAvailable():
             raise ResourceWarning("Not enough resources available to start new job: %s" % job_dict)
 
         job_id = job_dict['JobId']
@@ -126,7 +126,7 @@ class IngestTransferServer:
         except Exception as e:
             logger.error('__clearFinishedJobs: %s', e)
 
-    def __enoughResourcesAvailable(self):
+    def enoughResourcesAvailable(self):
         try:
             now = datetime.utcnow()
             bytes_sent = _getBytesSent()
@@ -290,6 +290,11 @@ class IngestJobsForTransferHandler(AbstractMessageHandler):
         self._transfer_server = transfer_server
         super(IngestJobsForTransferHandler, self).__init__()
 
+    def before_receive_message(self):
+        while not self._transfer_server.enoughResourcesAvailable():
+            logger.info("Waiting for resources to become available before receiving a new job...")
+            time.sleep(10)
+
     def handle_message(self, msg: LofarMessage):
         if not isinstance(msg, CommandMessage):
             raise ValueError("%s: Ignoring non-CommandMessage: %s" % (self.__class__.__name__, msg))
@@ -303,7 +308,7 @@ class IngestJobsForTransferHandler(AbstractMessageHandler):
             # so jobs have a little time to start consuming resources
             # this limits the numer of jobs that can be started to 1000 starts per minute
             # it does not limit the total number of parallel jobs
-            # that is limited dynamically by __enoughResourcesAvailable
+            # that is limited dynamically by enoughResourcesAvailable
             # and by the hard limit self.max_nr_of_parallel_jobs
             time.sleep(0.1)