Developing a Split Engine with tdsplitengine

There are actions that every Engine container must take at startup and shutdown to interface with the Processing Framework. One option for minimizing the effort for required to implement the required common actions is to use the tdsplitengine Python package provided by the processing framework team.

The tdsplitengine targets the common case where an Engine’s unique logic is already implemented as a command line executable. An Engine derived from tdsplitenge shells out to the command line executable to implement the unique Engine logic. Installation instructions for the tdsplitengine Python package are available here.

The integritychecker Engine was developed to serve as an example of a very small Engine that performs useful work. The integritychecker Engine implements the compute_checksum Operation used as the annotated example in task-json. The small Operation demonstrates the use of input and output ports as well as input and output parameters. The Engine shells out the Linux sha256sum command to compute file checksums. Python is perfectly capable of computing the checksum but the example Engine shells out to demonstrate the most common Engine implementation pattern.

The file checksum computed by the Engine could easily be implemented in Python but the Engine shells out to the Linux sha256sum command to demonstrate the common usage pattern where a small Python glue process shells out to a custom executable to implement the Engine logic.

The Engine’s Python code can be implemented in a single file because it relies on the tdsplitengine package to implement the boilerplate code required for Engine setup and teardown.

The full source of the integrity checker Engine is in the eTools Bitbucket repository.

The Engine’s __main__.py file contains the Engine’s source and is reproduced below for discussion.

"""
A minimalist Engine implemented as an example of creating a Processing
Framework Engine that executes a sub process to perform its actual work.
The Python code in this module builds a command line and then executes
another program to perform the useful work
"""
import logging
from pathlib import Path
import subprocess

import tdsplitengine

_logger = logging.getLogger(__name__)

operation_dispatcher = tdsplitengine.Dispatcher()

@operation_dispatcher.operation
def compute_checksum(task_input: dict) -> dict:
    """
    The compute_checksum Operation executes the sha256sum program to
    compute the sha256 or md5 hash of each file in its input port.
    """
    parameters = task_input["parameters"]
    algorithm = parameters["algorithm"]

    input_ports = task_input["inputs"]
    sources_port = input_ports["sources"]

    # Create a path to store the file of generated hashes in the output port
    hashes_out_port: Path = Path(task_input["outputs"]["hashes"])
    output_path: Path = hashes_out_port / "file_hashes.txt"

    # Create a directory for exporting log messages written by the sha256sum executable
    secondary_logs_dir: Path = Path(task_input["secondary_logs"])
    stderr_path = secondary_logs_dir / "sum.log"

    # Build a list containing the relative paths of all files in the input port
    relative_port_contents = []
    for source in sources_port:
        base_directory = Path(source["base-directory"])
        relative_paths = [
            Path(local_path).relative_to(base_directory)
            for local_path in source["local-paths"]
        ]
        relative_port_contents.extend(relative_paths)

    executable = "sha256sum" if algorithm == "sha256" else "md5sum"

    if len(relative_port_contents) > 0:
        sum_args = (executable, "--tag", *relative_port_contents)
        _logger.debug("Argument Tuple: %s", sum_args)

        with output_path.open("wt") as out_hashes:
            with stderr_path.open("wt") as out_error:
                completed = subprocess.run(
                    sum_args,
                    check=True,
                    stdout=out_hashes,
                    stderr=out_error,
                    cwd=base_directory,
                    timeout=600,
                )
    else:
        # It is an error to call one of the sum programs with an empty list but it is legal for
        # an input port to be empty so just create an empty output file.
        output_path.touch()

    # This is an example of an Operation that sets 2 output parameters
    # Hash the generated file of hashes and return as an output parameter.
    verify_args = (executable, "--tag", output_path)
    verify_completed = subprocess.run(
        verify_args, capture_output=True, check=True, timeout=30
    )
    output_parameters = {
        "hashhash": verify_completed.stdout.decode("UTF-8").strip(),
        "count": str(len(relative_port_contents)),
    }

    return output_parameters

def main():
    """
    Configure logging and then call the dispatcher to process the Operation execution request
    """

    # Configure the logging module
    logging.basicConfig(level=logging.DEBUG)

    # execute was not passed a task json, so it will look for JSON at the path specified by
    # the PF_TASK_JSON_PATH environment variable
    Operation_dispatcher.execute()


if __name__ == "__main__":
    # Launching this module using the Python interpreter's -m switch executes
    # this code
    main()

Explanation of main.py

operation_dispatcher = tdsplitengine.Dispatcher()

Instantiates the Dispatcher instance that is used to register the Operation methods and process the request and result JSON files.

@operation_dispatcher.operation
def compute_checksum(task_input: dict) -> dict:

The compute_checksum function is decorated with @operation_dispatcher.operation to indicate that this function implements the compute_checksum Operation. Optionally the decorator can use a operation_name parameter to specify an operation name is different from the function name @operation_dispatcher.operation(operation_name="compute_checksum"). This Operation name must be what your Processing Framework Operation is called.

An Operation function must have a signature that accepts a single dictionary parameter and returns Union[None, Dict, Tuple[Optional[Dict], Optional[Reschedule]]]. See Task Request JSON Contents for an explanation of the contents of the task_input dictionary.

   parameters = task_input["parameters"]
   algorithm = parameters["algorithm"]

The Operation’s input parameters are specified in the “parameters” element of task_input. This Operation expects a single input parameter named “algorithm”.

    input_ports = task_input["inputs"]
    sources_port = input_ports["sources"]

The Operation’s input ports are specified in the “inputs” element of task_input. This Operation expects a single input port named “sources”.

    hashes_out_port: Path = Path(task_input["outputs"]["hashes"])
    output_path: Path = hashes_out_port / "file_hashes.txt"

The Operation’s output ports are specified in the “outputs” element of task_input. This Operation expects a single output port named “hashes”. An output port entry specifies the directory that represents the output port. The Operation is responsible for supplying individual file names. In this case the Operation will write a file named “file_hashes.txt” in the output port.

    secondary_logs_dir: Path = Path(task_input["secondary_logs"])
    stderr_path = secondary_logs_dir / "sum.log"

The Operation’s secondary logs directory is specified in the “secondary_logs” element of task_input. Any log files written to this directory can be retrieved using the Processing Framework Request log contents API.

    relative_port_contents = []
    for source in sources_port:
        base_directory = Path(source["base-directory"])
        relative_paths = [
            Path(local_path).relative_to(base_directory)
            for local_path in source["local-paths"]
        ]
        relative_port_contents.extend(relative_paths)

This section builds a list of all files in the input port.

    executable = "sha256sum" if algorithm == "sha256" else "md5sum"

This line determine which Linux command to run based on the value of the algorithm input parameter. This is just an example Engine, so it assumes md5sum if the value isn’t sha256. A real Engine might want to verify that the parameter is in the expected range.

    sum_args = (executable, "--tag", *relative_port_contents)

This line builds the Linux command line to execute.

    with output_path.open("wt") as out_hashes:
        with stderr_path.open("wt") as out_error:
            completed = subprocess.run(
                sum_args,
                check=True,
                stdout=out_hashes,
                stderr=out_error,
                cwd=base_directory,
                timeout=600,
            )

This section shells out to the Linux sha256sum or md5sum command to perform the actual work of the Operation. The md5sum command outputs results to stdout so the stdout must be redirected to the previously generated out_hashes path. Anything written to stderr is redirected to the out_error file for export.

The process of running external command varies by executable. Many executables write their own output and log files, thus not needing to redirect stdout and stderr.

    output_parameters = {
        "hashhash": verify_completed.stdout.decode("UTF-8").strip(),
        "count": str(len(relative_port_contents)),
    }

    return output_parameters

This section demonstrates how an Operation sets output parameter values. To set output parameters return a dictionary whose keys are parameter names and values are the parameter values.

If the Operation does not set output parameters it can return None instead. Raise an exception in the Operation function indicates the Operation failed. The Operation failure and exception message will be recorded in the result JSON file.

    logging.basicConfig(level=logging.DEBUG)

Configures the Python logging module to write log messages to stdout. When running the Engine in a Docker container the Docker log driver will collect messages written to stdout so the Engine’s log messages should always be written to stdout. Much more complex logging module configurations than that produced by basicConfig are possible. For example, you might want to also write the log messages to a file in the tertiary logs directory for subsequent export. See the engine logging page for more details about logging.

operation_dispatcher.execute()

This line notifies the dispatcher that all Operations have been registered so it should proceed to process the Operation specified in the task request JSON.

Custom Environment Variables (or Secrets)

Custom environment variables (or secrets) are useful for adding information to a container on launch. This prevents sensitive information from being included within the docker image. Some use cases include:

API Key/Secret
Executable license key

Secrets can be mounted to an Engine using Deployment Secret resources. See the API documentation for a practical example of attaching a Secret to an Engine.

Environment variables can be accessed within a tdsplitengine Operation definition using python’s builtin os package:

import os
my_secret_key = os.environ["MY_SECRET_KEY"]

Where:

MY_SECRET_KEY is the name of the environment variable (or Engine Secret) you set. Please note that while the client provides a base64 encoded string to the Processing Framework API when creating an Engine Secret, the value gets attached to the Engine container as decoded content. The Engine/Operation logic does not need to decode this value.

Developing a Split Engine with tdsplitengine

Explanation of __main__.py

Custom Environment Variables (or Secrets)

Explanation of main.py