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: (save it to workflow.yaml)

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 an input, a list of steps, and an output definition. 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:

deployer:
  type: docker|kubernetes
  # More deployer options
log:
  level: debug|info|warning|error

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

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:

type: 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:

type: 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)
• 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.

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.