flowx

中文文档

FlowX

A flexible and extensible workflow execution library for Go, supporting multiple execution backends and DAG-based workflow orchestration.

Features

• DAG-Based Workflow: Define complex workflows using Directed Acyclic Graph (DAG) structure with Mermaid syntax
• Cyclic Graph Support: Conditional back-edges for controlled loops with iteration counter
• Multi-Backend Execution: Support for Local, Docker, and Kubernetes executors
• Concurrent Execution: Independent tasks run in parallel for optimal performance
• Conditional Edges: Dynamic execution paths with template-based condition expressions
• Event-Driven Architecture: Monitor workflow lifecycle through event listeners
• Template Engine: Dynamic configuration rendering with Pongo2 templates
• Metadata Management: Process-safe metadata storage and retrieval
• Log Streaming: Real-time log output with customizable log pushing
• Output Extraction: Extract structured data from command output using codec-block or regex patterns
• Runtime Recovery: Resume workflow execution from saved state
• Pause & Resume: Pause running workflows and modify graph structure during pause
• Dynamic Graph Modification: Add/remove nodes and edges at runtime
• Data Passing: Share data between nodes using metadata

Installation

go get github.com/LerkoX/flowx

Quick Start

package main

import (
    "context"
    "fmt"

    "github.com/LerkoX/flowx"
)

func main() {
    ctx := context.Background()

    // Create runtime
    runtime := flowx.NewRuntime(ctx)

    // Workflow configuration
    config := `
Version: "1.0"
Name: example-workflow

Executors:
  local:
    type: local
    config:
      shell: bash

Graph: |
  stateDiagram-v2
    [*] --> Build
    Build --> Test
    Test --> [*]

Nodes:
  Build:
    executor: local
    steps:
      - name: build
        run: echo "Building..."
  Test:
    executor: local
    steps:
      - name: test
        run: echo "Testing..."
`

    // Execute workflow synchronously
    workflow, err := runtime.RunSync(ctx, "workflow-1", config, nil)
    if err != nil {
        fmt.Printf("Workflow failed: %v\n", err)
        return
    }

    fmt.Println("Workflow completed successfully!")
}

Output Extraction

FlowX supports extracting structured data from command output and saving it to workflow metadata for use in subsequent nodes. Extracted data includes field descriptions (from comments) and source node tracking.

Codec-Block Extraction

Automatically recognizes and parses flowx-yaml code blocks:

Nodes:
  Build:
    executor: local
    extract:
      type: codec-block
      maxOutputSize: 1048576  # Optional, default 1MB
    steps:
      - name: build
        run: |
          echo "Building..."
          echo '```flowx-yaml'
          echo 'buildId: "12345"  # 构建ID'
          echo 'version: "1.0.0"  # 版本号'
          echo '```'

This extracts buildId and version from output with descriptions, and makes them available as {{ .Metadata.Build.buildId }} and {{ .Metadata.Build.version }}.

Regex Extraction

Extract data using regular expressions:

Nodes:
  Test:
    executor: local
    extract:
      type: regex
      patterns:
        coverage: "coverage: (\\d+\\.\\d+)%"
        tests: "(\\d+) tests? passed"
      maxOutputSize: 524288
    steps:
      - name: test
        run: go test -cover

This extracts test coverage and count from command output.

Configuration

FlowX uses YAML configuration with the following structure:

Version: "1.0"              # Configuration version
Name: my-workflow           # Workflow name

Metadate:                   # Metadata configuration
  type: in-config           # Store type: in-config, redis, http
  data:
    key: value

Param:                      # Workflow parameters
  buildId: "123"
  branch: "main"

Executors:                  # Global executor definitions
  local:
    type: local
    config:
      shell: bash
      workdir: /tmp

  docker:
    type: docker
    config:
      registry: docker.io
      network: host
      volumes:
        - /var/run/docker.sock:/var/run/docker.sock

Graph: |                    # DAG definition (Mermaid stateDiagram-v2)
  stateDiagram-v2
    [*] --> Build
    Build --> Test
    Test --> Deploy
    Deploy --> [*]

Nodes:                      # Node definitions
  Build:
    executor: local
    steps:
      - name: build
        run: go build .

  Test:
    executor: docker
    image: golang:1.21
    steps:
      - name: test
        run: go test ./...

Executors

Local Executor

Executes commands on the local machine.

Executors:
  local:
    type: local
    config:
      shell: bash          # Shell to use (bash, sh, zsh)
      workdir: /tmp        # Working directory
      env:                 # Environment variables
        KEY: value
      timeout: "30s"       # Command timeout
      pty: true            # Enable PTY for interactive programs

Docker Executor

Executes commands inside Docker containers.

Executors:
  docker:
    type: docker
    config:
      registry: docker.io   # Image registry
      network: host         # Network mode
      workdir: /app         # Container working directory
      tty: true             # Enable TTY
      ttyWidth: 120         # TTY width
      ttyHeight: 40         # TTY height
      daemonTimeout: "15s"  # Daemon control-plane response timeout (default 15s);
                            # fail fast instead of hanging when the daemon is unresponsive
      volumes:              # Volume mounts
        - /host/path:/container/path
      env:                  # Environment variables
        GO_VERSION: "1.21"

Kubernetes Executor

Executes commands inside Kubernetes pods.

Executors:
  k8s:
    type: k8s
    config:
      namespace: default
      serviceAccount: workflow-sa
      podReadyTimeout: "60s"  # Pod ready timeout

Note: The ssh executor type is reserved but not yet implemented. Only local, docker, and k8s/kubernetes are currently supported.

Conditional Edges

Define conditional execution paths using template expressions:

Graph: |
  stateDiagram-v2
    [*] --> Build
    Build --> Deploy: {{ Param.branch == "main" }}
    Build --> Test: {{ Param.branch != "main" }}
    Test --> [*]
    Deploy --> [*]

Complex conditions are supported:

# Multiple conditions
QualityCheck --> DeployStaging: {{ QualityCheck.allTestsPassed == true and QualityCheck.codeCoverage >= 80 }}

# Nested conditions
Deploy --> Production: {{ Param.environment == "production" and ManualApproval.approved == true }}

Cyclic Graph (Loop Execution)

FlowX supports controlled loops through conditional back-edges. A conditional edge that creates a cycle is accepted as a back-edge, enabling iterative execution.

MaxLoopIterations: 5          # Safety limit (default: 100)

Graph: |
  stateDiagram-v2
    [*] --> A
    A --> B
    B --> C
    C --> A: {{ iteration < 3 }}    # Back-edge: loop 3 times
    C --> D
    D --> [*]

The iteration variable starts at 0 and increments each loop. In this example, nodes A→B→C→D execute 3 times, then the loop exits.

See Conditional Edges for details on back-edges and iteration.

Dynamic Graph Modification

FlowX supports modifying the workflow graph at runtime. You can add/remove nodes and edges while the workflow is paused.

Using ModifyGraph (Fine-grained Control)

import (
    "github.com/LerkoX/flowx/core"
    "github.com/LerkoX/flowx/dag"
)

err := runtime.ModifyGraph(ctx, "workflow-id", dag.GraphModifications{
    AddNodes: []core.NodeConfig{
        {
            Name: "NewNode",
            Executor: "local",
            Steps: []core.Step{
                {Name: "step1", Run: "echo 'New step'"},
            },
        },
    },
    AddEdges: []dag.EdgeModification{
        {Source: "ExistingNode", Target: "NewNode", Expression: ""},
    },
    RemoveNodes: []string{"UnusedNode"},
    RemoveEdges: []dag.EdgeRemoval{{Source: "A", Target: "B"}},
})

Using UpdateConfig (Config-based Diff)

Update the workflow by providing a new YAML configuration. FlowX automatically computes the difference and applies the changes:

newConfig := `
Version: "1.0"
Name: my-workflow

Graph: |
  stateDiagram-v2
    [*] --> Build
    Build --> Deploy
    Deploy --> [*]

Nodes:
  Build:
    executor: local
    steps:
      - name: build
        run: echo "Building..."
  Deploy:
    executor: local
    steps:
      - name: deploy
        run: echo "Deploying..."
`

err := runtime.UpdateConfig(ctx, "workflow-id", newConfig)

Rules:

• Nodes that have already executed cannot be removed or modified
• New nodes will execute on resume
• Edges can only be removed if neither endpoint has executed

See doc/runtime.md for detailed API documentation.

Data Passing

Share data between nodes using metadata:

Nodes:
  Generate:
    executor: local
    steps:
      - name: generate
        run: |
          echo '```flowx-yaml'
          echo 'value: 42  # 计算结果'
          echo 'message: "hello world"  # 消息内容'
          echo '```'
    extract:
      type: codec-block

  Process:
    executor: local
    steps:
      - name: process
        run: |
          echo "Processing value: {{ .Metadata.Generate.value }}"
          echo "Message: {{ .Metadata.Generate.message }}"

Runtime Recovery

Resume workflow execution from saved state:

Nodes:
  Build:
    executor: local
    runtime:                    # Node runtime status for recovery
      status: "SUCCESS"         # Already completed, will be skipped
      startTime: "2026-03-30T10:00:00Z"
      endTime: "2026-03-30T10:01:00Z"
      steps:
        - name: build
          status: "SUCCESS"
          output: "Build completed"
    steps:
      - name: build
        run: echo "Building..."

  Test:
    executor: local
    runtime:                    # Node runtime status for recovery
      status: "PENDING"         # Will execute
    steps:
      - name: test
        run: echo "Testing..."

Event Monitoring

Monitor workflow execution through event listeners. Implement the dag.Listener interface and use event constants from the core package:

import (
    "github.com/LerkoX/flowx/core"
    "github.com/LerkoX/flowx/dag"
)

type MyListener struct{}

func (l *MyListener) Events() []dag.Event {
    return []dag.Event{
        core.EventWorkflowInit,
        core.EventWorkflowStart,
        core.EventWorkflowFinish,
        core.EventWorkflowExecutorPrepare,
        core.EventWorkflowExecutorPrepareDone,
        core.EventWorkflowNodeStart,
        core.EventWorkflowNodeFinish,
        core.EventWorkflowNodeFailed,
        core.EventWorkflowCancelled,
        core.EventWorkflowStatusUpdate,
        core.EventWorkflowPaused,
        core.EventWorkflowResumed,
        core.EventWorkflowGraphModified,
    }
}

func (l *MyListener) Handle(p dag.Workflow, event dag.Event) {
    switch event {
    case core.EventWorkflowInit:
        fmt.Println("Workflow initialized")
    case core.EventWorkflowStart:
        fmt.Println("Workflow started")
    case core.EventWorkflowFinish:
        fmt.Println("Workflow finished")
    case core.EventWorkflowExecutorPrepare:
        fmt.Println("Executor preparing")
    case core.EventWorkflowExecutorPrepareDone:
        fmt.Println("Executor prepared")
    case core.EventWorkflowNodeStart:
        fmt.Println("Node started")
    case core.EventWorkflowNodeFinish:
        fmt.Println("Node completed")
    case core.EventWorkflowNodeFailed:
        fmt.Println("Node failed")
    case core.EventWorkflowCancelled:
        fmt.Println("Workflow cancelled")
    case core.EventWorkflowStatusUpdate:
        fmt.Println("Workflow status updated")
    case core.EventWorkflowPaused:
        fmt.Println("Workflow paused")
    case core.EventWorkflowResumed:
        fmt.Println("Workflow resumed")
    case core.EventWorkflowGraphModified:
        fmt.Println("Workflow graph modified")
    }
}

workflow, err := runtime.RunSync(ctx, "id", config, &MyListener{})

Available Events:

Architecture

graph TB
    subgraph "Runtime Layer"
        RT[Runtime]
        RTI[Runtime Impl]
    end

    subgraph "Workflow Core"
        PL[Workflow]
        PLI[Workflow Impl]
        NODE[Node]
        EDGE[Edge]
        EVAL[Eval Context]
    end

    subgraph "Configuration"
        CFG[Config]
        TPL[Template Engine]
    end

    subgraph "Metadata"
        MD[Metadata Store]
        EXT[Extractor]
    end

    subgraph "Executor Core"
        EP[Executor Provider]
        EI[Executor Interfaces]
        ADP[Adapter Pattern]
        BRG[Bridge Pattern]
    end

    subgraph "Executor Implementations"
        K8S[K8s Executor]
        DOCKER[Docker Executor]
        LOCAL[Local Executor]
    end

    subgraph "Utilities"
        LOG[Logger]
        EVT[Event System]
    end

    RT --> RTI
    RTI --> PL
    PL --> PLI
    PLI --> NODE
    PLI --> EDGE
    PLI --> EVAL
    PLI --> EP
    EP --> EI
    EI --> ADP
    EI --> BRG
    ADP --> K8S
    ADP --> DOCKER
    ADP --> LOCAL
    BRG --> K8S
    BRG --> DOCKER
    BRG --> LOCAL
    RTI --> CFG
    RTI --> MD
    PLI --> MD
    CFG --> TPL
    NODE --> EXT
    PLI --> EVT
    RTI --> LOG

Examples

See examples/workflows/README.md for detailed workflow examples:

• File Processing: Automated log archiving and cleanup
• Data ETL: Parallel data collection and transformation
• CI/CD Deployment: Complete deployment workflow with quality gates
• Weather Notification: Weather API integration with messaging

API Reference

Types Workflow, Listener, Event, GraphModifications, EdgeModification, and EdgeRemoval are defined in the dag package (github.com/LerkoX/flowx/dag). NodeConfig and Step are defined in the core package (github.com/LerkoX/flowx/core). Event constants (e.g., core.EventWorkflowStart) are also in the core package.

Runtime

type Runtime interface {
    Get(id string) (dag.Workflow, error)                          // Get workflow by ID
    Cancel(ctx context.Context, id string) error                  // Cancel running workflow
    RunAsync(ctx context.Context, id string, config string, listener dag.Listener) (dag.Workflow, error)  // Async execution
    RunSync(ctx context.Context, id string, config string, listener dag.Listener) (dag.Workflow, error)   // Sync execution
    Rm(id string)                                                 // Remove workflow record
    Done() chan struct{}                                          // Runtime completion signal
    Notify(data interface{}) error                                // Notify runtime
    Ctx() context.Context                                         // Get runtime context
    StopBackground()                                              // Stop background processing
    StartBackground()                                             // Start background processing
    SetPusher(pusher logger.Pusher)                               // Set log pusher
    SetTemplateEngine(engine template.TemplateEngine)             // Set template engine
    GetTemplateEngine() template.TemplateEngine                   // Get template engine
    ExportConfig(id string) (string, error)                       // Export workflow config
    Pause(ctx context.Context, id string) error                   // Pause running workflow
    Resume(ctx context.Context, id string) error                  // Resume paused/stopped workflow
    ModifyGraph(ctx context.Context, id string, modifications dag.GraphModifications) error // Modify graph atomically
    UpdateConfig(ctx context.Context, id string, newConfigYAML string) error // Update workflow via YAML diff
    ListWorkflows() []string                                      // List active workflow IDs
}

Workflow

type Workflow interface {
    Id() string                                               // Get workflow ID
    GetGraph() dag.Graph                                      // Get DAG graph
    SetGraph(graph dag.Graph)                                 // Set DAG graph
    Status() string                                           // Get workflow status
    SetMetadata(store metadata.MetadataStore)                 // Set metadata store
    Metadata() dag.Metadata                                   // Get workflow metadata
    Listening(listener dag.Listener)                          // Set event listener
    Done() <-chan struct{}                                    // Workflow completion signal
    Run(ctx context.Context) error                            // Run workflow
    Notify()                                                  // Step notifies workflow
    Cancel()                                                  // Cancel workflow
    SetExecutorProvider(provider dag.ExecutorProvider)        // Set executor provider
    SetTemplateEngine(engine template.TemplateEngine)          // Set template engine
    GetTemplateEngine() template.TemplateEngine               // Get template engine
    SetPusher(pusher logger.Pusher)                           // Set log pusher
    Pause() error                                             // Pause workflow (waits for current level)
    Resume(ctx context.Context) error                         // Resume paused workflow
    IsModifiable() bool                                       // Check if graph can be modified
    CurrentNode() dag.Node                                    // Return currently executing node (if any)
}

Testing

go test ./...

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

License

GNU General Public License v3.0 - see LICENSE for details.