Processing server extension (#1046)

ocrd_network/ocrd_network/models/job.py

@@ -37,6 +37,8 @@ class PYJobInput(BaseModel):
     agent_type: Optional[str] = 'worker'
     # Auto generated by the Processing Server when forwarding to the Processor Server
     job_id: Optional[str] = None
+    # If set, specifies a list of job ids this job depends on
+    depends_on: Optional[List[str]] = None
 
     class Config:
         schema_extra = {
@@ -74,6 +76,7 @@ class DBProcessorJob(Document):
     output_file_grps: Optional[List[str]]
     page_id: Optional[str]
     parameters: Optional[dict]
+    depends_on: Optional[List[str]]
     result_queue_name: Optional[str]
     callback_url: Optional[str]
     internal_callback_url: Optional[str]

ocrd_network/ocrd_network/processing_server.py

@@ -3,7 +3,6 @@
 import httpx
 from typing import Dict, List, Optional
 import uvicorn
-from queue import Queue
 
 from fastapi import FastAPI, status, Request, HTTPException
 from fastapi.exceptions import RequestValidationError
@@ -22,6 +21,7 @@
     DBProcessorJob,
     PYJobInput,
     PYJobOutput,
+    PYResultMessage,
     StateEnum
 )
 from .rabbitmq_utils import (
@@ -83,6 +83,9 @@ def __init__(self, config_path: str, host: str, port: int) -> None:
         # Value: Queue that holds PYInputJob elements
         self.processing_requests_cache = {}
 
+        # Used by processing workers and/or processor servers to report back the results
+        self.internal_job_callback_url = f'http://{host}:{port}/processor/result_callback'
+
         # Create routes
         self.router.add_api_route(
             path='/stop',
@@ -249,6 +252,7 @@ def create_processing_message(job: DBProcessorJob) -> OcrdProcessingMessage:
             parameters=job.parameters,
             result_queue_name=job.result_queue_name,
             callback_url=job.callback_url,
+            internal_callback_url=job.internal_callback_url
         )
         return processing_message
 
@@ -266,6 +270,34 @@ def check_if_queue_exists(self, processor_name):
                 detail=f"Process queue with id '{processor_name}' not existing"
             )
 
+    # Returns true if all dependent jobs' states are success, else false
+    @staticmethod
+    async def check_if_job_dependencies_met(dependencies: List[str]) -> bool:
+        # Check the states of all dependent jobs
+        for dependency_job_id in dependencies:
+            dependency_job_state = (await db_get_processing_job(dependency_job_id)).state
+            # Found a dependent job whose state is not success
+            if dependency_job_state != StateEnum.success:
+                return False
+        return True
+
+    async def find_next_request_from_internal_queue(self, internal_queue: List[PYJobInput]) -> PYJobInput:
+        found_index = None
+        found_request = None
+        for i in range(0, len(internal_queue)):
+            current_element = internal_queue[i]
+            # Request has other job dependencies
+            if current_element.depends_on:
+                if not await self.check_if_job_dependencies_met(current_element.depends_on):
+                    continue
+            found_index = i
+            break
+
+        if found_index:
+            # Consume the request from the internal queue
+            found_request = internal_queue.pop(found_index)
+        return found_request
+
     def query_ocrd_tool_json_from_server(self, processor_name):
         processor_server_url = self.deployer.resolve_processor_server_url(processor_name)
         if not processor_server_url:
@@ -323,17 +355,24 @@ async def push_processor_job(self, processor_name: str, data: PYJobInput) -> PYJ
         # is a page_id value that has been previously locked
         cache_current_request = False
 
-        # Check if there are any locked pages for the current request
+        # Check if there are any dependencies of the current request
+        if data.depends_on:
+            if not await self.check_if_job_dependencies_met(data.depends_on):
+                cache_current_request = True
+
         locked_ws_pages = workspace_db.pages_locked
-        for output_fileGrp in data.output_file_grps:
-            if output_fileGrp in locked_ws_pages:
-                if "all_pages" in locked_ws_pages[output_fileGrp]:
-                    cache_current_request = True
-                    break
-                # If there are request page ids that are already locked
-                if not locked_ws_pages[output_fileGrp].isdisjoint(page_ids):
-                    cache_current_request = True
-                    break
+        # No need for further check if the request should be cached
+        if not cache_current_request:
+            # Check if there are any locked pages for the current request
+            for output_fileGrp in data.output_file_grps:
+                if output_fileGrp in locked_ws_pages:
+                    if "all_pages" in locked_ws_pages[output_fileGrp]:
+                        cache_current_request = True
+                        break
+                    # If there are request page ids that are already locked
+                    if not locked_ws_pages[output_fileGrp].isdisjoint(page_ids):
+                        cache_current_request = True
+                        break
 
         if cache_current_request:
             # Append the processor name to the request itself
@@ -342,9 +381,9 @@ async def push_processor_job(self, processor_name: str, data: PYJobInput) -> PYJ
             workspace_key = data.workspace_id if data.workspace_id else data.path_to_mets
             # If a record queue of this workspace_id does not exist in the requests cache
             if not self.processing_requests_cache.get(workspace_key, None):
-                self.processing_requests_cache[workspace_key] = Queue()
-            # Add the processing request to the internal queue
-            self.processing_requests_cache[workspace_key].put(data)
+                self.processing_requests_cache[workspace_key] = []
+            # Add the processing request to the end of the internal queue
+            self.processing_requests_cache[workspace_key].append(data)
 
             return PYJobOutput(
                 job_id=data.job_id,
@@ -376,7 +415,7 @@ async def push_processor_job(self, processor_name: str, data: PYJobInput) -> PYJ
         job = DBProcessorJob(
             **data.dict(exclude_unset=True, exclude_none=True),
             processor_name=processor_name,
-            internal_callback_url=f"/processor/result_callback",
+            internal_callback_url=self.internal_job_callback_url,
             state=StateEnum.queued
         )
         await job.insert()
@@ -504,28 +543,29 @@ async def remove_from_request_cache(self, job_id: str, state: StateEnum,
         workspace_key = workspace_id if workspace_id else path_to_mets
 
         if workspace_key not in self.processing_requests_cache:
-            # No internal queue available for that workspace
+            self.log.exception(f"No internal queue available for workspace with key: {workspace_key}")
             return
 
-        if self.processing_requests_cache[workspace_key].empty():
+        if not len(self.processing_requests_cache[workspace_key]):
             # The queue is empty - delete it
             try:
                 del self.processing_requests_cache[workspace_key]
             except KeyError as ex:
                 self.log.warning(f"Trying to delete non-existing internal queue with key: {workspace_key}")
             return
 
-        # Process the next request in the internal queue
-        # TODO: Refactor and optimize the duplications here
-        #  and last lines in `push_processor_job` method
-        data = self.processing_requests_cache[workspace_key].get()
-        processor_name = data.processor_name
+        data = await self.find_next_request_from_internal_queue(self.processing_requests_cache[workspace_key])
+        # Nothing was consumed from the internal queue
+        if not data:
+            self.log.exception(f"No data was consumed from the internal queue")
+            return
 
+        processor_name = data.processor_name
         # Create a DB entry
         job = DBProcessorJob(
             **data.dict(exclude_unset=True, exclude_none=True),
             processor_name=processor_name,
-            internal_callback_url=f"/processor/result_callback",
+            internal_callback_url=self.internal_job_callback_url,
             state=StateEnum.queued
         )
         await job.insert()

ocrd_network/ocrd_network/processing_worker.py

@@ -192,7 +192,7 @@ def process_message(self, processing_message: OcrdProcessingMessage) -> None:
         page_id = processing_message.page_id if 'page_id' in pm_keys else None
         result_queue_name = processing_message.result_queue_name if 'result_queue_name' in pm_keys else None
         callback_url = processing_message.callback_url if 'callback_url' in pm_keys else None
-        internal_callback_url = processing_message.callback_url if 'internal_callback_url' in pm_keys else None
+        internal_callback_url = processing_message.internal_callback_url if 'internal_callback_url' in pm_keys else None
         parameters = processing_message.parameters if processing_message.parameters else {}
 
         if not path_to_mets and workspace_id:
@@ -239,7 +239,7 @@ def process_message(self, processing_message: OcrdProcessingMessage) -> None:
             # May not be always available
             workspace_id=workspace_id
         )
-        self.log.info(f'Result message: {result_message}')
+        self.log.info(f'Result message: {str(result_message)}')
         # If the result_queue field is set, send the result message to a result queue
         if result_queue_name:
             self.publish_to_result_queue(result_queue_name, result_message)
@@ -265,18 +265,6 @@ def publish_to_result_queue(self, result_queue: str, result_message: OcrdResultM
             message=encoded_result_message
         )
 
-    def post_to_callback_url(self, callback_url: str, result_message: OcrdResultMessage):
-        self.log.info(f'Posting result message to callback_url "{callback_url}"')
-        headers = {"Content-Type": "application/json"}
-        json_data = {
-            "job_id": result_message.job_id,
-            "state": result_message.state,
-            "path_to_mets": result_message.path_to_mets,
-            "workspace_id": result_message.workspace_id
-        }
-        response = requests.post(url=callback_url, headers=headers, json=json_data)
-        self.log.info(f'Response from callback_url "{response}"')
-
     def create_queue(self, connection_attempts=1, retry_delay=1):
         """Create the queue for this worker
 

ocrd_network/ocrd_network/rabbitmq_utils/ocrd_messages.py

@@ -18,7 +18,8 @@ def __init__(
             page_id: Optional[str],
             result_queue_name: Optional[str],
             callback_url: Optional[str],
-            parameters: Dict[str, Any] = None,
+            internal_callback_url: Optional[str],
+            parameters: Dict[str, Any] = None
     ) -> None:
         if not job_id:
             raise ValueError('job_id must be provided')
@@ -47,6 +48,8 @@ def __init__(
             self.result_queue_name = result_queue_name
         if callback_url:
             self.callback_url = callback_url
+        if internal_callback_url:
+            self.internal_callback_url = internal_callback_url
         self.parameters = parameters if parameters else {}
 
     @staticmethod
@@ -71,7 +74,8 @@ def decode_yml(ocrd_processing_message: bytes) -> OcrdProcessingMessage:
             page_id=data.get('page_id', None),
             parameters=data.get('parameters', None),
             result_queue_name=data.get('result_queue_name', None),
-            callback_url=data.get('callback_url', None)
+            callback_url=data.get('callback_url', None),
+            internal_callback_url=data.get('internal_callback_url', None)
         )
 
 

ocrd_validators/ocrd_validators/message_processing.schema.yml

@@ -58,8 +58,13 @@ properties:
   callback_url:
     description: The URL where the result message will be POST-ed to
     type: string
-    format: uri,
-    pattern: "^https?://"
+    format: uri
+    pattern: "^http?://"
+  internal_callback_url:
+    description: The URL where the internal result message will be POST-ed to the Processing Server
+    type: string
+    format: uri
+    pattern: "^http?://"
   created_time:
     description: The Unix timestamp when the message was created
     type: integer

ocrd_validators/ocrd_validators/message_result.schema.yml

@@ -20,8 +20,10 @@ properties:
     description: The current status of the job
     type: string
     enum:
-      - SUCCESS
+      - CACHED
+      - QUEUED
       - RUNNING
+      - SUCCESS
       - FAILED
   path_to_mets:
     description: Path to a METS file