Workflow
1. Argo workflow
1.1 Install
Code block
1 helm repo add argo [Link]
2 helm repo update
3 helm pull argo/argo-workflows --version 0.47.4
4 helm template argo/argo-workflows --version 0.47.4 | grep "image:"
5 # 在集群测试通
6 sudo /var/lib/rancher/rke2/bin/ctr -a "/run/k3s/containerd/[Link]" -n
"[Link]" image pull ***********************
7
8 helm pull argo/argo-workflows --untar
9 helm package argo-workflows/
10 helm push [Link] oci://[Link]/ch
yaml
1 controller:
2 workflowDefaults:
3 spec:
4 podGC:
5 strategy: OnWorkflowCompletion
6 deleteDelayDuration: "5m" 结束后5m删除pod
#workflow
7 ttlStrategy:
8 secondsAfterSuccess: 300 成功后5m删除，失败24h删除
#workflow
9 secondsAfterFailure: 86400
10
11 server:
12 authMode: server
13 authModes:
14 - server
15
16 servicePort: 2746
17 serviceTargetPort: 2746
18 serviceType: NodePort
19 serviceNodePort: 32333
[Link]
• Test
Code block
1 apiVersion: [Link]/v1alpha1
2 kind: Workflow
3 metadata:
4 generateName: demo-resource-flow-
5 namespace: argocd-test
6 spec:
7 entrypoint: main
8
9 templates:
10 - name: main
11 steps:
12 - - name: step-1-start
13 template: create-configmap
14 arguments:
15 parameters:
16 - name: cm-name
17 value: workflow-step-1-start
18 - name: message
19 value: "step 1: workflow started"
20
21 - - name: step-2-prepare
22 template: create-configmap
23 arguments:
24 parameters:
25 - name: cm-name
26 value: workflow-step-2-prepare
27 - name: message
28 value: "step 2: prepare data"
29
30 - - name: step-3-branch-a
31 template: create-configmap
32 arguments:
33 parameters:
34 - name: cm-name
35 value: workflow-step-3-branch-a
36 - name: message
37 value: "step 3A: process branch A"
38 - name: step-3-branch-b
39 template: create-configmap
40 arguments:
41 parameters:
42 - name: cm-name
43 value: workflow-step-3-branch-b
44 - name: message
45 value: "step 3B: process branch B"
46
47 - - name: step-4-finish
48 template: create-configmap
49 arguments:
50 parameters:
51 - name: cm-name
52 value: workflow-step-4-finish
53 - name: message
54 value: "step 4: workflow finished"
55
56 - name: create-configmap
57 inputs:
58 parameters:
59 - name: cm-name
60 - name: message
61 resource:
62 action: create
63 manifest: |
64 apiVersion: v1
65 kind: ConfigMap
66 metadata:
67 name: {{[Link]-name}}
68 namespace: argocd-test
69 labels:
70 app: argo-workflow-demo
71 data:
72 message: "{{[Link]}}"
73
74 ttlStrategy:
75 secondsAfterCompletion: 600
1.2 CLI
找和k8s相同releases的CLI
Code block
1 # Download the binary
2 curl -sLO "[Link]
workflows/releases/download/v3.7.10/[Link]"
3
4 # Unzip
5 gunzip "[Link]"
6
7 # Make binary executable
8 chmod +x "argo-linux-amd64"
9
10 # Move binary to path
11 sudo mv "./argo-linux-amd64" /usr/local/bin/argo
12
13 # Test installation
14 argo version
2. schema
2.1 Ticket
History生成payload返回payload_url，redirect到CEP时前端带上payload_url传给CEP后端创建
ticket
Name Type Description
Ticket id Uint Not null
Owner String
Create timestamp bigint
Start timestamp bigint
End timestamp bigint
Status Enum Pending，Running，Failed，Succeeded，
Terminated，Cancelled
Priority Enum 可实现根据优先级插队，P0/1/2/3 Not null
Payload url String Not null
Description String
2.2 Workflow
Name Type Description
Workflow id Uint Not null
Ticket id Uint Not null
Git commit String Not null
Git branch String Not null
Workflow type string Not null
Owner String
Create timestamp bigint
Start timestamp bigint
End timestamp biginy
Status Enum Pending，Running，Failed，Succeeded，
Terminated，Cancelled
Stage Enum
Total Uint
Succeeded Uint
Failed Uint
Running Uint
Pending Uint
error_message String
Description String
3. Workflow
Code block
1 mkdir -p /tmp/cep_payloads
2
3 cat > /tmp/cep_payloads/test_payload_type.json <<'EOF'
4 {
5 "scenario_names": [
6 "GSS_30001.1779351739875904688-PLANNING",
7 "GSS_30001.1779423774726617433-PERCEPTION"
8 ],
9 "scenarios": [
10 {
11 "scenario_name": "GSS_30001.1779351739875904688-PLANNING",
12 "scenario_tier": "CANDIDATE",
13 "owner": "pengwenzheng",
14 "incident_id": "GSS_30001.1779351739875904688",
15 "create_timestamp": 1779352592057440800,
16 "update_timestamp": 1779680554893488400,
17 "start_time": 0,
18 "sim_time": 0,
19 "end_time": 0,
20 "scenario_type": "PLANNING",
21 "description": null,
22 "planning_metrics": [],
23 "s3_gzidc_fullbag_url": "[Link]
[Link]/browser/govy-pilot-incident-gzidc/fullbag/2026-05-
21/GSS_30001/GSS_30001.[Link]"
24 },
25 {
26 "scenario_name": "GSS_30001.1779423774726617433-PERCEPTION",
27 "scenario_tier": "EXPERIMENTAL",
28 "owner": "pengwenzheng",
29 "incident_id": "GSS_30001.1779423774726617433",
30 "create_timestamp": 1780285583212648700,
31 "update_timestamp": 1780285583212648700,
32 "start_time": 0,
33 "sim_time": 1,
34 "end_time": 25.299999237060547,
35 "scenario_type": "PERCEPTION",
36 "description": null,
37 "planning_metrics": [],
38 "s3_gzidc_fullbag_url": "[Link]
[Link]/browser/govy-pilot-incident-gzidc/fullbag/2026-05-
22/GSS_30001/GSS_30001.[Link]"
39 }
40 ]
41 }
42 EOF
43
44 cat > /tmp/cep_payloads/test_payload.json <<'EOF'
45 {
46 "scenario_names": [
47 "GSS_30001.1779351739875904688-PLANNING"
48 ],
49 "scenarios": [
50 {
51 "scenario_name": "GSS_30001.1779351739875904688-PLANNING",
52 "scenario_tier": "CANDIDATE",
53 "owner": "pengwenzheng",
54 "incident_id": "GSS_30001.1779351739875904688",
55 "create_timestamp": 1779352592057440800,
56 "update_timestamp": 1779680554893488400,
57 "start_time": 0,
58 "sim_time": 0,
59 "end_time": 0,
60 "scenario_type": "PLANNING",
61 "description": null,
62 "planning_metrics": [],
63 "s3_gzidc_fullbag_url": "[Link]
[Link]/browser/govy-pilot-incident-gzidc/fullbag/2026-05-
21/GSS_30001/GSS_30001.[Link]"
64 }
65 ]
66 }
67 EOF
68
69 curl -X POST "[Link] \
70 -H "Content-Type: application/json" \
71 -d '{
72 "payload_url": "[Link]
73 "owner": "test_user",
74 "priority": 3,
75 "description": "create ticket only"
76 }'
77
78
79 cat > /tmp/cep_payloads/test_payload_sametype.json <<'EOF'
80 {
81 "scenario_names": [
82 "GSS_30001.1779351739875904688-PLANNING",
83 "GSS_30001.1779433246250136994-PLANNING"
84 ],
85 "scenarios": [
86 {
87 "scenario_name": "GSS_30001.1779351739875904688-PLANNING",
88 "scenario_tier": "CANDIDATE",
89 "owner": "pengwenzheng",
90 "incident_id": "GSS_30001.1779351739875904688",
91 "create_timestamp": 1779352592057440800,
92 "update_timestamp": 1779680554893488400,
93 "start_time": 0,
94 "sim_time": 0,
95 "end_time": 0,
96 "scenario_type": "PLANNING",
97 "description": null,
98 "planning_metrics": [],
99 "s3_gzidc_fullbag_url": "[Link]
[Link]/browser/govy-pilot-incident-gzidc/fullbag/2026-05-
21/GSS_30001/GSS_30001.[Link]"
100 },
101 {
102 "scenario_name": "GSS_30001.1779433246250136994-PLANNING",
103 "scenario_tier": "EXPERIMENTAL",
104 "owner": "pengwenzheng",
105 "incident_id": "GSS_30001.1779433246250136994",
106 "create_timestamp": 1780285610864629200,
107 "update_timestamp": 1780285610864629200,
108 "start_time": 0,
109 "sim_time": 1,
110 "end_time": 21,
111 "scenario_type": "PLANNING",
112 "description": null,
113 "planning_metrics": [],
114 "s3_gzidc_fullbag_url": "[Link]
[Link]/browser/govy-pilot-incident-gzidc/fullbag/2026-05-
22/GSS_30001/GSS_30001.[Link]"
115 },
116 {
117 "scenario_name": "GSS_30001.1779432688535733307-PLANNING",
118 "scenario_tier": "EXPERIMENTAL",
119 "owner": "pengwenzheng",
120 "incident_id": "GSS_30001.1779432688535733307",
121 "create_timestamp": 1780285635932308500,
122 "update_timestamp": 1780285635932308500,
123 "start_time": 0,
124 "sim_time": 1,
125 "end_time": 28.799999237060547,
126 "scenario_type": "PLANNING",
127 "description": null,
128 "planning_metrics": [],
129 "s3_gzidc_fullbag_url": "[Link]
[Link]/browser/govy-pilot-incident-gzidc/fullbag/2026-05-
22/GSS_30001/GSS_30001.[Link]"
130 }
131 ]
132 }
133 EOF
134
135 curl -X POST "[Link] \
136 -H "Content-Type: application/json" \
137 -d '{
138 "payload_url": "[Link]
139 "owner": "test_user",
140 "priority": 3,
141 "description": "create ticket only"
142 }'
143
3.1 Validate and Parse
创建前发起检查，payload校验失败则不创建workflow
Payload
payload_url 是否存在且可访问
下载payload
payload 是否为空
Payload schema
根据type分组校验
Scenario
校验 scenario_name
校验 incident_id
校验 scenario_type（前置）
创建 task
3.2 Create workflow
Code block
1 curl -X POST "[Link] \
2 -H "Content-Type: application/json" \
3 -d '{
4 "ticket_id": "178053870846198393",
5 "owner": "test-user",
6 "description": "test real argo workflow creation",
7 "workflow_configs": {
8 "PLANNING": {
9 "git_commit": "a2c29285c9ef6b13133e717055261de25d3ca44e",
10 "git_branch": "infra/runtime_docker_payload"
11 }
12 }
13 }'
14
根据workflow_type创建workflow
3.3 Steps
3.3.1 Check and Build
Code block
1 export HARBOR_USERNAME='admin'
2 export HARBOR_PASSWORD='govy@123'
3
4 export GITLAB_URL='[Link]
5 export GITLAB_PROJECT_ID='227'
6 export GITLAB_REF='infra/runtime_docker_payload'
7
8 export GITLAB_TRIGGER_TOKEN='glptt-ab6d06485c297ca0270a98d939254f8ddabd9322'
9 export GITLAB_PRIVATE_TOKEN='glpat-fzy2xtVfDsTnH6uXcJ2J'
10
11 export GITLAB_VERIFY_SSL=false
12 export PIPELINE_POLL_INTERVAL_SECONDS=10
13 export PIPELINE_TIMEOUT_SECONDS=3600
根据 git_commit 生成 image_tag = commit 前 8 位，image name 为
[Link]/euler/euler-runtime
查询 [Link]/euler/euler-runtime:${image_tag} 是否存在
如果存在则跳过 build，如果不存在则触发 CI build
并发workflow时不重复创建(Redis)
Hook捕捉pipeline success（gitlab setting)
polling轮询
设置 build_timeout
超时后标记 build failed
记录Image地址
3.3.2 Parse
输出json数组 -> 动态 fan-out： withParam
按照车架号分
单个车架号内按batch_size二次分配
3.3.3 Fan-Out Simulation
[Link] simulation
使用 Check and Build 产出的 image_url
手动batch Dagger
根据 workflow_type 设置 per_ticket_parallelism
每个 Task 传入自己的 task payload？
simulation container 启动 dagger
simulation container 写 exit_code
simulation container 写 stdout/stderr
simulation container 写 error_message
Code block
1 for task in task_ids:
2 download record from s3 to /tmp/
3 start_timestamp=now
4 dagger cyber/tools/dagger/conf/planning_template.json -i
~/tmp/aircraft_001.[Link] -o ~/tmp/result/
5 if success:
6 upload COS bag/log
7 else:
8 upload COS log
9 write error_message
10 send completion event to redis stream
11 -task_id
12 -task_status
13 -task_bag/log_url
14 -task end_timestamp
单个task挂：抓error&message
[Link] finalize
读取 simulation exit_code，stdout/stderr，判断 simulation 是否成功
每完成一个task,成功则上传COS bag,log，失败则不上传 bag,上传 log
更新 task DB( status,output_url,error_message）（轮询蠢，高频，找更优雅办法）
redis stream
1 simulation:completed Stream
2 ↓
3 Group A: db-writers → 回写数据库
4 Group B: metrics-workers → 统计指标
finalize 自身失败要可重试
[Link] 监控
1. Prepare
修改task model 新增s3_gzidc_fullbag_url字段
从payload获取s3 url，传建task时传入task
phase环节传递task_ids和bag_url
For task in task_ids
从 S3 下载 record
存到 /work/<task_id>/input/
创建输出目录 /work/<task_id>/output
创建日志目录 /work/<task_id>/log
写start_timestamp，向redis stream发送running
运行dagger
运行结束后
◦ 成功 → 上传 result bag + log
◦ 失败 → 只上传 log，并从 log 提取 error_message
无论成功失败 → 发送 Redis Stream 占位事件
有running无ending -> 卡死
2. 运行dagger
dagger cyber/tools/dagger/conf/planning_template.json -i
~/Downloads/record/aircraft_001.[Link] -o ~/Downloads/record/
Configmap 挂载 for脚本
Code block
1 #!/bin/sh
2 set +e
3
4 TASK_IDS="$1"
5 WORK_ROOT="${2:-/work}"
6 DAGGER_CONFIG="${3:-cyber/tools/dagger/conf/planning_template.json}"
7
8 OLD_IFS="$IFS"
9 IFS=","
10
11 for task_id in $TASK_IDS; do
12 task_root="${WORK_ROOT}/${task_id}"
13 input="${task_root}/input/${task_id}.record"
14 output_dir="${task_root}/result"
15 log_dir="${task_root}/log"
16 log_file="${log_dir}/${task_id}.log"
17 exit_code_file="${task_root}/dagger_exit_code.txt"
18
19 mkdir -p "${output_dir}" "${log_dir}"
20
21 {
22 echo "[task] task_id=${task_id}"
23 echo "[simulation] start_timestamp=$(date +%s%3N)"
24 echo "[simulation] command=dagger ${DAGGER_CONFIG} -i ${input} -o
${output_dir}"
25 } >> "${log_file}"
26
27 dagger "${DAGGER_CONFIG}" \
28 -i "${input}" \
29 -o "${output_dir}" \
30 >> "${log_file}" 2>&1
31
32 rc=$?
33
34 {
35 echo "[simulation] end_timestamp=$(date +%s%3N)"
36 echo "[simulation] return_code=${rc}"
37 } >> "${log_file}"
38
39 # exit_code 最后写。watcher 看到这个文件，就认为该 task 完成。
40 echo "${rc}" > "${exit_code_file}"
41
42 # Pod 总日志里只打印摘要，方便 Argo UI 看进度。
43 echo "[fan-out-simulation] task_id=${task_id} return_code=${rc}"
44 done
45
46 IFS="$OLD_IFS"
47
48 # 返回 0，让 watcher 有机会处理所有 task 的上传和事件。
49 exit 0
遍历prepare [Link]获取input/output path
运行dagger
每个dagger运行完写work/dagger_exit_code.txt
3. Watch finalize
如果 dagger_exit_code 出现，就说明 task 已经跑完
根据 exit_code 判断成功或失败：
• 成功 → 上传 result bag + log
• 失败 → 只上传 log，并从 log 提取 error_message
无论成功失败 → 发送 Redis Stream 占位事件
3.3.4 Aggregate
查询 workflow_id 下所有 tasks，重算ticket total/succeeded/failed
更新 workflow status，End timestamp
如果所有workflow结束，更新TIcket状态/end_timestamp
3.4 Workflow sync
查询Pending/Running workflow_status / workflow_stage并同步到数据库
如果phase=failed(failed,terminated,cancelled)，写error_message
Workflow total/succeeded/failed
如果failed大于某个数量 kill
3.5 DB Writer