Standalone Processor Server (#884 + #974)

ocrd/ocrd/cli/__init__.py

@@ -31,8 +31,7 @@ def get_help(self, ctx):
 from ocrd.decorators import ocrd_loglevel
 from .zip import zip_cli
 from .log import log_cli
-from .processing_server import processing_server_cli
-from .processing_worker import processing_worker_cli
+from .network import network_cli
 
 
 @click.group()
@@ -51,5 +50,4 @@ def cli(**kwargs): # pylint: disable=unused-argument
 cli.add_command(validate_cli)
 cli.add_command(log_cli)
 cli.add_command(resmgr_cli)
-cli.add_command(processing_server_cli)
-cli.add_command(processing_worker_cli)
+cli.add_command(network_cli)

ocrd/ocrd/cli/network.py

@@ -0,0 +1,34 @@
+"""
+OCR-D CLI: management of network components
+
+.. click:: ocrd.cli.network:network_cli
+    :prog: ocrd network
+    :nested: full
+"""
+
+import click
+import logging
+from ocrd_utils import initLogging
+from ocrd_network.cli import (
+    client_cli,
+    processing_server_cli,
+    processing_worker_cli,
+    processor_server_cli,
+)
+
+
+@click.group("network")
+def network_cli():
+    """
+    Managing network components
+    """
+    initLogging()
+    # TODO: Remove after the logging fix in core
+    logging.getLogger('paramiko.transport').setLevel(logging.INFO)
+    logging.getLogger('ocrd_network').setLevel(logging.DEBUG)
+
+
+network_cli.add_command(client_cli)
+network_cli.add_command(processing_server_cli)
+network_cli.add_command(processing_worker_cli)
+network_cli.add_command(processor_server_cli)

ocrd/ocrd/decorators/__init__.py

@@ -1,10 +1,5 @@
-from os.path import isfile
 from os import environ
 import sys
-from contextlib import redirect_stdout
-from io import StringIO
-
-import click
 
 from ocrd_utils import (
     is_local_filename,
@@ -15,7 +10,7 @@
 from ocrd_utils import getLogger, initLogging, parse_json_string_with_comments
 from ocrd_validators import WorkspaceValidator
 
-from ocrd_network import ProcessingWorker
+from ocrd_network import ProcessingWorker, ProcessorServer
 
 from ..resolver import Resolver
 from ..processor.base import run_processor
@@ -38,8 +33,12 @@ def ocrd_cli_wrap_processor(
     overwrite=False,
     show_resource=None,
     list_resources=False,
+    # ocrd_network params start #
+    agent_type=None,
+    agent_address=None,
     queue=None,
     database=None,
+    # ocrd_network params end #
     **kwargs
 ):
     if not sys.argv[1:]:
@@ -56,96 +55,130 @@ def ocrd_cli_wrap_processor(
             list_resources=list_resources
         )
         sys.exit()
-    # If either of these two is provided but not both
-    if bool(queue) != bool(database):
-        raise Exception("Options --queue and --database require each other.")
-    # If both of these are provided - start the processing worker instead of the processor - processorClass
-    if queue and database:
-        initLogging()
-        # TODO: Remove before the release
-        # We are importing the logging here because it's not the ocrd logging but python one
-        import logging
-        logging.getLogger('ocrd.network').setLevel(logging.DEBUG)
 
-        # Get the ocrd_tool dictionary
-        processor = processorClass(workspace=None, dump_json=True)
-        ocrd_tool = processor.ocrd_tool
+    initLogging()
+
+    # Used for checking/starting network agents for the WebAPI architecture
+    # Has no side effects if neither of the 4 ocrd_network parameters are passed
+    check_and_run_network_agent(processorClass, agent_type, agent_address, database, queue)
+
+    LOG = getLogger('ocrd_cli_wrap_processor')
+    # LOG.info('kwargs=%s' % kwargs)
+    # Merge parameter overrides and parameters
+    if 'parameter_override' in kwargs:
+        set_json_key_value_overrides(kwargs['parameter'], *kwargs['parameter_override'])
+    # TODO OCR-D/core#274
+    # Assert -I / -O
+    # if not kwargs['input_file_grp']:
+    #     raise ValueError('-I/--input-file-grp is required')
+    # if not kwargs['output_file_grp']:
+    #     raise ValueError('-O/--output-file-grp is required')
+    resolver = Resolver()
+    working_dir, mets, _ = resolver.resolve_mets_arguments(working_dir, mets, None)
+    workspace = resolver.workspace_from_url(mets, working_dir)
+    page_id = kwargs.get('page_id')
+    # XXX not possible while processors do not adhere to # https://github.com/OCR-D/core/issues/505
+    # if overwrite
+    #     if 'output_file_grp' not in kwargs or not kwargs['output_file_grp']:
+    #         raise Exception("--overwrite requires --output-file-grp")
+    #     LOG.info("Removing files because of --overwrite")
+    #     for grp in kwargs['output_file_grp'].split(','):
+    #         if page_id:
+    #             for one_page_id in kwargs['page_id'].split(','):
+    #                 LOG.debug("Removing files in output file group %s with page ID %s", grp, one_page_id)
+    #                 for file in workspace.mets.find_files(pageId=one_page_id, fileGrp=grp):
+    #                     workspace.remove_file(file, force=True, keep_file=False, page_recursive=True)
+    #         else:
+    #             LOG.debug("Removing all files in output file group %s ", grp)
+    #             # TODO: can be reduced to `page_same_group=True` as soon as core#505 has landed (in all processors)
+    #             workspace.remove_file_group(grp, recursive=True, force=True, keep_files=False, page_recursive=True, page_same_group=False)
+    #     workspace.save_mets()
+    # XXX While https://github.com/OCR-D/core/issues/505 is open, set 'overwrite_mode' globally on the workspace
+    if overwrite:
+        workspace.overwrite_mode = True
+    report = WorkspaceValidator.check_file_grp(workspace, kwargs['input_file_grp'], '' if overwrite else kwargs['output_file_grp'], page_id)
+    if not report.is_valid:
+        raise Exception("Invalid input/output file grps:\n\t%s" % '\n\t'.join(report.errors))
+    # Set up profiling behavior from environment variables/flags
+    if not profile and 'OCRD_PROFILE' in environ:
+        if 'CPU' in environ['OCRD_PROFILE']:
+            profile = True
+    if not profile_file and 'OCRD_PROFILE_FILE' in environ:
+        profile_file = environ['OCRD_PROFILE_FILE']
+    if profile or profile_file:
+        import cProfile
+        import pstats
+        import io
+        import atexit
+        print("Profiling...")
+        pr = cProfile.Profile()
+        pr.enable()
+        def exit():
+            pr.disable()
+            print("Profiling completed")
+            if profile_file:
+                with open(profile_file, 'wb') as f:
+                    pr.dump_stats(profile_file)
+            s = io.StringIO()
+            pstats.Stats(pr, stream=s).sort_stats("cumulative").print_stats()
+            print(s.getvalue())
+        atexit.register(exit)
+    run_processor(processorClass, mets_url=mets, workspace=workspace, **kwargs)
+
+
+def check_and_run_network_agent(ProcessorClass, agent_type: str, agent_address: str, database: str, queue: str):
+    if not agent_type and (agent_address or database or queue):
+        raise ValueError("Options '--database', '--queue', and '--address' are valid only with '--type'")
+    if not agent_type:
+        return
+
+    if not database:
+        raise ValueError("Options '--type' and '--database' are mutually inclusive")
+    allowed_agent_types = ['server', 'worker']
+    if agent_type not in allowed_agent_types:
+        agents_str = ', '.join(allowed_agent_types)
+        raise ValueError(f"Wrong type parameter. Allowed types: {agents_str}")
+    if agent_type == 'server':
+        if not agent_address:
+            raise ValueError("Options '--type=server' and '--address' are mutually inclusive")
+        if queue:
+            raise ValueError("Options '--type=server' and '--queue' are mutually exclusive")
+    if agent_type == 'worker':
+        if not queue:
+            raise ValueError("Options '--type=worker' and '--queue' are mutually inclusive")
+        if agent_address:
+            raise ValueError("Options '--type=worker' and '--address' are mutually exclusive")
 
+    import logging
+    logging.getLogger('ocrd.network').setLevel(logging.DEBUG)
+
+    processor = ProcessorClass(workspace=None, dump_json=True)
+    if agent_type == 'worker':
         try:
+            # TODO: Passing processor_name and ocrd_tool is reduntant
             processing_worker = ProcessingWorker(
                 rabbitmq_addr=queue,
                 mongodb_addr=database,
-                processor_name=ocrd_tool['executable'],
-                ocrd_tool=ocrd_tool,
-                processor_class=processorClass,
+                processor_name=processor.ocrd_tool['executable'],
+                ocrd_tool=processor.ocrd_tool,
+                processor_class=ProcessorClass,
             )
             # The RMQConsumer is initialized and a connection to the RabbitMQ is performed
             processing_worker.connect_consumer()
             # Start consuming from the queue with name `processor_name`
             processing_worker.start_consuming()
         except Exception as e:
-            raise Exception("Processing worker has failed with error") from e
-    else:
-        initLogging()
-        LOG = getLogger('ocrd_cli_wrap_processor')
-        # LOG.info('kwargs=%s' % kwargs)
-        # Merge parameter overrides and parameters
-        if 'parameter_override' in kwargs:
-            set_json_key_value_overrides(kwargs['parameter'], *kwargs['parameter_override'])
-        # TODO OCR-D/core#274
-        # Assert -I / -O
-        # if not kwargs['input_file_grp']:
-        #     raise ValueError('-I/--input-file-grp is required')
-        # if not kwargs['output_file_grp']:
-        #     raise ValueError('-O/--output-file-grp is required')
-        resolver = Resolver()
-        working_dir, mets, _ = resolver.resolve_mets_arguments(working_dir, mets, None)
-        workspace = resolver.workspace_from_url(mets, working_dir)
-        page_id = kwargs.get('page_id')
-        # XXX not possible while processors do not adhere to # https://github.com/OCR-D/core/issues/505
-        # if overwrite
-        #     if 'output_file_grp' not in kwargs or not kwargs['output_file_grp']:
-        #         raise Exception("--overwrite requires --output-file-grp")
-        #     LOG.info("Removing files because of --overwrite")
-        #     for grp in kwargs['output_file_grp'].split(','):
-        #         if page_id:
-        #             for one_page_id in kwargs['page_id'].split(','):
-        #                 LOG.debug("Removing files in output file group %s with page ID %s", grp, one_page_id)
-        #                 for file in workspace.mets.find_files(pageId=one_page_id, fileGrp=grp):
-        #                     workspace.remove_file(file, force=True, keep_file=False, page_recursive=True)
-        #         else:
-        #             LOG.debug("Removing all files in output file group %s ", grp)
-        #             # TODO: can be reduced to `page_same_group=True` as soon as core#505 has landed (in all processors)
-        #             workspace.remove_file_group(grp, recursive=True, force=True, keep_files=False, page_recursive=True, page_same_group=False)
-        #     workspace.save_mets()
-        # XXX While https://github.com/OCR-D/core/issues/505 is open, set 'overwrite_mode' globally on the workspace
-        if overwrite:
-            workspace.overwrite_mode = True
-        report = WorkspaceValidator.check_file_grp(workspace, kwargs['input_file_grp'], '' if overwrite else kwargs['output_file_grp'], page_id)
-        if not report.is_valid:
-            raise Exception("Invalid input/output file grps:\n\t%s" % '\n\t'.join(report.errors))
-        # Set up profiling behavior from environment variables/flags
-        if not profile and 'OCRD_PROFILE' in environ:
-            if 'CPU' in environ['OCRD_PROFILE']:
-                profile = True
-        if not profile_file and 'OCRD_PROFILE_FILE' in environ:
-            profile_file = environ['OCRD_PROFILE_FILE']
-        if profile or profile_file:
-            import cProfile
-            import pstats
-            import io
-            import atexit
-            print("Profiling...")
-            pr = cProfile.Profile()
-            pr.enable()
-            def exit():
-                pr.disable()
-                print("Profiling completed")
-                if profile_file:
-                    with open(profile_file, 'wb') as f:
-                        pr.dump_stats(profile_file)
-                s = io.StringIO()
-                pstats.Stats(pr, stream=s).sort_stats("cumulative").print_stats()
-                print(s.getvalue())
-            atexit.register(exit)
-        run_processor(processorClass, mets_url=mets, workspace=workspace, **kwargs)
+            sys.exit(f"Processing worker has failed with error: {e}")
+    if agent_type == 'server':
+        try:
+            # TODO: Better validate that inside the ProcessorServer itself
+            host, port = agent_address.split(':')
+            processor_server = ProcessorServer(
+                mongodb_addr=database,
+                processor_name=processor.ocrd_tool['executable'],
+                processor_class=ProcessorClass,
+            )
+            processor_server.run_server(host=host, port=int(port))
+        except Exception as e:
+            sys.exit(f"Processor server has failed with error: {e}")
+    sys.exit(0)

ocrd/ocrd/decorators/ocrd_cli_options.py

@@ -1,7 +1,12 @@
+import click
 from click import option, Path
 from .parameter_option import parameter_option, parameter_override_option
 from .loglevel_option import loglevel_option
-from ocrd_network import QueueServerParamType, DatabaseParamType
+from ocrd_network import (
+    DatabaseParamType,
+    ServerAddressParamType,
+    QueueServerParamType
+)
 
 
 def ocrd_cli_options(f):
@@ -33,6 +38,8 @@ def cli(mets_url):
         parameter_option,
         parameter_override_option,
         loglevel_option,
+        option('--type', 'agent_type', type=click.Choice(['worker', 'server'])),
+        option('--address', 'agent_address', type=ServerAddressParamType()),
         option('--queue', type=QueueServerParamType()),
         option('--database', type=DatabaseParamType()),
         option('-C', '--show-resource'),
@@ -45,4 +52,3 @@ def cli(mets_url):
     for param in params:
         param(f)
     return f
-