engine

README

Arcaflow Engine

The Arcaflow Engine allows you to run workflows using container engines, such as Docker or Kubernetes. The plugins must be built with the Arcaflow SDK.

Pre-built binaries

If you want to use our pre-built binaries, you can find them in the releases section.

Building from source

This system requires at least Go 1.18 to run and can be built from source:

go build -o arcaflow cmd/arcaflow/main.go

This binary can then be used to run Arcaflow workflows.

Building a simple workflow

The simplest workflow is the example plugin workflow using the workflow schema version v0.2.0: (save it to workflow.yaml)

version: v0.2.0
input:
  root: RootObject
  objects:
    RootObject:
      id: RootObject
      properties:
        name:
          type:
            type_id: string
steps:
  example:
    plugin: ghcr.io/janosdebugs/arcaflow-example-plugin
    # step: step-id if the plugin has more than one step
    # deploy:
    #   type: docker|kubernetes
    #   ... more options
    input:
      name: !expr $.input.name
output:
  message: !expr $.steps.example.outputs.success.message

As you can see, it has a version, input, a list of steps, and an output definition. Each of these keys is required in a workflow. These can be linked together using JSONPath expressions (not all features are supported). The expressions also determine the execution order of plugins.

You can now create an input YAML for this workflow: (save it to input.yaml)

name: Arca Lot

If you have a local Docker / Moby setup installed, you can run it immediately:

./arcaflow -input input.yaml

If you don't have a local Docker setup, you can also create a config.yaml with the following structure:

deployers:
  image: 
    deployer_name: docker|podman|kubernetes
  python:
    deployer_name: python
  # More deployer options
log:
  level: debug|info|warning|error

You can load this config by passing the -config flag to Arcaflow.

Supported Workflow Schema Versions

• v0.2.0

Deployer options

Currently, the two deployer types supported are Docker and Kubernetes.

The Docker deployer

This deployer uses the Docker socket to launch containers. It has the following config structure:

image:
  deployer_name: docker
  connection:
    host: # Docker connection string
    cacert: # CA certificate for engine connection in PEM format
    cert: # Client cert in PEM format
    key: # Client key in PEM format
  deployment:
    container: # Container options, see https://docs.docker.com/engine/api/v1.41/#tag/Container/operation/ContainerCreate
    host: # Host options, see https://docs.docker.com/engine/api/v1.41/#tag/Container/operation/ContainerCreate
    network: # Network options, see https://docs.docker.com/engine/api/v1.41/#tag/Container/operation/ContainerCreate
    platform: # Platform options, see https://docs.docker.com/engine/api/v1.41/#tag/Container/operation/ContainerCreate

    # Pull policy, similar to Kubernetes
    imagePullPolicy: Always|IfNotPresent|Never
  timeouts:
    http: 15s

Note: not all container options are supported. STDIN/STDOUT-related options are disabled. Some other options may not be implemented yet, but you will always get an error message explaining missing options.

The Kubernetes deployer

The Kubernetes deployer deploys on a Kubernetes cluster. It has the following config structure:

image:
  deployer_name: kubernetes
  connection:
    host: api.server.host
    path: /api
    username: foo
    password: bar
    serverName: tls.server.name
    cert: PEM-encoded certificate
    key: PEM-encoded key
    cacert: PEM-encoded CA certificate
    bearerToken: Bearer token for access
    qps: queries per second
    burst: burst value
  deployment:
    metadata:
      # Add pod metadata here
    spec:
      # Add a normal pod spec here, plus the following option here:
      pluginContainer:
        # A single container configuration the plugin will run in. Do not specify the image, the engine will fill that.
  timeouts:
    http: 15s

Documentation

Overview

Package engine provides an embeddable engine variant.

Index

• Variables
• func NewDefaultStepRegistry(logger log.Logger, deployerRegistry deployerRegistry.Registry, ...) (step.Registry, error)
• func SupportedVersion(version string) (string, error)
• type Workflow
• type WorkflowEngine
  • func New(config *config.Config) (WorkflowEngine, error)

Variables

var DefaultDeployerRegistry = registry.New(
	deployer.Any(docker.NewFactory()),
	deployer.Any(kubernetes.NewFactory()),
	deployer.Any(podman.NewFactory()),
	deployer.Any(python.NewFactory()),
)

DefaultDeployerRegistry contains the deployers.

var ErrNoWorkflowFile = fmt.Errorf("no workflow file provided in context")

ErrNoWorkflowFile signals that no workflow file was provided in the context.

Functions

func NewDefaultStepRegistry

func NewDefaultStepRegistry(logger log.Logger, deployerRegistry deployerRegistry.Registry, config *config.Config) (step.Registry, error)

NewDefaultStepRegistry creates a registry with the default step types applied.

func SupportedVersion

func SupportedVersion(version string) (string, error)

SupportedVersion confirms whether a given version string is in the set of supported workflow specifications. It returns true when the version is in the set, false otherwise. Earlier schema validation already applies version's regular expression.

Types

type Workflow

type Workflow interface {
	// Run executes the workflow with the passed, YAML-formatted input data.
	Run(
		ctx context.Context,
		input []byte,
	) (
		outputID string,
		outputData any,
		outputIsError bool,
		err error,
	)

	// InputSchema returns the requested input schema for the workflow.
	InputSchema() schema.Scope
	// Outputs returns the list of possible outputs and their schema for the workflow.
	Outputs() map[string]schema.StepOutput
}

Workflow is a runnable, queryable workflow. You can execute it, or query it for schema information.

type WorkflowEngine

type WorkflowEngine interface {
	// RunWorkflow is a simplified shortcut to parse and immediately run a workflow.
	RunWorkflow(
		ctx context.Context,
		input []byte,
		workflowContext map[string][]byte,
		workflowFileName string,
	) (outputID string, outputData any, outputError bool, err error)

	// Parse ingests a workflow context as a map of files to their contents and a workflow file name and
	// parses the data into an executable workflow.
	Parse(
		workflowContext map[string][]byte,
		workflowFileName string,
	) (
		workflow Workflow,
		err error,
	)
}

WorkflowEngine is responsible for executing workflows and returning their result.

func New

func New(
	config *config.Config,
) (WorkflowEngine, error)

New creates a new workflow engine with the provided configuration. The passed deployerRegistry is responsible for providing deployment plugins.