Dagu Workflows

Dagu Workflow Orchestration Skill

Expert-level skill for building, deploying, and managing Dagu workflows for task orchestration, automation, and pipeline management.

Overview

Dagu is a compact, portable workflow engine that provides declarative orchestration of command execution across diverse environments. This skill covers complete workflow design patterns from simple cron replacements to complex multi-step pipelines with error handling, notifications, and distributed execution.

What Dagu Can Do

• Task Orchestration: Chain shell commands, scripts, Docker containers, or remote SSH commands
• Scheduling: Cron-based scheduling, one-time execution, or event-driven triggers
• Error Handling: Automatic retry with exponential backoff, failure notifications, error isolation
• Parallel Execution: Run steps concurrently with dependency management
• Monitoring: Web UI, logs, history tracking, and status monitoring
• Integration: HTTP APIs, databases (PostgreSQL, SQLite), message queues (Redis), cloud storage (S3)

What Dagu Cannot Do

• Long-running durable workflows (use Temporal instead for multi-day processes)
• Native event-driven triggers (requires external polling or webhooks)
• Stateful checkpointing across server restarts
• Complex decision trees with persistent state

When to Use Dagu vs Alternatives

Core Concepts

DAG (Directed Acyclic Graph)

A workflow defined as steps with dependencies forming a graph. Steps run in order respecting dependencies - a step only runs after all its dependencies complete successfully.

Execution Modes

Chain Mode (Default): Steps run sequentially top-to-bottom

steps:
  - command: step1.sh  # runs first
  - command: step2.sh  # waits for step1
  - command: step3.sh  # waits for step2

Graph Mode: Explicit dependencies, allows parallel execution

type: graph
steps:
  - id: step_a
    command: echo A
    depends: []         # runs immediately
  - id: step_b
    command: echo B
    depends: [step_a]   # waits for step_a
  - id: step_c
    command: echo C
    depends: [step_a]   # runs in parallel with step_b

Key Dagu 2.x Requirements

• Step IDs must use snake_case (not hyphens)
• Graph mode required for explicit dependencies
• All workflow files end in .yaml

Workflow Structure

Minimal Workflow

steps:
  - command: echo "Hello from Dagu!"

Complete Production Workflow

# Metadata
name: data-pipeline
description: Daily ETL pipeline for sales data
owner: data-team@company.com
tags:
  - etl
  - production
  - sales

# Execution mode
type: graph

# Scheduling (cron format)
schedule:
  - "0 2 * * *"  # Daily at 2 AM

# Global settings
timeout_sec: 3600  # 1 hour max
max_active_steps: 3  # Parallelism limit

# Environment variables
env:
  - ENV: production
  - DB_HOST: db.company.com

# Secrets (referenced from Dagu secret store)
secrets:
  - DB_PASSWORD
  - API_KEY

# Parameters (user input or defaults)
params:
  - DATE: "2026-01-01"
  - DRY_RUN: "false"

# DAG-level retry (entire workflow)
retry_policy:
  limit: 3
  interval_sec: 300  # 5 minutes
  backoff: true

# Step definitions
steps:
  # Extract
  - id: extract_sales
    name: Extract Sales Data
    command: |
      psql -h ${DB_HOST} -U etl_user -c "
        COPY (SELECT * FROM sales WHERE date = '${DATE}')
        TO '/tmp/sales_${DATE}.csv'
        WITH CSV HEADER;
      "
    retry_policy:
      limit: 3
      interval_sec: 10
    timeout_sec: 300

  # Transform (parallel transforms possible)
  - id: transform_revenue
    name: Calculate Revenue
    depends: [extract_sales]
    command: python /scripts/calc_revenue.py --input /tmp/sales_${DATE}.csv
    output: REVENUE_DATA

  - id: transform_metrics
    name: Calculate Metrics
    depends: [extract_sales]
    command: python /scripts/calc_metrics.py --input /tmp/sales_${DATE}.csv
    output: METRICS_DATA

  # Load
  - id: load_warehouse
    name: Load to Data Warehouse
    depends: [transform_revenue, transform_metrics]
    command: |
      python /scripts/load_dw.py \
        --revenue ${REVENUE_DATA} \
        --metrics ${METRICS_DATA} \
        --date ${DATE}

  # Cleanup
  - id: cleanup
    name: Cleanup Temporary Files
    depends: [load_warehouse]
    command: rm -f /tmp/sales_${DATE}.csv
    continue_on:
      exit_code: [0, 1]  # Continue even if cleanup fails

# Lifecycle handlers
handler_on:
  success:
    command: notify.sh "ETL succeeded for ${DATE}"
  failure:
    command: pager.sh "ETL failed for ${DATE}"
  exit:
    command: rm -rf /tmp/${DAG_RUN_ID}

# Notifications
mail_on:
  failure: true
  success: false

Step Types Reference

Shell Steps (Most Common)

steps:
  # Simple command
  - command: echo "Hello"

  # Multi-line script
  - command: |
      echo "Step 1"
      echo "Step 2"
      ./process.sh

  # With specific shell
  - command: python3 process.py
    executor: python3

  # With working directory
  - command: npm install
    working_dir: /app/frontend

  # Environment variables
  - command: python process.py
    env:
      - DEBUG: "1"
      - BATCH_SIZE: "1000"

  # Timeout
  - command: long_process.sh
    timeout_sec: 3600

  # Signal handling for graceful shutdown
  - command: server.sh
    signal_on_stop: SIGTERM  # Default, can use SIGINT, SIGKILL, etc.

HTTP Steps

steps:
  - type: http
    name: Fetch API Data
    config:
      url: https://api.example.com/data
      method: GET
      headers:
        Authorization: Bearer ${API_TOKEN}
        Content-Type: application/json
      timeout: 30
      silent: false  # Show response in logs
    output: API_RESPONSE

Docker Steps

steps:
  - type: docker
    name: Run Container
    config:
      image: python:3.11
      auto_remove: true
      volumes:
        - /host/data:/data
      environment:
        - ENV=production
      command: python /data/process.py

SSH Steps (Remote Execution)

steps:
  - type: ssh
    name: Deploy to Server
    config:
      host: prod-server-01
      user: deploy
      key: /home/user/.ssh/deploy_key
      command: |
        cd /app
        git pull
        docker-compose up -d

JQ Steps (JSON Processing)

steps:
  - type: jq
    name: Extract Price
    config:
      input: ${API_RESPONSE}  # From previous step
      query: .bitcoin.usd
    output: BTC_PRICE

Mail Steps

steps:
  - type: mail
    name: Send Report
    config:
      to: team@company.com
      from: dagu@company.com
      subject: "Daily Report - ${DATE}"
      message: |
        Revenue: ${REVENUE}
        Metrics: ${METRICS}
      attachments:
        - /tmp/report.pdf

SQL Steps

steps:
  # PostgreSQL
  - type: sql
    name: Query Database
    config:
      driver: postgres
      dsn: postgresql://user:pass@host/db
      query: SELECT * FROM sales WHERE date = $1
      args:
        - ${DATE}
    output: SALES_DATA

  # SQLite
  - type: sql
    name: Local Query
    config:
      driver: sqlite3
      dsn: /path/to/local.db
      query: SELECT count(*) FROM events

Template Steps

steps:
  - type: template
    name: Generate Config
    config:
      data:
        db_host: ${DB_HOST}
        db_name: ${DB_NAME}
    script: |
      #!/bin/bash
      cat > config.json << EOF
      {
        "database": {
          "host": "{{ .db_host }}",
          "name": "{{ .db_name }}"
        }
      }
      EOF

S3 Steps

steps:
  - type: s3
    name: Upload to S3
    config:
      endpoint: s3.amazonaws.com
      bucket: my-backup-bucket
      key: backups/${DATE}/data.tar.gz
      source: /tmp/data.tar.gz
      access_key: ${AWS_ACCESS_KEY}
      secret_key: ${AWS_SECRET_KEY}

Scheduling Patterns

Cron Scheduling

# Every minute
schedule:
  - "* * * * *"

# Every 5 minutes
schedule:
  - "*/5 * * * *"

# Daily at midnight
schedule:
  - "0 0 * * *"

# Daily at 2 AM
schedule:
  - "0 2 * * *"

# Every Monday at 9 AM
schedule:
  - "0 9 * * 1"

# First day of month at midnight
schedule:
  - "0 0 1 * *"

Multiple Schedules

schedule:
  - "0 9 * * 1"    # Monday 9 AM
  - "0 14 * * 5"   # Friday 2 PM

No Schedule (Manual/Event Triggered)

# No schedule field - only runs manually or via API/webhook
name: on-demand-job
steps:
  - command: ./process.sh

Error Handling & Retry

Step-Level Retry

steps:
  - id: fetch_api
    command: curl -fsS https://api.example.com/data
    retry_policy:
      limit: 5              # Max 5 retries
      interval_sec: 10      # Start with 10s delay
      exit_code: [6, 22, 28]  # Network errors only
      backoff: true         # Exponential: 10s, 20s, 40s, 60s
      max_interval_sec: 60  # Cap at 60s

Default Retry (All Steps)

defaults:
  retry_policy:
    limit: 3
    interval_sec: 5
    exit_code: [1, 2]

steps:
  - command: step1.sh  # Uses default retry
  - command: step2.sh  # Uses default retry
  - command: step3.sh
    retry_policy:
      limit: 0  # Override - no retry for this step

DAG-Level Retry (Entire Workflow)

retry_policy:
  limit: 3              # Retry whole DAG 3 times
  interval_sec: 300     # Wait 5 minutes between attempts
  backoff: true
  max_interval_sec: 3600  # Max 1 hour wait

Continue On Failure

steps:
  # Continue even if cleanup fails
  - command: rm -rf /tmp/cache
    continue_on:
      exit_code: [0, 1]  # 0=success, 1=minor error
      
  # Continue on specific output
  - command: ./validate.sh
    continue_on:
      output:
        - "WARNING"
        - "SKIP"
        - "re:^INFO:.*"  # Regex pattern

Pre-Conditions

steps:
  - command: ./deploy.sh
    preconditions:
      - condition: "${ENV}"
        expected: "production"

Lifecycle Handlers

Run code when workflow reaches certain states:

handler_on:
  # Run before any steps
  init:
    command: setup.sh ${DAG_NAME}
  
  # Run on successful completion
  success:
    command: notify.sh "${DAG_NAME} succeeded"
  
  # Run on failure
  failure:
    command: pager.sh "${DAG_NAME} failed"
  
  # Run on manual stop/timeout
  abort:
    command: rollback.sh
  
  # Run when waiting for approval
  wait:
    command: notify-slack.sh "Approval needed"
  
  # Always runs last (success or failure)
  exit:
    command: cleanup.sh

Control Flow Patterns

Sequential Execution (Default)

steps:
  - command: step1.sh
  - command: step2.sh  # waits for step1
  - command: step3.sh  # waits for step2

Parallel Execution

type: graph
steps:
  - id: task_a
    command: echo A
    depends: []
  - id: task_b
    command: echo B
    depends: []
  - id: task_c
    command: echo C
    depends: []
  # All three run in parallel

Fan-Out / Fan-In

type: graph
steps:
  # Fan out: One to many
  - id: prepare
    command: prepare.sh
    depends: []
  
  - id: process_1
    command: process_1.sh
    depends: [prepare]
  - id: process_2
    command: process_2.sh
    depends: [prepare]
  - id: process_3
    command: process_3.sh
    depends: [prepare]
  # process_1, 2, 3 run in parallel after prepare
  
  # Fan in: Many to one
  - id: combine
    command: combine.sh
    depends: [process_1, process_2, process_3]
  # Waits for all three processes

Diamond Pattern

type: graph
steps:
  - id: start
    command: start.sh
    depends: []
  
  - id: left
    command: left.sh
    depends: [start]
  - id: right
    command: right.sh
    depends: [start]
  
  - id: end
    command: end.sh
    depends: [left, right]

Conditional Execution

steps:
  - id: check_env
    command: |
      if [ "${ENV}" = "production" ]; then
        exit 0
      else
        exit 1
      fi
    continue_on:
      exit_code: [0, 1]
  
  - id: prod_deploy
    depends: [check_env]
    command: deploy_prod.sh
    preconditions:
      - condition: "${ENV}"
        expected: "production"

Loop Over Items

steps:
  - call: process_file
    parallel:
      items:
        - file1.csv
        - file2.csv
        - file3.csv
      max_concurrent: 2  # Process 2 at a time
    params: "FILE=${ITEM}"

---
name: process_file
params:
  - FILE: ""
steps:
  - command: python process.py --file ${FILE}

Data Flow & Variables

Environment Variables

# Global env (all steps)
env:
  - GLOBAL_VAR: "value"

steps:
  # Step-specific env
  - command: echo ${LOCAL_VAR}
    env:
      - LOCAL_VAR: "step_value"
      - GLOBAL_VAR: "overridden"  # Overrides global

Step Output to Variable

steps:
  - id: get_version
    command: cat VERSION
    output: VERSION_NUM  # Captures stdout
  
  - id: build
    depends: [get_version]
    command: |
      echo "Building version ${VERSION_NUM}"
      ./build.sh --version ${VERSION_NUM}

Dynamic Variables

env:
  - TODAY: "`date +%Y-%m-%d`"  # Evaluated at runtime
  - TIMESTAMP: "`date +%s`"

Template Variables

steps:
  - command: echo "Run ID: ${DAG_RUN_ID}"
  - command: echo "Started: ${DAG_RUN_STARTED_AT}"
  - command: echo "Status: ${DAG_RUN_STATUS}"

Deployment Patterns

Local Development

# Install
curl -fsSL https://dagu.dev/install.sh | sh

# Start server + scheduler
dagu start-all

# Access Web UI
open http://localhost:8080

macOS Service (Persistent)

# Install
dagu server

# Create LaunchAgent plist
cat > ~/Library/LaunchAgents/com.dagu.plist << 'EOF'
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN">
<plist version="1.0">
<dict>
  <key>Label</key>
  <string>com.dagu</string>
  <key>ProgramArguments</key>
  <array>
    <string>/opt/homebrew/bin/dagu</string>
    <string>start-all</string>
  </array>
  <key>RunAtLoad</key>
  <true/>
  <key>KeepAlive</key>
  <true/>
</dict>
</plist>
EOF

# Load service
launchctl load ~/Library/LaunchAgents/com.dagu.plist
launchctl start com.dagu

Linux Systemd

# Create service file
sudo tee /etc/systemd/system/dagu.service << 'EOF'
[Unit]
Description=Dagu Workflow Scheduler
After=network.target

[Service]
Type=simple
ExecStart=/usr/local/bin/dagu start-all
Restart=always
RestartSec=5
User=dagu

[Install]
WantedBy=multi-user.target
EOF

# Enable and start
sudo systemctl enable dagu
sudo systemctl start dagu

Docker Compose

# docker-compose.yml
version: '3'
services:
  dagu:
    image: ghcr.io/dagu-org/dagu:latest
    ports:
      - "8080:8080"
    volumes:
      - ./dags:/home/dagu/.config/dagu/dags
      - dagu-data:/home/dagu/.local/share/dagu
    environment:
      - DAGU_HOST=0.0.0.0
      - DAGU_PORT=8080
    command: dagu start-all

volumes:
  dagu-data:

Common Workflow Patterns

ETL Pipeline

name: daily-etl
schedule:
  - "0 2 * * *"
type: graph

steps:
  - id: extract
    command: python extract.py --date ${DATE}
    output: RAW_DATA
  
  - id: transform_clean
    depends: [extract]
    command: python transform.py --input ${RAW_DATA}
    output: CLEAN_DATA
  
  - id: load_dw
    depends: [transform_clean]
    command: python load.py --input ${CLEAN_DATA}
  
  - id: cleanup
    depends: [load_dw]
    command: rm -f ${RAW_DATA} ${CLEAN_DATA}
    continue_on:
      exit_code: [0, 1]

handler_on:
  success:
    command: curl -X POST slack-webhook -d '{"text":"ETL Complete"}'
  failure:
    command: curl -X POST pager-duty -d '{"incident":"ETL Failed"}'

CI/CD Pipeline

name: app-deploy
env:
  - APP_NAME: myapp

steps:
  - id: test
    command: npm test
  
  - id: build
    depends: [test]
    command: docker build -t ${APP_NAME}:${VERSION} .
  
  - id: push
    depends: [build]
    command: docker push ${APP_NAME}:${VERSION}
  
  - id: deploy_staging
    depends: [push]
    command: kubectl apply -f k8s/staging/
    approval:
      prompt: "Approve production deployment?"
  
  - id: deploy_prod
    depends: [deploy_staging]
    command: kubectl apply -f k8s/production/

Data Backup

name: backup-database
schedule:
  - "0 3 * * *"  # 3 AM daily

steps:
  - id: backup
    command: |
      pg_dump -h ${DB_HOST} -U ${DB_USER} ${DB_NAME} \
        | gzip > /backup/db_${DATE}.sql.gz
  
  - id: verify
    depends: [backup]
    command: gzip -t /backup/db_${DATE}.sql.gz
  
  - id: upload
    depends: [verify]
    type: s3
    config:
      bucket: my-backups
      key: database/db_${DATE}.sql.gz
      source: /backup/db_${DATE}.sql.gz
  
  - id: cleanup_local
    depends: [upload]
    command: find /backup -name "db_*.sql.gz" -mtime +7 -delete

API Monitoring

name: api-health-check
schedule:
  - "*/5 * * * *"  # Every 5 minutes

steps:
  - id: check_api
    command: |
      curl -fsS https://api.example.com/health \
        -H "Authorization: Bearer ${API_TOKEN}"
    retry_policy:
      limit: 3
      interval_sec: 5
      exit_code: [6, 22, 28]
  
  - id: check_db
    depends: [check_api]
    command: |
      psql ${DATABASE_URL} -c "SELECT 1;" > /dev/null
  
  handler_on:
    failure:
      command: |
        curl -X POST ${PAGERDUTY_WEBHOOK} \
          -d '{"incident": "API Down", "severity": "critical"}'

Advanced Features

Approval Gates

steps:
  - id: prepare_deploy
    command: prepare.sh
  
  - id: deploy_prod
    depends: [prepare_deploy]
    command: deploy.sh production
    approval:
      prompt: "Review staging deployment before approving production"
      input: [APPROVED_BY, MAINTENANCE_WINDOW]
      required: [APPROVED_BY]
  
  - id: notify
    depends: [deploy_prod]
    command: |
      echo "Deployed by ${APPROVED_BY} during ${MAINTENANCE_WINDOW}"

Queue Management

# ~/.config/dagu/config.yaml
queues:
  enabled: true
  config:
    - name: "critical"
      max_concurrency: 5
    - name: "batch"
      max_concurrency: 1

---
# workflow.yaml
name: critical-job
queue: critical  # Uses critical queue

---
# batch.yaml
name: batch-job
queue: batch  # Only 1 runs at a time

Distributed Execution

# ~/.config/dagu/config.yaml
distributed:
  enabled: true
  workers:
    - name: worker-1
      labels:
        - gpu
        - high-memory
    - name: worker-2
      labels:
        - cpu

---
# workflow.yaml
name: ml-training
steps:
  - id: train_model
    command: python train.py
    labels:
      - gpu
      - high-memory  # Runs only on worker-1

Git Sync

# ~/.config/dagu/config.yaml
git:
  enabled: true
  remote: https://github.com/org/dagu-workflows.git
  branch: main
  path: /home/dagu/workflows
  ssh_key: /home/dagu/.ssh/id_rsa
  sync_interval: 60  # Sync every 60 seconds

CLI Commands

# Start workflow
dagu start workflow.yaml
dagu start workflow.yaml -- P1=value1 P2=value2

# Dry run (simulate without executing)
dagu dry workflow.yaml

# Check status
dagu status workflow-name
dagu history workflow-name

# List workflows
dagu list

# Validate workflow
dagu validate workflow.yaml

# Stop running workflow
dagu stop workflow-name

# Retry failed workflow
dagu retry workflow-name --run-id=<id>

# View logs
dagu log workflow-name
dagu log workflow-name <run-id>

# Start scheduler
dagu scheduler

# Start web UI
dagu server

# Start both
dagu start-all

Best Practices

1. Idempotency: Design steps to be safe if rerun (retry logic)
2. Timeouts: Always set reasonable timeout_sec for long steps
3. Secrets: Use Dagu's secret store, never commit credentials
4. Logs: Capture important output to variables for debugging
5. Cleanup: Use handler_on.exit for guaranteed cleanup
6. Naming: Use descriptive step IDs and names
7. Documentation: Add descriptions to workflows and steps
8. Testing: Use dagu dry to validate before production
9. Monitoring: Set up failure notifications
10. Version Control: Store workflows in Git with Git Sync

Troubleshooting

Workflow not running

• Check scheduler: pgrep -f "dagu scheduler"
• Check schedule format (cron valid?)
• Check logs: dagu log workflow-name

Step keeps failing

• Check retry_policy configuration
• Verify exit_code values
• Check step timeout

Permission denied

• Check file permissions on DAG files
• Verify user running Dagu has access to commands

Scheduler won't start

• Check port 8090/8080 not in use
• Check ~/.config/dagu/ directory permissions
• Review logs in ~/.local/share/dagu/logs/

References

• Official Docs: https://docs.dagu.sh
• GitHub: https://github.com/dagu-org/dagu
• Examples: https://docs.dagu.sh/writing-workflows/examples
• Step Types: https://docs.dagu.sh/step-types/shell

Skill Usage Summary

When user asks to create a Dagu workflow:

1. Ask what the workflow should do (use case)
2. Determine if cron schedule or event-triggered needed
3. Identify steps required (shell, HTTP, Docker, etc.)
4. Define dependencies between steps
5. Add error handling (retry_policy, continue_on)
6. Add lifecycle handlers for notifications/cleanup
7. Test with dagu validate and dagu dry
8. Deploy to ~/.config/dagu/dags/
9. Start scheduler if not running