03-Argo Workflows 深度实践

# argo-install.yaml - 生产级部署配置
apiVersion: v1
kind: Namespace
metadata:
  name: argo
---
# Argo Workflows Controller
apiVersion: apps/v1
kind: Deployment
metadata:
  name: workflow-controller
  namespace: argo
spec:
  replicas: 2
  selector:
    matchLabels:
      app: workflow-controller
  template:
    metadata:
      labels:
        app: workflow-controller
    spec:
      serviceAccountName: argo-workflow-controller
      containers:
      - name: workflow-controller
        image: quay.io/argoproj/workflow-controller:v3.5.0
        args:
        - --configmap
        - workflow-controller-configmap
        - --executor-image
        - quay.io/argoproj/argoexec:v3.5.0
        - --namespaced
        - --managed-namespace
        - argo-workflows
        env:
        - name: LEADER_ELECTION_IDENTITY
          valueFrom:
            fieldRef:
              fieldPath: metadata.name
        resources:
          requests:
            cpu: 100m
            memory: 256Mi
          limits:
            cpu: 500m
            memory: 512Mi
        ports:
        - containerPort: 9090
          name: metrics
        livenessProbe:
          httpGet:
            path: /healthz
            port: 6060
          initialDelaySeconds: 30
          periodSeconds: 30
---
# Argo Server
apiVersion: apps/v1
kind: Deployment
metadata:
  name: argo-server
  namespace: argo
spec:
  replicas: 2
  selector:
    matchLabels:
      app: argo-server
  template:
    metadata:
      labels:
        app: argo-server
    spec:
      serviceAccountName: argo-server
      containers:
      - name: argo-server
        image: quay.io/argoproj/argocli:v3.5.0
        args:
        - server
        - --auth-mode=sso
        - --secure
        ports:
        - containerPort: 2746
          name: web
        env:
        - name: ARGO_SECURE
          value: "true"
        - name: ARGO_BASE_HREF
          value: /argo/
        resources:
          requests:
            cpu: 100m
            memory: 256Mi
          limits:
            cpu: 500m
            memory: 512Mi
        readinessProbe:
          httpGet:
            path: /
            port: 2746
            scheme: HTTPS
          initialDelaySeconds: 10
          periodSeconds: 10
---
# Controller ConfigMap
apiVersion: v1
kind: ConfigMap
metadata:
  name: workflow-controller-configmap
  namespace: argo
data:
  # 执行器配置
  executor: |
    resources:
      requests:
        cpu: 10m
        memory: 64Mi
      limits:
        cpu: 100m
        memory: 128Mi

  # Artifact 存储配置
  artifactRepository: |
    archiveLogs: true
    s3:
      bucket: argo-artifacts
      endpoint: minio.argo:9000
      insecure: true
      accessKeySecret:
        name: argo-artifacts
        key: accesskey
      secretKeySecret:
        name: argo-artifacts
        key: secretkey

  # 持久化配置
  persistence: |
    connectionPool:
      maxIdleConns: 100
      maxOpenConns: 0
    nodeStatusOffLoad: true
    archive: true
    archiveTTL: 30d
    postgresql:
      host: postgres.argo
      port: 5432
      database: argo
      tableName: argo_workflows
      userNameSecret:
        name: argo-postgres-config
        key: username
      passwordSecret:
        name: argo-postgres-config
        key: password

  # 资源限制
  workflowDefaults: |
    spec:
      ttlStrategy:
        secondsAfterCompletion: 86400
        secondsAfterSuccess: 3600
        secondsAfterFailure: 172800
      podGC:
        strategy: OnPodCompletion
      activeDeadlineSeconds: 86400

  # 并发控制
  parallelism: "50"

# workflow-templates.yaml
apiVersion: argoproj.io/v1alpha1
kind: WorkflowTemplate
metadata:
  name: ml-workflow-templates
  namespace: argo-workflows
spec:
  # 1. Container 模板
  templates:
  - name: container-example
    container:
      image: python:3.9
      command: [python]
      args: ["{{inputs.parameters.script}}"]
      resources:
        requests:
          memory: "1Gi"
          cpu: "500m"
        limits:
          memory: "2Gi"
          cpu: "1"
      volumeMounts:
      - name: workdir
        mountPath: /workdir
    inputs:
      parameters:
      - name: script
    outputs:
      artifacts:
      - name: output
        path: /workdir/output

  # 2. Script 模板
  - name: script-example
    script:
      image: python:3.9
      command: [python]
      source: |
        import json
        import sys

        data = json.loads('{{inputs.parameters.data}}')
        result = {"processed": data, "count": len(data)}

        with open('/tmp/result.json', 'w') as f:
            json.dump(result, f)

        print(json.dumps(result))
      resources:
        requests:
          memory: "512Mi"
          cpu: "200m"
    inputs:
      parameters:
      - name: data
    outputs:
      parameters:
      - name: result
        valueFrom:
          path: /tmp/result.json

  # 3. Resource 模板 - 创建 K8s 资源
  - name: resource-example
    resource:
      action: create
      setOwnerReference: true
      manifest: |
        apiVersion: batch/v1
        kind: Job
        metadata:
          generateName: pytorch-job-
        spec:
          template:
            spec:
              containers:
              - name: pytorch
                image: pytorch/pytorch:2.0.0-cuda11.7-cudnn8-runtime
                command: ["python", "-c", "print('Hello PyTorch')"]
              restartPolicy: Never

  # 4. Suspend 模板 - 人工审批
  - name: approval-gate
    suspend: {}

  # 5. HTTP 模板 - API 调用
  - name: http-example
    http:
      url: "https://api.example.com/webhook"
      method: POST
      headers:
      - name: Content-Type
        value: application/json
      body: |
        {"workflow": "{{workflow.name}}", "status": "{{inputs.parameters.status}}"}
      successCondition: response.statusCode == 200
    inputs:
      parameters:
      - name: status

---
# DAG 工作流模板
apiVersion: argoproj.io/v1alpha1
kind: WorkflowTemplate
metadata:
  name: ml-dag-template
  namespace: argo-workflows
spec:
  entrypoint: ml-pipeline

  arguments:
    parameters:
    - name: dataset-path
      value: "s3://data/training"
    - name: model-type
      value: "resnet50"
    - name: epochs
      value: "100"

  templates:
  - name: ml-pipeline
    dag:
      tasks:
      # 数据准备阶段
      - name: validate-data
        template: data-validation
        arguments:
          parameters:
          - name: path
            value: "{{workflow.parameters.dataset-path}}"

      - name: preprocess
        template: data-preprocessing
        dependencies: [validate-data]
        arguments:
          artifacts:
          - name: raw-data
            from: "{{tasks.validate-data.outputs.artifacts.validated-data}}"

      # 特征工程（并行）
      - name: feature-extraction
        template: extract-features
        dependencies: [preprocess]
        arguments:
          artifacts:
          - name: processed-data
            from: "{{tasks.preprocess.outputs.artifacts.processed-data}}"

      - name: feature-selection
        template: select-features
        dependencies: [feature-extraction]
        arguments:
          artifacts:
          - name: features
            from: "{{tasks.feature-extraction.outputs.artifacts.features}}"

      # 模型训练
      - name: train-model
        template: model-training
        dependencies: [feature-selection]
        arguments:
          parameters:
          - name: model-type
            value: "{{workflow.parameters.model-type}}"
          - name: epochs
            value: "{{workflow.parameters.epochs}}"
          artifacts:
          - name: training-data
            from: "{{tasks.feature-selection.outputs.artifacts.selected-features}}"

      # 模型评估
      - name: evaluate-model
        template: model-evaluation
        dependencies: [train-model]
        arguments:
          artifacts:
          - name: model
            from: "{{tasks.train-model.outputs.artifacts.model}}"
          - name: test-data
            from: "{{tasks.preprocess.outputs.artifacts.test-data}}"

      # 条件部署
      - name: deploy-model
        template: model-deployment
        dependencies: [evaluate-model]
        when: "{{tasks.evaluate-model.outputs.parameters.accuracy}} > 0.9"
        arguments:
          artifacts:
          - name: model
            from: "{{tasks.train-model.outputs.artifacts.model}}"

  # 模板定义
  - name: data-validation
    inputs:
      parameters:
      - name: path
    outputs:
      artifacts:
      - name: validated-data
        path: /data/validated
    container:
      image: ml-platform/data-validator:v1.0
      command: [python, validate.py]
      args:
      - --input={{inputs.parameters.path}}
      - --output=/data/validated

  - name: data-preprocessing
    inputs:
      artifacts:
      - name: raw-data
        path: /data/raw
    outputs:
      artifacts:
      - name: processed-data
        path: /data/processed
      - name: test-data
        path: /data/test
    container:
      image: ml-platform/preprocessor:v1.0
      command: [python, preprocess.py]
      args:
      - --input=/data/raw
      - --output=/data/processed
      - --test-output=/data/test

  - name: model-training
    inputs:
      parameters:
      - name: model-type
      - name: epochs
      artifacts:
      - name: training-data
        path: /data/train
    outputs:
      artifacts:
      - name: model
        path: /models/trained
      - name: metrics
        path: /metrics
    container:
      image: ml-platform/trainer:v1.0
      command: [python, train.py]
      args:
      - --model={{inputs.parameters.model-type}}
      - --epochs={{inputs.parameters.epochs}}
      - --data=/data/train
      - --output=/models/trained
      resources:
        requests:
          nvidia.com/gpu: 1
          memory: "16Gi"
          cpu: "4"
        limits:
          nvidia.com/gpu: 1
          memory: "32Gi"
          cpu: "8"

  - name: model-evaluation
    inputs:
      artifacts:
      - name: model
        path: /models
      - name: test-data
        path: /data/test
    outputs:
      parameters:
      - name: accuracy
        valueFrom:
          path: /metrics/accuracy.txt
      artifacts:
      - name: report
        path: /reports
    container:
      image: ml-platform/evaluator:v1.0
      command: [python, evaluate.py]

  - name: model-deployment
    inputs:
      artifacts:
      - name: model
        path: /models
    container:
      image: ml-platform/deployer:v1.0
      command: [python, deploy.py]

# dynamic-dag.yaml
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: dynamic-dag-
spec:
  entrypoint: main

  templates:
  - name: main
    steps:
    # 第一步：生成任务列表
    - - name: generate-tasks
        template: generate-task-list

    # 第二步：动态扩展执行
    - - name: process-tasks
        template: process-task
        arguments:
          parameters:
          - name: task-id
            value: "{{item}}"
        withParam: "{{steps.generate-tasks.outputs.result}}"

    # 第三步：聚合结果
    - - name: aggregate
        template: aggregate-results
        arguments:
          artifacts:
          - name: results
            from: "{{steps.process-tasks.outputs.artifacts.result}}"

  - name: generate-task-list
    script:
      image: python:3.9
      command: [python]
      source: |
        import json
        # 动态生成任务列表
        tasks = [f"task-{i}" for i in range(10)]
        print(json.dumps(tasks))

  - name: process-task
    inputs:
      parameters:
      - name: task-id
    outputs:
      artifacts:
      - name: result
        path: /tmp/result.json
    script:
      image: python:3.9
      command: [python]
      source: |
        import json
        task_id = "{{inputs.parameters.task-id}}"
        result = {"task_id": task_id, "status": "completed"}
        with open('/tmp/result.json', 'w') as f:
            json.dump(result, f)

  - name: aggregate-results
    inputs:
      artifacts:
      - name: results
        path: /tmp/results
    script:
      image: python:3.9
      command: [python]
      source: |
        import os
        import json

        results = []
        for f in os.listdir('/tmp/results'):
            with open(f'/tmp/results/{f}') as file:
                results.append(json.load(file))

        print(f"Aggregated {len(results)} results")
        print(json.dumps(results, indent=2))

---
# 条件分支与循环
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: conditional-loop-
spec:
  entrypoint: main
  arguments:
    parameters:
    - name: iterations
      value: "5"
    - name: threshold
      value: "0.8"

  templates:
  - name: main
    dag:
      tasks:
      - name: init
        template: initialize

      - name: training-loop
        template: training-iteration
        dependencies: [init]
        arguments:
          parameters:
          - name: iteration
            value: "{{item}}"
          - name: prev-model
            value: "{{tasks.init.outputs.parameters.model-path}}"
        withSequence:
          count: "{{workflow.parameters.iterations}}"

      - name: final-eval
        template: final-evaluation
        dependencies: [training-loop]

  - name: initialize
    outputs:
      parameters:
      - name: model-path
        valueFrom:
          path: /tmp/model-path
    script:
      image: python:3.9
      command: [python]
      source: |
        model_path = "/models/initial"
        with open('/tmp/model-path', 'w') as f:
            f.write(model_path)
        print(f"Initialized model at {model_path}")

  - name: training-iteration
    inputs:
      parameters:
      - name: iteration
      - name: prev-model
    outputs:
      parameters:
      - name: accuracy
        valueFrom:
          path: /tmp/accuracy
    script:
      image: python:3.9
      command: [python]
      source: |
        import random

        iteration = {{inputs.parameters.iteration}}
        prev_model = "{{inputs.parameters.prev-model}}"

        # 模拟训练
        accuracy = 0.7 + (iteration * 0.05) + random.uniform(0, 0.1)

        with open('/tmp/accuracy', 'w') as f:
            f.write(str(accuracy))

        print(f"Iteration {iteration}: accuracy = {accuracy:.4f}")

  - name: final-evaluation
    script:
      image: python:3.9
      command: [python]
      source: |
        print("Final evaluation completed")

# distributed-training-workflow.yaml
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: distributed-training-
spec:
  entrypoint: distributed-pipeline

  arguments:
    parameters:
    - name: num-workers
      value: "4"
    - name: model
      value: "llama-7b"
    - name: dataset
      value: "s3://data/pretrain"

  volumes:
  - name: shared-data
    persistentVolumeClaim:
      claimName: training-data-pvc

  templates:
  - name: distributed-pipeline
    dag:
      tasks:
      - name: prepare-data
        template: data-preparation
        arguments:
          parameters:
          - name: dataset
            value: "{{workflow.parameters.dataset}}"
          - name: num-shards
            value: "{{workflow.parameters.num-workers}}"

      - name: launch-master
        template: training-master
        dependencies: [prepare-data]
        arguments:
          parameters:
          - name: num-workers
            value: "{{workflow.parameters.num-workers}}"
          - name: model
            value: "{{workflow.parameters.model}}"

      - name: launch-workers
        template: training-worker
        dependencies: [prepare-data, launch-master]
        arguments:
          parameters:
          - name: worker-id
            value: "{{item}}"
          - name: master-addr
            value: "{{tasks.launch-master.outputs.parameters.master-addr}}"
        withSequence:
          count: "{{workflow.parameters.num-workers}}"

      - name: wait-completion
        template: wait-for-training
        dependencies: [launch-workers]
        arguments:
          parameters:
          - name: master-addr
            value: "{{tasks.launch-master.outputs.parameters.master-addr}}"

      - name: collect-checkpoints
        template: checkpoint-collection
        dependencies: [wait-completion]

      - name: merge-model
        template: model-merging
        dependencies: [collect-checkpoints]

  - name: data-preparation
    inputs:
      parameters:
      - name: dataset
      - name: num-shards
    outputs:
      artifacts:
      - name: data-manifest
        path: /output/manifest.json
    container:
      image: ml-platform/data-prep:v1.0
      command: [python, shard_data.py]
      args:
      - --input={{inputs.parameters.dataset}}
      - --num-shards={{inputs.parameters.num-shards}}
      - --output=/output
      volumeMounts:
      - name: shared-data
        mountPath: /output
      resources:
        requests:
          cpu: "4"
          memory: "16Gi"

  - name: training-master
    inputs:
      parameters:
      - name: num-workers
      - name: model
    outputs:
      parameters:
      - name: master-addr
        valueFrom:
          path: /tmp/master-addr
    # 使用 daemon 模式保持 master 运行
    daemon: true
    container:
      image: ml-platform/distributed-trainer:v1.0
      command: [python, master.py]
      args:
      - --model={{inputs.parameters.model}}
      - --num-workers={{inputs.parameters.num-workers}}
      - --port=29500
      env:
      - name: MASTER_ADDR
        valueFrom:
          fieldRef:
            fieldPath: status.podIP
      - name: MASTER_PORT
        value: "29500"
      ports:
      - containerPort: 29500
      resources:
        requests:
          nvidia.com/gpu: 1
          cpu: "8"
          memory: "32Gi"
        limits:
          nvidia.com/gpu: 1
          memory: "64Gi"
      volumeMounts:
      - name: shared-data
        mountPath: /data

  - name: training-worker
    inputs:
      parameters:
      - name: worker-id
      - name: master-addr
    container:
      image: ml-platform/distributed-trainer:v1.0
      command: [python, worker.py]
      args:
      - --master-addr={{inputs.parameters.master-addr}}
      - --master-port=29500
      - --worker-id={{inputs.parameters.worker-id}}
      - --data-shard=/data/shard-{{inputs.parameters.worker-id}}
      env:
      - name: WORLD_SIZE
        value: "{{workflow.parameters.num-workers}}"
      - name: RANK
        value: "{{inputs.parameters.worker-id}}"
      resources:
        requests:
          nvidia.com/gpu: 1
          cpu: "8"
          memory: "32Gi"
        limits:
          nvidia.com/gpu: 1
          memory: "64Gi"
      volumeMounts:
      - name: shared-data
        mountPath: /data

  - name: wait-for-training
    inputs:
      parameters:
      - name: master-addr
    script:
      image: curlimages/curl:latest
      command: [sh]
      source: |
        while true; do
          status=$(curl -s http://{{inputs.parameters.master-addr}}:29500/status)
          if [ "$status" = "completed" ]; then
            echo "Training completed"
            break
          fi
          sleep 60
        done

  - name: checkpoint-collection
    outputs:
      artifacts:
      - name: checkpoints
        path: /checkpoints
    container:
      image: ml-platform/checkpoint-collector:v1.0
      command: [python, collect.py]
      volumeMounts:
      - name: shared-data
        mountPath: /data

  - name: model-merging
    container:
      image: ml-platform/model-merger:v1.0
      command: [python, merge.py]
      args:
      - --checkpoints=/data/checkpoints
      - --output=/models/final
      resources:
        requests:
          cpu: "16"
          memory: "64Gi"
      volumeMounts:
      - name: shared-data
        mountPath: /data

# argo-monitoring.yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: argo-prometheus-rules
  namespace: monitoring
data:
  argo-workflows.yaml: |
    groups:
    - name: argo-workflows
      rules:
      # 工作流状态指标
      - record: argo_workflow_status_phase
        expr: |
          sum by (namespace, phase) (
            argo_workflows_count{phase!=""}
          )

      # 工作流成功率
      - record: argo_workflow_success_rate
        expr: |
          sum(rate(argo_workflows_count{phase="Succeeded"}[1h])) /
          sum(rate(argo_workflows_count{phase=~"Succeeded|Failed|Error"}[1h]))

      # 平均工作流执行时间
      - record: argo_workflow_duration_seconds_avg
        expr: |
          avg by (workflow_name) (
            argo_workflows_pods_duration_seconds_sum /
            argo_workflows_pods_duration_seconds_count
          )

      # 工作流排队时间
      - record: argo_workflow_queue_duration_seconds_avg
        expr: |
          avg(argo_workflows_queue_duration_seconds_sum /
              argo_workflows_queue_duration_seconds_count)

      # 告警规则
      - alert: ArgoWorkflowHighFailureRate
        expr: |
          (1 - argo_workflow_success_rate) > 0.1
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "High workflow failure rate"
          description: "Workflow failure rate is above 10%"

      - alert: ArgoWorkflowStuck
        expr: |
          argo_workflows_count{phase="Running"} > 0
          and
          changes(argo_workflows_count{phase="Running"}[30m]) == 0
        for: 30m
        labels:
          severity: critical
        annotations:
          summary: "Workflow appears stuck"
          description: "Workflow has been in Running state for 30+ minutes without change"

      - alert: ArgoWorkflowQueueBacklog
        expr: |
          argo_workflows_count{phase="Pending"} > 50
        for: 15m
        labels:
          severity: warning
        annotations:
          summary: "Large workflow queue backlog"
          description: "More than 50 workflows pending"

      - alert: ArgoControllerDown
        expr: |
          up{job="workflow-controller"} == 0
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "Argo Workflow Controller is down"
          description: "The workflow controller has been down for 5 minutes"

---
# Grafana Dashboard ConfigMap
apiVersion: v1
kind: ConfigMap
metadata:
  name: argo-grafana-dashboard
  namespace: monitoring
  labels:
    grafana_dashboard: "1"
data:
  argo-workflows.json: |
    {
      "title": "Argo Workflows Dashboard",
      "uid": "argo-workflows",
      "panels": [
        {
          "title": "Workflow Status Distribution",
          "type": "piechart",
          "targets": [
            {
              "expr": "sum by (phase) (argo_workflows_count)",
              "legendFormat": "{{phase}}"
            }
          ]
        },
        {
          "title": "Workflow Success Rate (1h)",
          "type": "gauge",
          "targets": [
            {
              "expr": "argo_workflow_success_rate * 100"
            }
          ],
          "options": {
            "thresholds": {
              "steps": [
                {"color": "red", "value": 0},
                {"color": "yellow", "value": 80},
                {"color": "green", "value": 95}
              ]
            }
          }
        },
        {
          "title": "Workflow Execution Time (p95)",
          "type": "timeseries",
          "targets": [
            {
              "expr": "histogram_quantile(0.95, rate(argo_workflows_pods_duration_seconds_bucket[5m]))",
              "legendFormat": "p95"
            },
            {
              "expr": "histogram_quantile(0.50, rate(argo_workflows_pods_duration_seconds_bucket[5m]))",
              "legendFormat": "p50"
            }
          ]
        },
        {
          "title": "Active Workflows",
          "type": "timeseries",
          "targets": [
            {
              "expr": "argo_workflows_count{phase='Running'}",
              "legendFormat": "Running"
            },
            {
              "expr": "argo_workflows_count{phase='Pending'}",
              "legendFormat": "Pending"
            }
          ]
        },
        {
          "title": "Workflow Completion Rate",
          "type": "timeseries",
          "targets": [
            {
              "expr": "sum(rate(argo_workflows_count{phase='Succeeded'}[5m]))",
              "legendFormat": "Succeeded"
            },
            {
              "expr": "sum(rate(argo_workflows_count{phase='Failed'}[5m]))",
              "legendFormat": "Failed"
            }
          ]
        },
        {
          "title": "Resource Usage by Workflow",
          "type": "table",
          "targets": [
            {
              "expr": "topk(10, sum by (workflow) (argo_workflows_pods_cpu_usage_seconds_total))",
              "legendFormat": "{{workflow}}"
            }
          ]
        }
      ]
    }