Argo Workflows Reference: DAG, Steps, Artifacts, CronWorkflow & K8s ML Pipelines

# Install Argo Workflows:
kubectl create namespace argo
kubectl apply -n argo -f https://github.com/argoproj/argo-workflows/releases/latest/download/install.yaml

# Install argo CLI:
brew install argo           # macOS
# Or: curl -sLO https://github.com/argoproj/argo-workflows/releases/latest/download/argo-linux-amd64.gz

# Expose the UI:
kubectl -n argo port-forward deploy/argo-server 2746:2746
# Open: http://localhost:2746

# List workflows:
argo list -n argo
argo get my-workflow -n argo
argo logs my-workflow -n argo

apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: pipeline-   # auto-generates unique name: pipeline-xxxxx
  namespace: argo
spec:
  entrypoint: main          # which template to run first
  arguments:
    parameters:
      - name: env
        value: production   # default value; override: argo submit --parameter env=staging

  templates:
    # Main template — orchestrates other templates as steps:
    - name: main
      steps:
        - - name: fetch-data            # sequential: step 1
            template: fetch
            arguments:
              parameters: [{name: env, value: "{{workflow.parameters.env}}"}]
        - - name: process-a             # parallel: runs simultaneously
            template: process
            arguments: {parameters: [{name: shard, value: "a"}]}
          - name: process-b             # also in step 2 (same indent = parallel)
            template: process
            arguments: {parameters: [{name: shard, value: "b"}]}
        - - name: aggregate             # sequential: step 3 (after both 2a + 2b)
            template: aggregate

    # Leaf templates — actual work (each runs as a K8s pod):
    - name: fetch
      inputs:
        parameters:
          - name: env
      container:
        image: my-fetcher:latest
        command: [python, fetch.py]
        args: ["--env", "{{inputs.parameters.env}}"]

    - name: process
      inputs:
        parameters: [{name: shard}]
      container:
        image: my-processor:latest
        command: [python, process.py, "--shard", "{{inputs.parameters.shard}}"]

    - name: aggregate
      container:
        image: my-aggregator:latest
        command: [python, aggregate.py]

templates:
  - name: main-dag
    dag:
      tasks:
        - name: ingest
          template: run-job
          arguments: {parameters: [{name: cmd, value: "python ingest.py"}]}

        - name: validate
          dependencies: [ingest]       # runs after ingest completes
          template: run-job
          arguments: {parameters: [{name: cmd, value: "python validate.py"}]}

        - name: transform-a
          dependencies: [validate]     # runs after validate
          template: run-job
          arguments: {parameters: [{name: cmd, value: "python transform.py --part a"}]}

        - name: transform-b
          dependencies: [validate]     # also runs after validate (parallel with transform-a)
          template: run-job
          arguments: {parameters: [{name: cmd, value: "python transform.py --part b"}]}

        - name: load
          dependencies: [transform-a, transform-b]   # waits for BOTH to finish
          template: run-job
          arguments: {parameters: [{name: cmd, value: "python load.py"}]}

  - name: run-job
    inputs:
      parameters: [{name: cmd}]
    container:
      image: my-pipeline:latest
      command: [sh, -c]
      args: ["{{inputs.parameters.cmd}}"]

# Artifacts — pass files between steps (S3, GCS, or Minio):
templates:
  - name: produce
    outputs:
      artifacts:
        - name: result
          path: /tmp/result.json         # file written by the container
    container:
      image: my-producer:latest
      command: [python, produce.py]      # writes to /tmp/result.json

  - name: consume
    inputs:
      artifacts:
        - name: result
          from: "{{tasks.produce.outputs.artifacts.result}}"  # DAG reference
          path: /tmp/input.json          # where to mount the artifact in this container
    container:
      image: my-consumer:latest
      command: [python, consume.py, --input, /tmp/input.json]

# Artifacts config (minio/S3):
spec:
  artifactRepositoryRef:
    configMap: artifact-repositories
    key: default
  # ConfigMap contains S3/GCS endpoint + bucket + credentials

# Volume claim template (PVC per workflow run):
spec:
  volumeClaimTemplates:
    - metadata: {name: work}
      spec:
        accessModes: [ReadWriteOnce]
        resources: {requests: {storage: 10Gi}}
  templates:
    - name: my-step
      container:
        volumeMounts:
          - name: work
            mountPath: /work

# Resource template (create/delete K8s resources as workflow steps):
templates:
  - name: create-job
    resource:
      action: create
      successCondition: status.succeeded > 0
      failureCondition: status.failed > 0
      manifest: |
        apiVersion: batch/v1
        kind: Job
        metadata: {generateName: my-job-}
        spec:
          template:
            spec:
              containers: [{name: main, image: my-image:latest, command: [python, run.py]}]
              restartPolicy: Never

# Submit a workflow:
argo submit workflow.yaml -n argo                          # submit and watch
argo submit workflow.yaml -n argo --wait                   # wait for completion
argo submit workflow.yaml -n argo -p env=staging           # override parameter
argo submit workflow.yaml -n argo --from=cronworkflow/my-cron  # run a cron now

# Monitor:
argo list -n argo                       # all workflows + status
argo get my-workflow-xxx -n argo        # detailed status
argo logs my-workflow-xxx -n argo       # all pod logs
argo logs my-workflow-xxx -n argo -c main  # specific container
argo watch my-workflow-xxx -n argo      # live status updates

# Retry failed workflows:
argo retry my-workflow-xxx -n argo      # retry from last failed node
argo retry my-workflow-xxx -n argo --restart-successful  # retry ALL nodes from scratch

# CronWorkflow (schedule a workflow):
apiVersion: argoproj.io/v1alpha1
kind: CronWorkflow
metadata:
  name: daily-pipeline
  namespace: argo
spec:
  schedule: "0 2 * * *"               # 2am daily
  timezone: Europe/London
  concurrencyPolicy: Forbid            # don't start new run if previous still running
  workflowSpec:
    entrypoint: main
    templates: [...]

# Suspend/resume a cron:
argo cron suspend daily-pipeline -n argo
argo cron resume daily-pipeline -n argo