container

README

Container Package

Temporal workflows for executing Docker containers with advanced orchestration patterns. This package provides Argo Workflow-like capabilities using Temporal, including builder patterns, pre-built workflow templates, and full lifecycle management.

Key Features

• Single, Pipeline, Parallel, and DAG workflows for container execution
• Fluent Builder API with container, script, and HTTP templates
• Wait strategies — log, port, HTTP, and health-check based readiness detection
• Pre-built patterns — CI/CD, fan-out/fan-in, map-reduce, parallel testing
• Workflow lifecycle — submit, wait, watch, cancel, terminate, signal, query
• Resource management — CPU, memory, GPU limits, artifacts, and secrets

Documentation

• Container Workflows Guide — comprehensive usage guide with examples
• Architecture — how this package fits in the overall system
• Workflow Patterns — orchestration patterns (pipeline, parallel, DAG)
• Getting Started — quick start guide

Quick Example

input := container.ContainerExecutionInput{
    Image:      "postgres:16-alpine",
    Env:        map[string]string{"POSTGRES_PASSWORD": "test"},
    Ports:      []string{"5432:5432"},
    WaitStrategy: container.WaitStrategyConfig{
        Type:       "log",
        LogMessage: "ready to accept connections",
    },
    AutoRemove: true,
}

we, _ := c.ExecuteWorkflow(ctx,
    client.StartWorkflowOptions{ID: "pg", TaskQueue: "container-tasks"},
    container.ExecuteContainerWorkflow, input)

See examples/container/ for complete working examples.

Documentation

Overview

Package container provides Temporal workflow activities for executing container workloads via Docker or Podman.

It implements the workflow.TaskInput and workflow.TaskOutput interfaces with container-specific payload types (image, command, environment, volumes, etc.) and registers them as Temporal activities so they can be composed using the generic orchestration patterns in the workflow package (pipeline, parallel, loop, DAG).

Typical usage involves calling RegisterAll to register all container workflows and activities with a Temporal worker.

Index

• Variables
• func CancelWorkflow(ctx context.Context, c client.Client, workflowID, runID string) error
• func QueryWorkflow(ctx context.Context, c client.Client, workflowID, runID, queryType string, ...) error
• func RegisterActivities(w worker.Worker)
• func RegisterAll(w worker.Worker)
• func RegisterWorkflows(w worker.Worker)
• func SignalWorkflow(ctx context.Context, c client.Client, workflowID, runID, signalName string, ...) error
• func TerminateWorkflow(ctx context.Context, c client.Client, workflowID, runID, reason string) error
• func WatchWorkflow(ctx context.Context, c client.Client, workflowID, runID string, ...) error
• type Artifact
• type ConditionalBehavior
• type ContainerExecutionInput
• type ContainerExecutionOutput
• type DAGNode
• type DAGWorkflowInput
• type DAGWorkflowOutput
• type ExtendedContainerInput
• type InputMapping
• type NodeResult
• type OutputDefinition
• type ResourceLimits
• type SecretReference
• type WaitStrategyConfig
• type WorkflowExecutionInfo
  • func GetWorkflowHistory(ctx context.Context, c client.Client, workflowID, runID string) (*WorkflowExecutionInfo, error)
• type WorkflowParameter
• type WorkflowStatus
  • func GetWorkflowStatus(ctx context.Context, c client.Client, workflowID, runID string) (*WorkflowStatus, error)
  • func ListWorkflows(ctx context.Context, c client.Client, query string) ([]*WorkflowStatus, error)
  • func SubmitAndWait(ctx context.Context, c client.Client, input interface{}, taskQueue string, ...) (*WorkflowStatus, error)
  • func SubmitAndWaitTyped[O any](ctx context.Context, c client.Client, workflowFunc interface{}, ...) (*WorkflowStatus, *O, error)
  • func SubmitTypedWorkflow[I any](ctx context.Context, c client.Client, workflowFunc interface{}, input I, ...) (*WorkflowStatus, error)
  • func SubmitWorkflow(ctx context.Context, c client.Client, input interface{}, taskQueue string) (*WorkflowStatus, error)

Variables

var ValidateVolumes = payload.ValidateVolumes

ValidateVolumes re-exports the volume validation function.

Functions

func CancelWorkflow

func CancelWorkflow(ctx context.Context, c client.Client, workflowID, runID string) error

CancelWorkflow cancels a running workflow.

Example:

err := docker.CancelWorkflow(ctx, temporalClient, workflowID, runID)

func QueryWorkflow

func QueryWorkflow(ctx context.Context, c client.Client, workflowID, runID, queryType string, result interface{}) error

QueryWorkflow queries a running workflow.

Example:

var result string
err := docker.QueryWorkflow(ctx, temporalClient, workflowID, runID, "status", &result)

func RegisterActivities

func RegisterActivities(w worker.Worker)

RegisterActivities registers all container activities with a worker. The activity is wrapped with OTel instrumentation that activates when otel.Config is present in the activity context. Calling this function multiple times on the same worker is a no-op for subsequent calls — each activity type is registered at most once per worker.

func RegisterAll

func RegisterAll(w worker.Worker)

RegisterAll registers both workflows and activities.

func RegisterWorkflows

func RegisterWorkflows(w worker.Worker)

RegisterWorkflows registers all container workflows with a worker. Calling this function multiple times on the same worker is a no-op for subsequent calls — each workflow type is registered at most once per worker.

func SignalWorkflow

func SignalWorkflow(ctx context.Context, c client.Client, workflowID, runID, signalName string, arg interface{}) error

SignalWorkflow sends a signal to a running workflow.

Example:

err := docker.SignalWorkflow(ctx, temporalClient, workflowID, runID, "pause", nil)

func TerminateWorkflow

func TerminateWorkflow(ctx context.Context, c client.Client, workflowID, runID, reason string) error

TerminateWorkflow terminates a running workflow.

Example:

err := docker.TerminateWorkflow(ctx, temporalClient, workflowID, runID, "reason")

func WatchWorkflow

func WatchWorkflow(ctx context.Context, c client.Client, workflowID, runID string, updates chan<- *WorkflowStatus) error

WatchWorkflow watches a workflow execution and streams updates.

Example:

updates := make(chan *WorkflowStatus)
err := docker.WatchWorkflow(ctx, temporalClient, workflowID, runID, updates)

Types

type Artifact

type Artifact = payload.Artifact

Type aliases re-exported from container/payload for convenience.

type ConditionalBehavior

type ConditionalBehavior = payload.ConditionalBehavior

Type aliases re-exported from container/payload for convenience.

type ContainerExecutionInput

type ContainerExecutionInput = payload.ContainerExecutionInput

Type aliases re-exported from container/payload for convenience. This allows consumers to use container.ContainerExecutionInput instead of container/payload.ContainerExecutionInput.

type ContainerExecutionOutput

type ContainerExecutionOutput = payload.ContainerExecutionOutput

Type aliases re-exported from container/payload for convenience. This allows consumers to use container.ContainerExecutionInput instead of container/payload.ContainerExecutionInput.

type DAGNode

type DAGNode = payload.DAGNode

Type aliases re-exported from container/payload for convenience.

type DAGWorkflowInput

type DAGWorkflowInput = payload.DAGWorkflowInput

Type aliases re-exported from container/payload for convenience.

type DAGWorkflowOutput

type DAGWorkflowOutput = payload.DAGWorkflowOutput

Type aliases re-exported from container/payload for convenience.

type ExtendedContainerInput

type ExtendedContainerInput = payload.ExtendedContainerInput

Type aliases re-exported from container/payload for convenience.

type InputMapping

type InputMapping = payload.InputMapping

Type aliases re-exported from container/payload for convenience.

type NodeResult

type NodeResult = payload.NodeResult

Type aliases re-exported from container/payload for convenience.

type OutputDefinition

type OutputDefinition = payload.OutputDefinition

Type aliases re-exported from container/payload for convenience.

type ResourceLimits

type ResourceLimits = payload.ResourceLimits

Type aliases re-exported from container/payload for convenience.

type SecretReference

type SecretReference = payload.SecretReference

Type aliases re-exported from container/payload for convenience.

type WaitStrategyConfig

type WaitStrategyConfig = payload.WaitStrategyConfig

Type aliases re-exported from container/payload for convenience. This allows consumers to use container.ContainerExecutionInput instead of container/payload.ContainerExecutionInput.

type WorkflowExecutionInfo

type WorkflowExecutionInfo struct {
	WorkflowID    string
	RunID         string
	WorkflowType  string
	StartTime     time.Time
	CloseTime     *time.Time
	Status        string
	HistoryLength int64
}

WorkflowExecutionInfo provides detailed information about a workflow execution.

func GetWorkflowHistory

func GetWorkflowHistory(ctx context.Context, c client.Client, workflowID, runID string) (*WorkflowExecutionInfo, error)

GetWorkflowHistory retrieves the history of a workflow execution.

Example:

history, err := docker.GetWorkflowHistory(ctx, temporalClient, workflowID, runID)

type WorkflowParameter

type WorkflowParameter = payload.WorkflowParameter

Type aliases re-exported from container/payload for convenience.

type WorkflowStatus

type WorkflowStatus struct {
	WorkflowID string
	RunID      string
	Status     string
	StartTime  time.Time
	CloseTime  *time.Time
	Result     interface{}
	Error      error
}

WorkflowStatus represents the status of a workflow execution.

func GetWorkflowStatus

func GetWorkflowStatus(ctx context.Context, c client.Client, workflowID, runID string) (*WorkflowStatus, error)

GetWorkflowStatus retrieves the current status of a workflow.

Example:

status, err := docker.GetWorkflowStatus(ctx, temporalClient, workflowID, runID)

func ListWorkflows

func ListWorkflows(ctx context.Context, c client.Client, query string) ([]*WorkflowStatus, error)

ListWorkflows lists workflows with optional filtering.

Example:

workflows, err := docker.ListWorkflows(ctx, temporalClient, "container-queue")

func SubmitAndWait

func SubmitAndWait(ctx context.Context, c client.Client, input interface{}, taskQueue string, timeout time.Duration) (*WorkflowStatus, error)

SubmitAndWait submits a workflow and waits for completion.

Example:

status, err := docker.SubmitAndWait(ctx, temporalClient, input, "container-queue", 10*time.Minute)

func SubmitAndWaitTyped

func SubmitAndWaitTyped[O any](
	ctx context.Context, c client.Client, workflowFunc interface{},
	input interface{}, taskQueue string, timeout time.Duration,
) (*WorkflowStatus, *O, error)

SubmitAndWaitTyped submits a workflow and waits for a typed result.

Example:

status, result, err := container.SubmitAndWaitTyped[payload.ContainerExecutionOutput](
    ctx, temporalClient, workflow.ExecuteContainerWorkflow, input, "container-queue", 10*time.Minute)

func SubmitTypedWorkflow

func SubmitTypedWorkflow[I any](ctx context.Context, c client.Client, workflowFunc interface{}, input I, taskQueue string) (*WorkflowStatus, error)

SubmitTypedWorkflow submits a typed workflow for execution. Unlike SubmitWorkflow which uses interface{} and a type switch, this function accepts a concrete workflow function and input, providing compile-time type safety.

Example:

status, err := container.SubmitTypedWorkflow(ctx, temporalClient, workflow.ExecuteContainerWorkflow, input, "container-queue")

func SubmitWorkflow

func SubmitWorkflow(ctx context.Context, c client.Client, input interface{}, taskQueue string) (*WorkflowStatus, error)

SubmitWorkflow submits a container workflow for execution.

Example:

status, err := docker.SubmitWorkflow(ctx, temporalClient, input, "container-queue")