Workflow endpoint of the Processing Server

ocrd/ocrd/decorators/__init__.py

@@ -39,6 +39,7 @@ def ocrd_cli_wrap_processor(
     subcommand=None,
     address=None,
     queue=None,
+    log_filename=None,
     database=None,
     # ocrd_network params end #
     **kwargs

ocrd/ocrd/decorators/ocrd_cli_options.py

@@ -48,6 +48,7 @@ def cli(mets_url):
         option('-D', '--dump-module-dir', is_flag=True, default=False),
         option('-h', '--help', is_flag=True, default=False),
         option('-V', '--version', is_flag=True, default=False),
+        option('--log-filename', default=None),
         # Subcommand, only used for 'worker'/'server'. Cannot be handled in
         # click because processors use the @command decorator and even if they
         # were using `group`, you cannot combine have a command with

ocrd/ocrd/lib.bash

@@ -145,6 +145,7 @@ ocrd__parse_argv () {
             -I|--input-file-grp) ocrd__argv[input_file_grp]=$2 ; shift ;;
             -w|--working-dir) ocrd__argv[working_dir]=$(realpath "$2") ; shift ;;
             -m|--mets) ocrd__argv[mets_file]=$(realpath "$2") ; shift ;;
+            --log-filename) ocrd__argv[log_filename]="$2" ; shift ;;
             --mets-server-url) ocrd_argv[mets_server_url]="$2" ; shift ;;
             --overwrite) ocrd__argv[overwrite]=true ;;
             --profile) ocrd__argv[profile]=true ;;
@@ -168,7 +169,7 @@ ocrd__parse_argv () {
             if ! [ -v ocrd__worker_queue ]; then
                 ocrd__raise "For the Processing Worker --queue is required"
             fi
-            ocrd network processing-worker $OCRD_TOOL_NAME --queue "${ocrd__worker_queue}" --database "${ocrd__worker_database}"
+            ocrd network processing-worker $OCRD_TOOL_NAME --queue "${ocrd__worker_queue}" --database "${ocrd__worker_database}" --log-filename "${ocrd__argv[log_filename]}"
         elif [ ${ocrd__subcommand} = "server" ]; then
             if ! [ -v ocrd__worker_address ]; then
                 ocrd__raise "For the Processor Server --address is required"

ocrd/ocrd/mets_server.py

@@ -198,7 +198,7 @@ def shutdown(self):
         _exit(0)
 
     def startup(self):
-        self.log.info("Starting down METS server")
+        self.log.info("Starting up METS server")
 
         workspace = self.workspace
 

ocrd/ocrd/processor/base.py

@@ -353,7 +353,11 @@ def zip_input_files(self, require_first=True, mimetype=None, on_error='skip'):
             # Warn if no files found but pageId was specified because that
             # might be because of invalid page_id (range)
             if self.page_id and not files_:
-                LOG.warning(f"Could not find any files for --page-id {self.page_id} - compare '{self.page_id}' with the output of 'orcd workspace list-page'.")
+                msg = (f"Could not find any files for --page-id {self.page_id} - "
+                       f"compare '{self.page_id}' with the output of 'orcd workspace list-page'.")
+                if on_error == 'abort':
+                    raise ValueError(msg)
+                LOG.warning(msg)
             for file_ in files_:
                 if not file_.pageId:
                     continue

ocrd/ocrd/processor/helpers.py

@@ -251,6 +251,8 @@ def generate_processor_help(ocrd_tool, processor_instance=None, subcommand=None)
   --database                      The MongoDB server address in format
                                   "mongodb://{host}:{port}"
                                   [mongodb://localhost:27018]
+  --log-filename                  Filename to redirect STDOUT/STDERR to,
+                                  if specified.
 '''
 
     processing_server_options = '''\

ocrd/ocrd/workspace.py

@@ -123,7 +123,6 @@ def merge(self, other_workspace, copy_files=True, overwrite=False, **kwargs):
         """
         def after_add_cb(f):
             """callback to run on merged OcrdFile instances in the destination"""
-            print(f)
             if not f.local_filename:
                 # OcrdFile has no local_filename, so nothing to be copied
                 return
@@ -177,7 +176,6 @@ def download_file(self, f, _recursion_count=0):
         """
         log = getLogger('ocrd.workspace.download_file')
         with pushd_popd(self.directory):
-            print(f)
             if f.local_filename:
                 file_path = Path(f.local_filename).absolute()
                 if file_path.exists():

ocrd_models/ocrd_models/ocrd_mets.py

@@ -577,9 +577,9 @@ def physical_pages(self):
         if self._cache_flag:
             return list(self._page_cache.keys())
 
-        return self._tree.getroot().xpath(
+        return [str(x) for x in self._tree.getroot().xpath(
             'mets:structMap[@TYPE="PHYSICAL"]/mets:div[@TYPE="physSequence"]/mets:div[@TYPE="page"]/@ID',
-            namespaces=NS)
+            namespaces=NS)]
 
     def get_physical_pages(self, for_fileIds=None):
         """

ocrd_network/ocrd_network/__init__.py

@@ -1,27 +1,3 @@
-# This network package is supposed to contain all the packages and modules to realize the network architecture:
-# https://github.com/OCR-D/spec/pull/222/files#diff-8d0dae8c9277ff1003df93c5359c82a12d3f5c8452281f87781921921204d283
-
-# For reference, currently:
-# 1. The WebAPI is available here: https://github.com/OCR-D/ocrd-webapi-implementation
-# The ocrd-webapi-implementation repo implements the Discovery / Workflow / Workspace endpoints of the WebAPI currently.
-# This Processing Server PR implements just the Processor endpoint of the WebAPI. 
-# Once we have this merged to core under ocrd-network, the other endpoints will be adapted to ocrd-network 
-# and then the ocrd-webapi-implementation repo can be archived for reference.
-
-# 2. The RabbitMQ Library (i.e., utils) is used as an API to abstract and 
-# simplify (from the view point of processing server and workers) interactions with the RabbitMQ Server.
-# The library was adopted from: https://github.com/OCR-D/ocrd-webapi-implementation/tree/main/ocrd_webapi/rabbitmq
-
-# 3. Some potentially more useful code to be adopted for the Processing Server/Worker is available here:
-# https://github.com/OCR-D/core/pull/884
-# Update: Should be revisited again for adopting any relevant parts (if necessary). 
-# Nothing relevant is under the radar for now.
-
-# 4. The Mets Server discussion/implementation is available here:
-# https://github.com/OCR-D/core/pull/966
-
-# Note: The Mets Server is still not placed on the architecture diagram and probably won't be a part of
-# the network package. The reason, Mets Server is tightly coupled with the `OcrdWorkspace`.
 from .client import Client
 from .processing_server import ProcessingServer
 from .processing_worker import ProcessingWorker

ocrd_network/ocrd_network/database.py

@@ -13,10 +13,15 @@
 database (runs in docker) currently has no volume set.
 """
 from beanie import init_beanie
+from beanie.operators import In
 from motor.motor_asyncio import AsyncIOMotorClient
+from uuid import uuid4
+from pathlib import Path
+from typing import List
 
 from .models import (
     DBProcessorJob,
+    DBWorkflowJob,
     DBWorkspace
 )
 from .utils import call_sync
@@ -26,7 +31,7 @@ async def initiate_database(db_url: str):
     client = AsyncIOMotorClient(db_url)
     await init_beanie(
         database=client.get_default_database(default='ocrd'),
-        document_models=[DBProcessorJob, DBWorkspace]
+        document_models=[DBProcessorJob, DBWorkflowJob, DBWorkspace]
     )
 
 
@@ -35,6 +40,25 @@ async def sync_initiate_database(db_url: str):
     await initiate_database(db_url)
 
 
+async def db_create_workspace(mets_path: str) -> DBWorkspace:
+    """ Create a workspace-database entry only from a mets-path
+    """
+    if not Path(mets_path).exists():
+        raise FileNotFoundError(f'Cannot create DB workspace entry, `{mets_path}` does not exist!')
+    try:
+        return await db_get_workspace(workspace_mets_path=mets_path)
+    except ValueError:
+        workspace_db = DBWorkspace(
+            workspace_id=str(uuid4()),
+            workspace_path=Path(mets_path).parent,
+            workspace_mets_path=mets_path,
+            ocrd_identifier="",
+            bagit_profile_identifier="",
+        )
+        await workspace_db.save()
+        return workspace_db
+
+
 async def db_get_workspace(workspace_id: str = None, workspace_mets_path: str = None) -> DBWorkspace:
     workspace = None
     if not workspace_id and not workspace_mets_path:
@@ -59,7 +83,7 @@ async def sync_db_get_workspace(workspace_id: str = None, workspace_mets_path: s
     return await db_get_workspace(workspace_id=workspace_id, workspace_mets_path=workspace_mets_path)
 
 
-async def db_update_workspace(workspace_id: str = None, workspace_mets_path: str = None, **kwargs):
+async def db_update_workspace(workspace_id: str = None, workspace_mets_path: str = None, **kwargs) -> DBWorkspace:
     workspace = None
     if not workspace_id and not workspace_mets_path:
         raise ValueError(f'Either `workspace_id` or `workspace_mets_path` field must be used as a search key')
@@ -96,16 +120,17 @@ async def db_update_workspace(workspace_id: str = None, workspace_mets_path: str
             workspace.bag_info_adds = value
         elif key == 'deleted':
             workspace.deleted = value
-        elif key == 'pages_locked':
-            workspace.pages_locked = value
+        elif key == 'mets_server_url':
+            workspace.mets_server_url = value
         else:
             raise ValueError(f'Field "{key}" is not updatable.')
     await workspace.save()
+    return workspace
 
 
 @call_sync
-async def sync_db_update_workspace(workspace_id: str = None, workspace_mets_path: str = None, **kwargs):
-    await db_update_workspace(workspace_id=workspace_id, workspace_mets_path=workspace_mets_path, **kwargs)
+async def sync_db_update_workspace(workspace_id: str = None, workspace_mets_path: str = None, **kwargs) -> DBWorkspace:
+    return await db_update_workspace(workspace_id=workspace_id, workspace_mets_path=workspace_mets_path, **kwargs)
 
 
 async def db_get_processing_job(job_id: str) -> DBProcessorJob:
@@ -121,7 +146,7 @@ async def sync_db_get_processing_job(job_id: str) -> DBProcessorJob:
     return await db_get_processing_job(job_id)
 
 
-async def db_update_processing_job(job_id: str, **kwargs):
+async def db_update_processing_job(job_id: str, **kwargs) -> DBProcessorJob:
     job = await DBProcessorJob.find_one(
         DBProcessorJob.job_id == job_id)
     if not job:
@@ -144,8 +169,31 @@ async def db_update_processing_job(job_id: str, **kwargs):
         else:
             raise ValueError(f'Field "{key}" is not updatable.')
     await job.save()
+    return job
+
+
+@call_sync
+async def sync_db_update_processing_job(job_id: str, **kwargs) -> DBProcessorJob:
+    return await db_update_processing_job(job_id=job_id, **kwargs)
+
+
+async def db_get_workflow_job(job_id: str) -> DBWorkflowJob:
+    job = await DBWorkflowJob.find_one(DBWorkflowJob.job_id == job_id)
+    if not job:
+        raise ValueError(f'Workflow job with id "{job_id}" not in the DB.')
+    return job
+
+
+@call_sync
+async def sync_db_get_workflow_job(job_id: str) -> DBWorkflowJob:
+    return await db_get_workflow_job(job_id)
+
+
+async def db_get_processing_jobs(job_ids: List[str]) -> [DBProcessorJob]:
+    jobs = await DBProcessorJob.find(In(DBProcessorJob.job_id, job_ids)).to_list()
+    return jobs
 
 
 @call_sync
-async def sync_db_update_processing_job(job_id: str, **kwargs):
-    await db_update_processing_job(job_id=job_id, **kwargs)
+async def sync_db_get_processing_jobs(job_ids: List[str]) -> [DBProcessorJob]:
+    return await db_get_processing_jobs(job_ids)

ocrd_network/ocrd_network/deployer.py

@@ -10,9 +10,11 @@
 from typing import Dict, List, Union
 from re import search as re_search
 from os import getpid
+from pathlib import Path
+import subprocess
 from time import sleep
 
-from ocrd_utils import getLogger
+from ocrd_utils import getLogger, safe_filename
 
 from .deployment_utils import (
     create_docker_client,
@@ -28,7 +30,11 @@
     DataProcessorServer,
     DataRabbitMQ
 )
-from .utils import validate_and_load_config
+from .utils import (
+    is_mets_server_running,
+    stop_mets_server,
+    validate_and_load_config
+)
 
 
 class Deployer:
@@ -42,6 +48,7 @@ def __init__(self, config_path: str) -> None:
         self.internal_callback_url = config.get('internal_callback_url', None)
         for config_host in config['hosts']:
             self.data_hosts.append(DataHost(config_host))
+        self.mets_servers: Dict = {}  # {"mets_server_url": "mets_server_pid"}
 
     # TODO: Reconsider this.
     def find_matching_processors(
@@ -467,9 +474,12 @@ def start_native_processor(
         # printed with `echo $!` but it is printed inbetween other output. Because of that I added
         # `xyz` before and after the code to easily be able to filter out the pid via regex when
         # returning from the function
-        log_path = '/tmp/ocrd-processing-server-startup.log'
-        stdin.write(f"echo starting processing worker with '{cmd}' >> '{log_path}'\n")
-        stdin.write(f'{cmd} >> {log_path} 2>&1 &\n')
+
+        # TODO: Check here again
+        # log_path = f'/tmp/deployed_{processor_name}.log'
+        # stdin.write(f"echo starting processing worker with '{cmd}' >> '{log_path}'\n")
+        # stdin.write(f'{cmd} >> {log_path} 2>&1 &\n')
+        stdin.write(f'{cmd} &\n')
         stdin.write('echo xyz$!xyz \n exit \n')
         output = stdout.read().decode('utf-8')
         stdout.close()
@@ -505,13 +515,58 @@ def start_native_processor_server(
         channel = ssh_client.invoke_shell()
         stdin, stdout = channel.makefile('wb'), channel.makefile('rb')
         cmd = f'{processor_name} server --address {agent_address} --database {database_url}'
-        port = agent_address.split(':')[1]
-        log_path = f'/tmp/server_{processor_name}_{port}_{getpid()}.log'
-        # TODO: This entire stdin/stdout thing is broken with servers!
-        stdin.write(f"echo starting processor server with '{cmd}' >> '{log_path}'\n")
-        stdin.write(f'{cmd} >> {log_path} 2>&1 &\n')
+        stdin.write(f"echo starting processor server with '{cmd}'\n")
+        stdin.write(f'{cmd} &\n')
         stdin.write('echo xyz$!xyz \n exit \n')
         output = stdout.read().decode('utf-8')
         stdout.close()
         stdin.close()
         return re_search(r'xyz([0-9]+)xyz', output).group(1)  # type: ignore
+
+    # TODO: No support for TCP version yet
+    def start_unix_mets_server(self, mets_path: str) -> str:
+        socket_file = f'{safe_filename(mets_path)}.sock'
+        log_path = f'/tmp/{safe_filename(mets_path)}.log'
+        mets_server_url = f'/tmp/{socket_file}'
+
+        if is_mets_server_running(mets_server_url=mets_server_url):
+            self.log.info(f"The mets server is already started: {mets_server_url}")
+            return mets_server_url
+
+        cwd = Path(mets_path).parent
+        self.log.info(f'Starting UDS mets server: {mets_server_url}')
+        sub_process = subprocess.Popen(
+            args=['nohup', 'ocrd', 'workspace', '--mets-server-url', f'{mets_server_url}',
+                  '-d', f'{cwd}', 'server', 'start'],
+            shell=False,
+            stdout=open(log_path, 'w'),
+            stderr=open(log_path, 'a'),
+            cwd=cwd,
+            universal_newlines=True
+        )
+        # Wait for the mets server to start
+        sleep(2)
+        self.mets_servers[mets_server_url] = sub_process.pid
+        return mets_server_url
+
+    def stop_unix_mets_server(self, mets_server_url: str) -> None:
+        self.log.info(f'Stopping UDS mets server: {mets_server_url}')
+        if mets_server_url in self.mets_servers:
+            mets_server_pid = self.mets_servers[mets_server_url]
+        else:
+            raise Exception(f"Mets server not found: {mets_server_url}")
+
+        '''
+        subprocess.run(
+            args=['kill', '-s', 'SIGINT', f'{mets_server_pid}'],
+            shell=False,
+            universal_newlines=True
+        )
+        '''
+
+        # TODO: Reconsider this again
+        #  Not having this sleep here causes connection errors
+        #  on the last request processed by the processing worker.
+        #  Sometimes 3 seconds is enough, sometimes not.
+        sleep(5)
+        stop_mets_server(mets_server_url=mets_server_url)

ocrd_network/ocrd_network/deployment_utils.py

@@ -13,7 +13,8 @@
     'create_docker_client',
     'create_ssh_client',
     'DeployType',
-    'wait_for_rabbitmq_availability'
+    'verify_mongodb_available',
+    'verify_rabbitmq_available'
 ]
 
 

ocrd_network/ocrd_network/models/__init__.py

@@ -5,18 +5,22 @@
 
 __all__ = [
     'DBProcessorJob',
+    'DBWorkflowJob',
     'DBWorkspace',
     'PYJobInput',
     'PYJobOutput',
     'PYOcrdTool',
     'PYResultMessage',
+    'PYWorkflowJobOutput',
     'StateEnum',
 ]
 
 from .job import (
     DBProcessorJob,
+    DBWorkflowJob,
     PYJobInput,
     PYJobOutput,
+    PYWorkflowJobOutput,
     StateEnum
 )
 from .messages import PYResultMessage