Add job queue name to response (#380)

* add job queue name to response

* lint

* Update src/actinia_core/models/response_models.py

Co-authored-by: Markus Neteler <[email protected]>

* black

* undo-black

Co-authored-by: Markus Neteler <[email protected]>

src/actinia_core/core/common/redis_interface.py

@@ -127,6 +127,7 @@ def enqueue_job(timeout, func, *args, queue_type_overwrite=None):
     global job_queues, redis_conn
     num_queues = global_config.NUMBER_OF_WORKERS
     queue_type = global_config.QUEUE_TYPE
+    queue_name = "local"
     if queue_type_overwrite:
         queue_type = global_config.QUEUE_TYPE_OVERWRITE
 
@@ -136,6 +137,7 @@ def enqueue_job(timeout, func, *args, queue_type_overwrite=None):
         __create_job_queue(queue_name)
         for i in job_queues:
             if i.name == queue_name:
+                args[0].set_queue_name(queue_name)
                 __enqueue_job_redis(i, timeout, func, *args)
 
     elif queue_type == "redis":
@@ -147,10 +149,12 @@ def enqueue_job(timeout, func, *args, queue_type_overwrite=None):
         # to chose for each job a different queue
         num = redis_conn.incr("actinia_worker_count", 1)
         current_queue = num % num_queues
+        args[0].set_queue_name(job_queues[current_queue].name)
         __enqueue_job_redis(job_queues[current_queue], timeout, func, *args)
 
     elif queue_type == "local":
         # __enqueue_job_local(timeout, func, *args)
+        args[0].set_queue_name(queue_name)
         enqueue_job_local(timeout, func, *args)
         return
         # Just in case the current process queue does not work

src/actinia_core/core/resource_data_container.py

@@ -129,6 +129,9 @@ def set_user_data(self, user_data):
     def set_request_data(self, request_data):
         self.request_data = request_data
 
+    def set_queue_name(self, queue_name):
+        self.queue = queue_name
+
     def set_storage_model_to_file(self):
         self.storage_model = "file"
 

src/actinia_core/models/response_models.py

@@ -342,6 +342,10 @@ class ProcessingResponseModel(Schema):
             "type": "string",
             "description": "The unique resource id",
         },
+        "queue": {
+            "type": "string",
+            "description": "The name of the queue in which the job is queued",
+        },
         "process_log": {
             "type": "array",
             "items": ProcessLogModel,
@@ -1186,6 +1190,7 @@ def create_response_from_model(
     status=None,
     user_id=None,
     resource_id=None,
+    queue=None,
     iteration=None,
     process_log=None,
     progress=None,
@@ -1220,6 +1225,7 @@ def create_response_from_model(
         status (str): One of: accepted, running, finished, error
         user_id (str): The user id
         resource_id (str): The resource id
+        queue (str): The name of the queue
         iteration (int): The iteration of the job
         process_log (dict, str, list): The log from the running GRASS module
         progress (ProgressInfoModel): Progress information
@@ -1258,6 +1264,7 @@ def create_response_from_model(
         status=status,
         user_id=user_id,
         resource_id=resource_id,
+        queue=str(queue),
         # iteration=iteration,
         accept_timestamp=orig_time,
         accept_datetime=orig_datetime,

src/actinia_core/processing/actinia_processing/ephemeral_processing.py

@@ -292,6 +292,7 @@ def _send_resource_update(self, message, results=None):
             status="running",
             user_id=self.user_id,
             resource_id=self.resource_id,
+            queue=self.rdc.queue,
             iteration=self.iteration,
             # process_log=self.module_output_log,
             progress=self.progress,
@@ -319,6 +320,7 @@ def _send_resource_finished(self, message, results=None):
             status="finished",
             user_id=self.user_id,
             resource_id=self.resource_id,
+            queue=self.rdc.queue,
             iteration=self.iteration,
             process_log=self.module_output_log,
             progress=self.progress,
@@ -348,6 +350,7 @@ def _send_resource_terminated(self, message, results=None):
             status="terminated",
             user_id=self.user_id,
             resource_id=self.resource_id,
+            queue=self.rdc.queue,
             iteration=self.iteration,
             process_log=self.module_output_log,
             progress=self.progress,
@@ -376,6 +379,7 @@ def _send_resource_time_limit_exceeded(self, message, results=None):
             status="terminated",
             user_id=self.user_id,
             resource_id=self.resource_id,
+            queue=self.rdc.queue,
             iteration=self.iteration,
             process_log=self.module_output_log,
             progress=self.progress,
@@ -404,6 +408,7 @@ def _send_resource_error(self, message, results=None, exception=None):
             status="error",
             user_id=self.user_id,
             resource_id=self.resource_id,
+            queue=self.rdc.queue,
             iteration=self.iteration,
             process_log=self.module_output_log,
             progress=self.progress,

src/actinia_core/rest/base/resource_base.py

@@ -127,6 +127,16 @@ def __init__(self, resource_id=None, iteration=None, post_url=None):
             self.resource_id = resource_id
             self.request_id = self.generate_request_id_from_resource_id()
 
+        if global_config.QUEUE_TYPE == "per_job":
+            self.queue = "%s_%s" % (
+                global_config.WORKER_QUEUE_PREFIX,
+                self.resource_id,
+            )
+        elif global_config.QUEUE_TYPE == "redis":
+            self.queue = "%s_%s" % (global_config.WORKER_QUEUE_PREFIX, "count")
+        else:
+            self.queue = "local"
+
         # set iteration and post_url
         self.iteration = iteration
         self.post_url = post_url
@@ -186,6 +196,7 @@ def create_error_response(self, message, status="error", http_code=400):
             status=status,
             user_id=self.user_id,
             resource_id=self.resource_id,
+            queue=self.queue,
             iteration=self.iteration,
             process_log=None,
             results={},
@@ -360,6 +371,7 @@ def preprocess(
             status="accepted",
             user_id=self.user_id,
             resource_id=self.resource_id,
+            queue=self.queue,
             iteration=self.iteration,
             process_log=None,
             results={},