Source code for ocrd_network.processing_worker

Abstraction for the Processing Server unit in this arch:

Calls to native OCR-D processor should happen through
the Processing Worker wrapper to hide low level details.
According to the current requirements, each ProcessingWorker
is a single OCR-D Processor instance.

from datetime import datetime
from logging import FileHandler, Formatter
from os import getpid
from time import sleep
import pika.spec
import pika.adapters.blocking_connection
from pika.exceptions import AMQPConnectionError

from ocrd_utils import config, getLogger, LOG_FORMAT
from .database import (
from .logging import (
from .models import StateEnum
from .process_helpers import invoke_processor
from .rabbitmq_utils import (
from .utils import (

[docs]class ProcessingWorker: def __init__(self, rabbitmq_addr, mongodb_addr, processor_name, ocrd_tool: dict, processor_class=None) -> None: self.log = getLogger(f'ocrd_network.processing_worker') log_file = get_processing_worker_logging_file_path(processor_name=processor_name, pid=getpid()) file_handler = FileHandler(filename=log_file, mode='a') file_handler.setFormatter(Formatter(LOG_FORMAT)) self.log.addHandler(file_handler) try: verify_database_uri(mongodb_addr) self.log.debug(f'Verified MongoDB URL: {mongodb_addr}') rmq_data = verify_and_parse_mq_uri(rabbitmq_addr) self.rmq_username = rmq_data['username'] self.rmq_password = rmq_data['password'] self.rmq_host = rmq_data['host'] self.rmq_port = rmq_data['port'] self.rmq_vhost = rmq_data['vhost'] self.log.debug(f'Verified RabbitMQ Credentials: {self.rmq_username}:{self.rmq_password}') self.log.debug(f'Verified RabbitMQ Server URL: {self.rmq_host}:{self.rmq_port}{self.rmq_vhost}') except ValueError as e: raise ValueError(e) sync_initiate_database(mongodb_addr) # Database client self.ocrd_tool = ocrd_tool # The str name of the OCR-D processor instance to be started self.processor_name = processor_name # The processor class to be used to instantiate the processor # Think of this as a func pointer to the constructor of the respective OCR-D processor self.processor_class = processor_class # Gets assigned when `connect_consumer` is called on the worker object # Used to consume OcrdProcessingMessage from the queue with name {processor_name} self.rmq_consumer = None # Gets assigned when the `connect_publisher` is called on the worker object # The publisher is connected when the `result_queue` field of the OcrdProcessingMessage is set for first time # Used to publish OcrdResultMessage type message to the queue with name {processor_name}-result self.rmq_publisher = None # Always create a queue (idempotent) self.create_queue()
[docs] def connect_consumer(self) -> None:'Connecting RMQConsumer to RabbitMQ server: ' f'{self.rmq_host}:{self.rmq_port}{self.rmq_vhost}') self.rmq_consumer = RMQConsumer( host=self.rmq_host, port=self.rmq_port, vhost=self.rmq_vhost ) self.log.debug(f'RMQConsumer authenticates with username: ' f'{self.rmq_username}, password: {self.rmq_password}') self.rmq_consumer.authenticate_and_connect( username=self.rmq_username, password=self.rmq_password )'Successfully connected RMQConsumer.')
[docs] def connect_publisher(self, enable_acks: bool = True) -> None:'Connecting RMQPublisher to RabbitMQ server: ' f'{self.rmq_host}:{self.rmq_port}{self.rmq_vhost}') self.rmq_publisher = RMQPublisher( host=self.rmq_host, port=self.rmq_port, vhost=self.rmq_vhost ) self.log.debug(f'RMQPublisher authenticates with username: ' f'{self.rmq_username}, password: {self.rmq_password}') self.rmq_publisher.authenticate_and_connect( username=self.rmq_username, password=self.rmq_password ) if enable_acks: self.rmq_publisher.enable_delivery_confirmations()'Delivery confirmations are enabled')'Successfully connected RMQPublisher.')
# Define what happens every time a message is consumed # from the queue with name self.processor_name
[docs] def on_consumed_message( self, channel: pika.adapters.blocking_connection.BlockingChannel, delivery: pika.spec.Basic.Deliver, properties: pika.spec.BasicProperties, body: bytes) -> None: consumer_tag = delivery.consumer_tag delivery_tag: int = delivery.delivery_tag is_redelivered: bool = delivery.redelivered message_headers: dict = properties.headers self.log.debug(f'Consumer tag: {consumer_tag}, ' f'message delivery tag: {delivery_tag}, ' f'redelivered: {is_redelivered}') self.log.debug(f'Message headers: {message_headers}') try: self.log.debug(f'Trying to decode processing message with tag: {delivery_tag}') processing_message: OcrdProcessingMessage = OcrdProcessingMessage.decode_yml(body) except Exception as e: self.log.error(f'Failed to decode processing message body: {body}') self.log.error(f'Nacking processing message with tag: {delivery_tag}') channel.basic_nack(delivery_tag=delivery_tag, multiple=False, requeue=False) raise Exception(f'Failed to decode processing message with tag: {delivery_tag}, reason: {e}') try:'Starting to process the received message: {processing_message.__dict__}') self.process_message(processing_message=processing_message) except Exception as e: self.log.error(f'Failed to process processing message with tag: {delivery_tag}') self.log.error(f'Nacking processing message with tag: {delivery_tag}') channel.basic_nack(delivery_tag=delivery_tag, multiple=False, requeue=False) raise Exception(f'Failed to process processing message with tag: {delivery_tag}, reason: {e}')'Successfully processed RabbitMQ message') self.log.debug(f'Acking message with tag: {delivery_tag}') channel.basic_ack(delivery_tag=delivery_tag, multiple=False)
[docs] def start_consuming(self) -> None: if self.rmq_consumer:'Configuring consuming from queue: {self.processor_name}') self.rmq_consumer.configure_consuming( queue_name=self.processor_name, callback_method=self.on_consumed_message )'Starting consuming from queue: {self.processor_name}') # Starting consuming is a blocking action self.rmq_consumer.start_consuming() else: raise Exception('The RMQConsumer is not connected/configured properly')
# TODO: Better error handling required to catch exceptions
[docs] def process_message(self, processing_message: OcrdProcessingMessage) -> None: # Verify that the processor name in the processing message # matches the processor name of the current processing worker if self.processor_name != processing_message.processor_name: raise ValueError(f'Processor name is not matching. Expected: {self.processor_name},' f'Got: {processing_message.processor_name}') # All of this is needed because the OcrdProcessingMessage object # may not contain certain keys. Simply passing None in the OcrdProcessingMessage constructor # breaks the message validator schema which expects String, but not None due to the Optional[] wrapper. pm_keys = processing_message.__dict__.keys() job_id = processing_message.job_id input_file_grps = processing_message.input_file_grps output_file_grps = processing_message.output_file_grps if 'output_file_grps' in pm_keys else None path_to_mets = processing_message.path_to_mets if 'path_to_mets' in pm_keys else None workspace_id = processing_message.workspace_id if 'workspace_id' in pm_keys else 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.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 not workspace_id: raise ValueError(f'`path_to_mets` nor `workspace_id` was set in the ocrd processing message') if path_to_mets: mets_server_url = sync_db_get_workspace(workspace_mets_path=path_to_mets).mets_server_url if not path_to_mets and workspace_id: path_to_mets = sync_db_get_workspace(workspace_id).workspace_mets_path mets_server_url = sync_db_get_workspace(workspace_id).mets_server_url execution_failed = False self.log.debug(f'Invoking processor: {self.processor_name}') start_time = job_log_file = get_processing_job_logging_file_path(job_id=job_id) sync_db_update_processing_job( job_id=job_id, state=StateEnum.running, path_to_mets=path_to_mets, start_time=start_time, log_file_path=job_log_file ) try: invoke_processor( processor_class=self.processor_class, executable=self.processor_name, abs_path_to_mets=path_to_mets, input_file_grps=input_file_grps, output_file_grps=output_file_grps, page_id=page_id, log_filename=job_log_file, parameters=processing_message.parameters, mets_server_url=mets_server_url ) except Exception as error: self.log.debug(f"processor_name: {self.processor_name}, path_to_mets: {path_to_mets}, " f"input_grps: {input_file_grps}, output_file_grps: {output_file_grps}, " f"page_id: {page_id}, parameters: {parameters}") self.log.exception(error) execution_failed = True end_time = exec_duration = calculate_execution_time(start_time, end_time) job_state = StateEnum.success if not execution_failed else StateEnum.failed sync_db_update_processing_job( job_id=job_id, state=job_state, end_time=end_time, exec_time=f'{exec_duration} ms' ) result_message = OcrdResultMessage( job_id=job_id, state=job_state.value, path_to_mets=path_to_mets, # May not be always available workspace_id=workspace_id )'Result message: {result_message.__dict__}') # 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) if callback_url: # If the callback_url field is set, # post the result message (callback to a user defined endpoint) post_to_callback_url(self.log, callback_url, result_message) if internal_callback_url: # If the internal callback_url field is set, # post the result message (callback to Processing Server endpoint) post_to_callback_url(self.log, internal_callback_url, result_message)
[docs] def publish_to_result_queue(self, result_queue: str, result_message: OcrdResultMessage): if self.rmq_publisher is None: self.connect_publisher() # create_queue method is idempotent - nothing happens if # a queue with the specified name already exists self.rmq_publisher.create_queue(queue_name=result_queue)'Publishing result message to queue: {result_queue}') encoded_result_message = OcrdResultMessage.encode_yml(result_message) self.rmq_publisher.publish_to_queue( queue_name=result_queue, message=encoded_result_message )
[docs] def create_queue( self, connection_attempts: int = config.OCRD_NETWORK_WORKER_QUEUE_CONNECT_ATTEMPTS, retry_delay: int = 3) -> None: """Create the queue for this worker Originally only the processing-server created the queues for the workers according to the configuration file. This is intended to make external deployment of workers possible. """ if self.rmq_publisher is None: attempts_left = connection_attempts if connection_attempts > 0 else 1 while attempts_left > 0: try: self.connect_publisher() break except AMQPConnectionError as e: if attempts_left <= 1: raise e attempts_left -= 1 sleep(retry_delay) # the following function is idempotent self.rmq_publisher.create_queue(queue_name=self.processor_name)