Keyboard shortcuts

Press ← or → to navigate between chapters

Press S or / to search in the book

Press ? to show this help

Press Esc to hide this help

pipeline

This is a command line tool to setup a processing pipeline.

The pipeline has two components:

  • a client, watching a repository for files to process and sending them to a server;
  • a server, waiting for files from any number of clients and starting a processing pipeline.

Both the client and server processes are designed as long running processes with low CPU- and memory-footprints. They are intended to run as daemons to process a large number of files over hours, days, or longer durations.

Quickstart

You can create a configuration file and start the server with:

pipeline server config server.toml
# edit `server.toml` as required
pipeline server start server.toml

Similarly on the clients:

pipeline client config [--ssh-tunnel] client.toml
# edit `client.toml` as required
pipeline client start client.toml

The default configuration files generated as above contain comments explaining each configuration option. The --ssh-tunnel option produces a configuration file that uses SSH tunnelling to connect to the server.

You can set the PIPELINE_LOG environment variable to change the verbosity of logs. Accepted values in order of decreasing verbosity are:

  • debug: the most verbose level;
  • info: the default level;
  • warn: only show warnings;
  • off: disable logging.

Acknowledgment

This tool was initially developed for processing of images produced by an Apollo microscope.

We acknowledge the Scottish Centre for Macromolecular Imaging (SCMI) and Kako Stapleton for assistance with cryo-EM experiments and access to instrumentation, funded by the MRC (MC_PC_17135, MC_UU_00034/7, MR/X011879/1) and SFC (H17007).

Funding for this project was provided by the MRC (MC_UU_00034/7, MR/X011879/1).

Server configuration

You can generate a configuration file with

pipeline server config server.toml

Important

Relative paths in the configuration file are resolved with respect to the configuration file location.

Incoming directory

The incoming_directory option sets the path on the pipeline server where files to process will be expected to be copied by the client.

# in server configuration
incoming_directory = "./server/buckets"

The copy_to_server command on the client side must copy the files to process to that directory, with the name {server_filename}. For example with rsync:

# in client configuration
copy_to_server = [
    "rsync",
    "{client_path}",
    "server_host:path/to/server/buckets/{server_filename}",
]

Tip

In practice, {server_filename} will be replaced by a hash distributed in subfolders, e.g. 2d/af/2daf63e8. The length and algorithm for the hash as well as the number and names of subfolders are implementation details and should not be relied upon. If the server runs on unix, you can control the mode (as in permissions typically set via chmod) of the last subfolder (here af) via the unix_mode option, e.g.

# in server configuration
unix_mode = "0o770"

to set 770 as mode.

Server address

The server address is set in a dedicated section:

[server]
address = "127.0.0.1:47890"

Pick an appropriate subnet and port number so that the client(s) can access the server. See the corresponding client configuration.

Processing groups

A pipeline server can define an arbitrary set of “processing groups”. Each group defines what commands should be ran on each file to process.

For example:

[processing.main]
processing = [
    { create_directory = "./server/{client_relative_directory}" },
    [ "cp", "{server_path}", "./server/{client_relative_directory}/{client_file_stem}.out" ],
]
after_processing = { mark_as = "Done" }

defines the main processing group. You can define as many groups as you want. To define a group with name group_name, append a section [processing.group_name] to the configuration file.

Important

The processing option of each watching group in the client configuration must match a processing group in the server configuration. The connection to the server will be denied if the client configuration references non-existent processing groups.

Processing command

processing defines the command(s) to apply to incoming files. This can be either:

  • an external command given as a list of strings representing the program name and its arguments;
  • a { create_directory = "path" } directive;
  • a { delete_file = "path" } directive;
  • a { delete_directory = "path" } directive;
  • a list where each element is either of the previous;
  • "pass" to not do anything.

The following placeholders are replaced at runtime:

  • {server_path} is the path of the file on the server;
  • {client_name} is the name of the client as defined in the client configuration file;
  • {client_relative_directory} is the path to the file on the client, relative to the watched directory;
  • {client_file_stem} is the file name on the client without its extension;
  • {client_file_name} is the full file name on the client;
  • {hash} is a unique hash identifying the file. Using it as part of the output filename of your processing command guarantees its uniqueness, so that processing different files does not overwrite output.

After processing

What to do if the processing step was successful.

This can be either:

  • { mark_as = "Done" }: mark the job as completed;
  • { mark_as = "ToPrune" }: mark the job as completed and for pruning;
  • "pass": leave the job marked as being processed;
  • { move_to_and_prune = "destination/path" }: move the processed file to the given destination and prune it from database (the same substitutions as in processing are available). This uses an efficient rename if the destination is on the same filesystem as the incoming_directory, and is therefore more efficient than making a copy as part of processing. Note that this doesn’t automatically create directories, make sure the processing command creates directories if necessary.

The "pass" option is intended for cases when the processing command above doesn’t actually perform the desired processing but instead schedules it for execution (e.g. via a SLURM queue). In this scenario, the job has to be manually marked as done, failed or to-prune by calling

pipeline query mark query.toml {hash} done|failed|to-prune

Retrying failed tasks

You can control how often failed tasks are retried.

retry_tasks_every_secs = 60

will restart failed tasks every 60 seconds.

Pruning ToPrune tasks

You can control how often tasks marked as ToPrune should be pruned. A pruned task will be forgotten by the server and the incoming file deleted.

prune_every_secs = 120

will prune tasks every 120 seconds.

Concurrency

You can control how many instances of demanding tasks are done at once:

[concurrency]
max_hashes = 3
max_processing = 8
  • max_hashes controls the number of concurrent hash calculations;
  • max_processing controls the number of concurrent processing tasks.

Database

The server keeps track of incoming tasks and their status in an sqlite database.

[database]
wal = false

Setting wal to true enables the WAL journaling mode. See https://www.sqlite.org/wal.html for more information, in particular regarding filesystem-related restrictions. Note that the pipeline server holds an exclusive connection to the database, so it should in principle always be fine to enable WAL. If false, the “truncate” mode is used.

Client configuration

You can generate a configuration file with

pipeline client config client.toml

Important

Relative paths in the configuration file are resolved with respect to the configuration file location.

Client name

This is a convenience to identify more easily the provenance of files on the server side. This can also be accessed in processing commands on the server side via the {client_name} placeholder.

name = "client_name"

Sending files to the server

copy_to_server defines a command to copy a file to the server for processing. This is a list of strings representing the command and its arguments. The following placeholders are replaced at runtime:

  • {client_path} is the absolute path of the file on the client, see the [watching] section for how files are discovered;
  • {server_filename} is the name of the file as expected by the pipeline.

Important

This copy_to_server command must result in the file being copied to the incoming_directory (see the server configuration) with file name {server_filename}.

For example:

copy_to_server = [
    "cp",
    "{client_path}",
    "./server/buckets/{server_filename}",
]

Tip

If the server filesystem is mounted locally, you can ask pipeline to copy the file instead of relying on an external process. For instance, the copy in the example command can be more efficiently achieved with

copy_to_server = { destination = "./server/buckets" }

If the server and the client operate on the same filesystem, you can ask pipeline to merely rename the file for better performance:

copy_to_server = { move_in_same_fs_to = "./server/buckets" }

Connection to the server

This is done in a dedicated section.

Direct connection

If the socket is directly available, you can specify the address:

[server]
address = "127.0.0.1:47890"

SSH tunnelling

pipeline also support connecting to the server via an SSH tunnel.

[server]
ssh_host = "192.168.0.1"
ssh_port = 22
ssh_auth = { method = "none", user = "user" }
keepalive_every_secs = 60
server_addr_from_host = "127.0.0.1"
server_port_from_host = 47890
accepted_ssh_keys = []
  • ssh_host is the address of the ssh host for tunnelling.
  • ssh_port is the port used by the host for SSH connections.
  • ssh_auth specifies how SSH authentication should happen. Supported options are:
    • no authentication (e.g. if otherwise managed by network interface) 
      ssh_auth = { method = "none", user = "user" }
    • ask a password to user (requires keyboard interaction) 
      ssh_auth = { method = "password", user = "user" }
    • use a key via the openssh agent at SSH_AUTH_SOCK 
      ssh_auth = { method = "key", user = "user", public_key = "path/to/key.pub" }
      You can also specify the SSH socket to use with agent = 'path/to/socket'.
  • keepalive_every_secs sends a keepalive request if no communication occurs for this duration in seconds.
  • server_addr_from_host and server_port_from_host specify the network socket of the pipeline server as seen from the SSH host.
  • accepted_ssh_keys is a list of accepted keys from the SSH host (in OpenSSH format).

Watching for files to process

The pipeline client continuously watches a directory to find new files to process. This is done in the watching section. The high level options are the following:

[watching]
directory = "./client"
refresh_every_secs = 5
max_concurrent_hashes = 3
heartbeat_every_refreshes = 10
  • directory is the path of the directory being watched.
  • refresh_every_secs sets how often the client should look for new files in the watched directory, in seconds.
  • max_concurrent_hashes sets the maximum concurrent computations of file hashes.
  • heartbeat_every_refreshes sets the number of refreshes before logging out a heartbeat detailing how many files have been found since the last heartbeat. Set to 0 to disable heartbeat.

Watching groups

The client can sort files found in the watched directory into different groups (e.g. to request a different processing command to the server).

Defining a group is done as follow:

[[watching.groups]]
filters = { extension = "dat" }
processing = "main"
last_modif_secs = 10
full_hash = true

As many group as necessary can be defined by appending a [[watching.groups]] section to the configuration file.

For each file in the watched directory, groups are considered in the order they are defined in the configuration file. A file will end up in the first group whose filters it passes (even if last_modif_secs have not elapsed yet).

  • filters is the set of filters applying to this group. Available filters are:
    • extension, checking the file extension matches the given name
    • min_depth and max_depth, checking the depth of the file relative to the watched directory (note, the directory itself is depth 0, all files are therefore of depth 1 at least). Both values are inclusive.
  • processing defines which processing group should be run for these files (see server configuration).
  • last_modif_secs defines how long ago the last modification should be (in seconds) for the file to be considered ready for processing. This is useful if the files are created and enriched in a non-atomic manner by some other process.
  • full_hash specifies whether to use full hashes (of the entire file contents), or shallow hashes. It is recommended to use full hashes when possible for more robust data integrity check. Shallow hashes should be reserved for when the pipeline has to process large files for which computing the full hash is too demanding.
  • keep_on_client (false if unspecified) requests the pipeline not to delete the files that are sent for processing on the client side. This is meant for workflows where those files are still being used on the client side by some other process. This option should be used parsimoniously as files might accumulate in the watched directory if not otherwise deleted, leading to a more and more costly watching-loop with time.

Queries

The pipeline query subcommand performs queries to the pipeline server to know about its status or mark tasks with a given status. Its configuration file can be generated with

pipeline query config query.toml

Note however that all that’s needed is the server address, be it from the server side or the client side. In practice, this means you can use a server or client configuration file instead of a dedicated query.toml