Argo Workflows 是一个开源的容器本机工作流引擎，用于在 Kubernetes 上协调并行作业。 Argo Workflows 被实现为 Kubernetes CRD（自定义资源定义）。跟其他传统的工作流引擎不同的是，他的每一个步骤都是一个容器。将多步骤工作流建模为一系列任务，或者使用有向无环图（DAG）捕获任务之间的依赖关系。
使用 Kubernetes 上的 Argo Workflow，可以在短时间内轻松运行用于计算机学习或数据处理的计算密集型作业。
安装 argo
安装 argo 十分容易
第一步先创建 namespace
kubectl create ns argo
第二步执行 kubectl -n argo apply -f install.yaml
# This is an auto-generated file. DO NOT EDIT
apiVersion: apiextensions.k8s.io/v1beta1
kind: CustomResourceDefinition
metadata:
name: clusterworkflowtemplates.argoproj.io
spec:
group: argoproj.io
names:
kind: ClusterWorkflowTemplate
listKind: ClusterWorkflowTemplateList
plural: clusterworkflowtemplates
shortNames:
- clusterwftmpl
- cwft
singular: clusterworkflowtemplate
scope: Cluster
version: v1alpha1
versions:
- name: v1alpha1
served: true
storage: true
---
apiVersion: apiextensions.k8s.io/v1beta1
kind: CustomResourceDefinition
metadata:
name: cronworkflows.argoproj.io
spec:
group: argoproj.io
names:
kind: CronWorkflow
listKind: CronWorkflowList
plural: cronworkflows
shortNames:
- cwf
- cronwf
singular: cronworkflow
scope: Namespaced
version: v1alpha1
versions:
- name: v1alpha1
served: true
storage: true
---
apiVersion: apiextensions.k8s.io/v1beta1
kind: CustomResourceDefinition
metadata:
name: workfloweventbindings.argoproj.io
spec:
group: argoproj.io
names:
kind: WorkflowEventBinding
listKind: WorkflowEventBindingList
plural: workfloweventbindings
shortNames:
- wfeb
singular: workfloweventbinding
scope: Namespaced
version: v1alpha1
versions:
- name: v1alpha1
served: true
storage: true
---
apiVersion: apiextensions.k8s.io/v1beta1
kind: CustomResourceDefinition
metadata:
name: workflows.argoproj.io
spec:
additionalPrinterColumns:
- JSONPath: .status.phase
description: Status of the workflow
name: Status
type: string
- JSONPath: .status.startedAt
description: When the workflow was started
format: date-time
name: Age
type: date
group: argoproj.io
names:
kind: Workflow
listKind: WorkflowList
plural: workflows
shortNames:
- wf
singular: workflow
scope: Namespaced
subresources: {}
version: v1alpha1
versions:
- name: v1alpha1
served: true
storage: true
---
apiVersion: apiextensions.k8s.io/v1beta1
kind: CustomResourceDefinition
metadata:
name: workflowtemplates.argoproj.io
spec:
group: argoproj.io
names:
kind: WorkflowTemplate
listKind: WorkflowTemplateList
plural: workflowtemplates
shortNames:
- wftmpl
singular: workflowtemplate
scope: Namespaced
version: v1alpha1
versions:
- name: v1alpha1
served: true
storage: true
---
apiVersion: v1
kind: ServiceAccount
metadata:
name: argo
---
apiVersion: v1
kind: ServiceAccount
metadata:
name: argo-server
---
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
name: argo-role
rules:
- apiGroups:
- ""
resources:
- secrets
verbs:
- get
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
labels:
rbac.authorization.k8s.io/aggregate-to-admin: "true"
name: argo-aggregate-to-admin
rules:
- apiGroups:
- argoproj.io
resources:
- workflows
- workflows/finalizers
- workflowtemplates
- workflowtemplates/finalizers
- cronworkflows
- cronworkflows/finalizers
- clusterworkflowtemplates
- clusterworkflowtemplates/finalizers
verbs:
- create
- delete
- deletecollection
- get
- list
- patch
- update
- watch
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
labels:
rbac.authorization.k8s.io/aggregate-to-edit: "true"
name: argo-aggregate-to-edit
rules:
- apiGroups:
- argoproj.io
resources:
- workflows
- workflows/finalizers
- workflowtemplates
- workflowtemplates/finalizers
- cronworkflows
- cronworkflows/finalizers
- clusterworkflowtemplates
- clusterworkflowtemplates/finalizers
verbs:
- create
- delete
- deletecollection
- get
- list
- patch
- update
- watch
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
labels:
rbac.authorization.k8s.io/aggregate-to-view: "true"
name: argo-aggregate-to-view
rules:
- apiGroups:
- argoproj.io
resources:
- workflows
- workflows/finalizers
- workflowtemplates
- workflowtemplates/finalizers
- cronworkflows
- cronworkflows/finalizers
- clusterworkflowtemplates
- clusterworkflowtemplates/finalizers
verbs:
- get
- list
- watch
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
name: argo-cluster-role
rules:
- apiGroups:
- ""
resources:
- pods
- pods/exec
verbs:
- create
- get
- list
- watch
- update
- patch
- delete
- apiGroups:
- ""
resources:
- configmaps
verbs:
- get
- watch
- list
- apiGroups:
- ""
resources:
- persistentvolumeclaims
verbs:
- create
- delete
- apiGroups:
- argoproj.io
resources:
- workflows
- workflows/finalizers
verbs:
- get
- list
- watch
- update
- patch
- delete
- create
- apiGroups:
- argoproj.io
resources:
- workflowtemplates
- workflowtemplates/finalizers
- clusterworkflowtemplates
- clusterworkflowtemplates/finalizers
verbs:
- get
- list
- watch
- apiGroups:
- ""
resources:
- serviceaccounts
verbs:
- get
- list
- apiGroups:
- argoproj.io
resources:
- cronworkflows
- cronworkflows/finalizers
verbs:
- get
- list
- watch
- update
- patch
- delete
- apiGroups:
- ""
resources:
- events
verbs:
- create
- patch
- apiGroups:
- policy
resources:
- poddisruptionbudgets
verbs:
- create
- get
- delete
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
name: argo-server-cluster-role
rules:
- apiGroups:
- ""
resources:
- configmaps
verbs:
- get
- watch
- list
- apiGroups:
- ""
resources:
- secrets
verbs:
- get
- apiGroups:
- ""
resources:
- pods
- pods/exec
- pods/log
verbs:
- get
- list
- watch
- delete
- apiGroups:
- ""
resources:
- events
verbs:
- watch
- create
- patch
- apiGroups:
- ""
resources:
- secrets
- serviceaccounts
verbs:
- get
- apiGroups:
- argoproj.io
resources:
- workflows
- workfloweventbindings
- workflowtemplates
- cronworkflows
- clusterworkflowtemplates
verbs:
- create
- get
- list
- watch
- update
- patch
- delete
---
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
name: argo-binding
roleRef:
apiGroup: rbac.authorization.k8s.io
kind: Role
name: argo-role
subjects:
- kind: ServiceAccount
name: argo
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
name: argo-binding
roleRef:
apiGroup: rbac.authorization.k8s.io
kind: ClusterRole
name: argo-cluster-role
subjects:
- kind: ServiceAccount
name: argo
namespace: argo
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
name: argo-server-binding
roleRef:
apiGroup: rbac.authorization.k8s.io
kind: ClusterRole
name: argo-server-cluster-role
subjects:
- kind: ServiceAccount
name: argo-server
namespace: argo
---
apiVersion: v1
kind: ConfigMap
metadata:
name: workflow-controller-configmap
---
apiVersion: v1
kind: Service
metadata:
name: argo-server
spec:
externalIPs: # 暴露Service到外部IP
- 192.168.1.146 # IP
ports:
- name: web
port: 2746
targetPort: 2746
selector:
app: argo-server
---
apiVersion: v1
kind: Service
metadata:
name: workflow-controller-metrics
spec:
ports:
- name: metrics
port: 9090
protocol: TCP
targetPort: 9090
selector:
app: workflow-controller
---
apiVersion: apps/v1
kind: Deployment
metadata:
name: argo-server
spec:
selector:
matchLabels:
app: argo-server
template:
metadata:
labels:
app: argo-server
spec:
containers:
- args:
- server
image: argoproj/argocli:v2.11.7
name: argo-server
ports:
- containerPort: 2746
name: web
readinessProbe:
httpGet:
path: /
port: 2746
scheme: HTTP
initialDelaySeconds: 10
periodSeconds: 20
volumeMounts:
- mountPath: /tmp
name: tmp
nodeSelector:
kubernetes.io/os: linux
serviceAccountName: argo-server
volumes:
- emptyDir: {}
name: tmp
---
apiVersion: apps/v1
kind: Deployment
metadata:
name: workflow-controller
spec:
selector:
matchLabels:
app: workflow-controller
template:
metadata:
labels:
app: workflow-controller
spec:
containers:
- args:
- --configmap
- workflow-controller-configmap
- --executor-image
- argoproj/argoexec:v2.11.7
command:
- workflow-controller
image: argoproj/workflow-controller:v2.11.7
name: workflow-controller
nodeSelector:
kubernetes.io/os: linux
serviceAccountName: argo
第三步 安装 argo-cli
# Download the binary
curl -sLO https://github.com/argoproj/argo/releases/download/v2.11.7/argo-darwin-amd64.gz
# Unzip
gunzip argo-darwin-amd64.gz
# Make binary executable
chmod +x argo-darwin-amd64
# Move binary to path
mv ./argo-darwin-amd64 /usr/local/bin/argo
# Test installation
argo version
如出现已下输出则 argo-cli 安装成功 我安装的是老版本 不用担心 是向下兼容的
然后使用 kubectl get all -n argo 查看 argo-server 的启动情况
启动完毕后，我们可以访问暴露出来的 argo-server-ui 界面访问 argo-dashboard 默认端口 2746
遇到的问题
安装完 argo 并且提交工作流后，发现工作流执行不成功，查看日志发现
这是因为默认的工作流 pod 是在默认的 namespace 也就是 default 下执行的，而这个 namespace 下名为 default 的默认的 serviceAccount 不具备操作资源的权限，则我们可以给他绑定权限
kind: ClusterRole
apiVersion: rbac.authorization.k8s.io/v1
metadata:
namespace: default
name: pod-reader
rules:
- apiGroups:
- ""
resources:
- configmaps
verbs:
- get
- watch
- list
- apiGroups:
- "batch"
resources:
- jobs
verbs:
- get
- watch
- list
- create
- apiGroups:
- ""
resources:
- secrets
verbs:
- get
- apiGroups:
- ""
resources:
- pods
- pods/exec
- pods/log
verbs:
- get
- list
- watch
- delete
- patch
- apiGroups:
- ""
resources:
- events
verbs:
- watch
- create
- patch
- apiGroups:
- ""
resources:
- secrets
- serviceaccounts
verbs:
- get
- apiGroups:
- argoproj.io
resources:
- workflows
- workfloweventbindings
- workflowtemplates
- cronworkflows
- clusterworkflowtemplates
verbs:
- create
- get
- list
- watch
- update
- patch
- delete
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
name: pod-reader-pod
namespace: default
roleRef:
apiGroup: rbac.authorization.k8s.io
kind: ClusterRole
name: pod-reader
subjects:
- kind: ServiceAccount
name: default
namespace: default
然后执行 kubectl apply -f role.yaml 就可以解决这个问题
我们也可以在创建工作流的 yaml 指定我们创建好的有权限的 serviceAccount
Hello World
我们可以使用官网的例子简单的开始第一个 Workflow，一个 Workflow 基本也是这样的构造
- generateName： workflow 会在 k8s 环境内产生一个 job 来执行 workflow(job 指的是 k8s 中一定会结束的任务)，然后 job 则会产生以 generateName 规定的字符为前缀的 pod(比如上面的例子，它会产生如 whalesay-abcde 字样的 pod)
- entrypoint：这里规定了一个入口，即我们的 workflow 会以哪一个模板作为第一个模板来启动
- templates：定义了 templates，templates 是 Argo 中比较重要的一块，我们的 Workflow 运行都是基于各种 template
argo-helloworld
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
generateName: whalesay-
spec:
entrypoint: whalesay
templates:
- name: whalesay # name of the template
container:
image: docker/whalesay
command: [cowsay]
args: ["hello world"]
以上是一个官网给出的 helloworld 样例。它规定了 workflow 会调用一个 docker/whalesay 的容器来打印 helloworld。使用以下命令可以启动这个 Workflow
argo submit helloworld.yaml
就可以看到 workflow 的启动，登录 ui 也可以看到此时有 workflow 的执行，打印 helloworld
查看执行结果
使用 argo watch 持续监控工作流状态
使用 argo logs 查看日志输出
步骤操作
步骤操作包含了比较多的模块类型，它提供了各种常见的对于步骤之间的操作，基本满足了我们对于步骤间操作的需求
container
这也是一个最常见的 templates 类型，它会创建一个容器，然后使用容器来完成我们的任务
- name: whalesay # name of the template
container:
image: docker/whalesay
command: [cowsay]
args: ["hello world"]
script
有时我们只希望我们的模板来运行一个脚本，那么 Argo 提供了 Scripts 来让我们运行脚本。
Script 允许我们使用 source 标签来创建一个脚本(临时文件)，然后这个临时文件的名称将会作为参数传递给 command 来执行。
使用 script，会将运行脚本的标准输出分配给输出参数 result，让其他的步骤来调用
针对不同的开发着，我们可以用不同的镜像来支持不同的执行方式
# shell脚本
- name: gen-random-int-bash
script:
image: debian:9.4
command: [bash]
source: | # Contents of the here-script
cat /dev/urandom | od -N2 -An -i | awk -v f=1 -v r=100 '{printf "%i\n", f + r * $1 / 65536}'
# python脚本
- name: gen-random-int-python
script:
image: python:alpine3.6
command: [python]
source: |
import random
i = random.randint(1, 100)
print(i)
# js脚本
- name: gen-random-int-javascript
script:
image: node:9.1-alpine
command: [node]
source: |
var rand = Math.floor(Math.random() * 100);
console.log(rand);
steps
steps 规定了执行的步骤，有并行也有串行模式，它以双横杠(- -)的形式来定义串行，然后以单横杠的形式来定义并行
- name: hello-hello-hello
steps:
- - name: hello1 # hello1 is run before the following steps
template: whalesay
arguments:
parameters:
- name: message
value: "hello1"
- - name: hello2a # double dash => run after previous step
template: whalesay
arguments:
parameters:
- name: message
value: "hello2a"
- name: hello2b # single dash => run in parallel with previous step
template: whalesay
arguments:
parameters:
- name: message
value: "hello2b"
dag
DAG 是一个有向无环图，Argo 使用 DAG 来定义一些比较复杂的 workflow 关系
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
generateName: dag-diamond-
spec:
entrypoint: diamond
templates:
- name: echo
inputs:
parameters:
- name: message
container:
image: alpine:3.7
command: [echo, "{{inputs.parameters.message}}"]
- name: diamond
dag:
tasks:
- name: A
template: echo
arguments:
parameters: [{name: message, value: A}]
- name: B
dependencies: [A]
template: echo
arguments:
parameters: [{name: message, value: B}]
- name: C
dependencies: [A]
template: echo
arguments:
parameters: [{name: message, value: C}]
- name: D
dependencies: [B, C]
template: echo
arguments:
parameters: [{name: message, value: D}]
如上所示，上面的 dag 定义了一个钻石类型的图 A -> (B C) -> D
loop
使用 loop 我们可以定义循环
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
generateName: loops-
spec:
entrypoint: loop-example
templates:
- name: loop-example
steps:
- - name: print-message
template: whalesay
arguments:
parameters:
- name: message
value: "{{item}}"
withItems: # invoke whalesay once for each item in parallel
- hello world # item 1
- goodbye world # item 2
- name: whalesay
inputs:
parameters:
- name: message
container:
image: docker/whalesay:latest
command: [cowsay]
args: ["{{inputs.parameters.message}}"]
这个例子中，我们通过 withItems 传入了两个参数，然后 Workflow 就会并行执行这个 templates 两次，依次使用我们给出的参数
当然我们也可以动态的使用循环，循环的参数基于其他模板来控制
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
generateName: loops-param-result-
spec:
entrypoint: loop-param-result-example
templates:
- name: loop-param-result-example
steps:
- - name: generate
template: gen-number-list
# Iterate over the list of numbers generated by the generate step above
- - name: sleep
template: sleep-n-sec
arguments:
parameters:
- name: seconds
value: "{{item}}"
withParam: "{{steps.generate.outputs.result}}"
# Generate a list of numbers in JSON format
- name: gen-number-list
script:
image: python:alpine3.6
command: [python]
source: |
import json
import sys
json.dump([i for i in range(20, 31)], sys.stdout)
- name: sleep-n-sec
inputs:
parameters:
- name: seconds
container:
image: alpine:latest
command: [sh, -c]
args: ["echo sleeping for {{inputs.parameters.seconds}} seconds; sleep {{inputs.parameters.seconds}}; echo done"]
条件控制
条件的控制需要用到 when 的关键字
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
generateName: coinflip-
spec:
entrypoint: coinflip
templates:
- name: coinflip
steps:
# flip a coin
- - name: flip-coin
template: flip-coin
# evaluate the result in parallel
- - name: heads
template: heads # call heads template if "heads"
when: "{{steps.flip-coin.outputs.result}} == heads"
- name: tails
template: tails # call tails template if "tails"
when: "{{steps.flip-coin.outputs.result}} == tails"
# Return heads or tails based on a random number
- name: flip-coin
script:
image: python:alpine3.6
command: [python]
source: |
import random
result = "heads" if random.randint(0,1) == 0 else "tails"
print(result)
- name: heads
container:
image: alpine:3.6
command: [sh, -c]
args: ["echo \"it was heads\""]
- name: tails
container:
image: alpine:3.6
command: [sh, -c]
args: ["echo \"it was tails\""]
在这个例子中，Workflow 通过 when 来判断第一步获取的值是 head 还是 tails，根据获取的值来条件判断下一步会执行的步骤
重试
重试模块会定义如果 Job 执行出现 failures 或 errors 时的情况，
- limit:指重试的最大次数
- retryOn:指定重启策略
- Always: errors 和 failures 时重启
- OnFailure: failures 时重启，默认采用
- OnError： error 时重启
- backoff:定义重启的一些参数
# This example demonstrates the use of retry back offs
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
generateName: retry-backoff-
spec:
entrypoint: retry-backoff
templates:
- name: retry-backoff
retryStrategy:
limit: 10
retryPolicy: "Always"
backoff:
duration: "1" # Must be a string. Default unit is seconds. Could also be a Duration, e.g.: "2m", "6h", "1d"
factor: 2
maxDuration: "1m" # Must be a string. Default unit is seconds. Could also be a Duration, e.g.: "2m", "6h", "1d"
container:
image: python:alpine3.6
command: ["python", -c]
# fail with a 66% probability
args: ["import random; import sys; exit_code = random.choice([0, 1, 1]); sys.exit(exit_code)"]
以上的例子定义了一个会重启的任务
递归
argo 也是支持递归的,我们只要将步骤的下一步指定为本步骤即可
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
generateName: coinflip-recursive-
spec:
entrypoint: coinflip
templates:
- name: coinflip
steps:
# flip a coin
- - name: flip-coin
template: flip-coin
# evaluate the result in parallel
- - name: heads
template: heads # call heads template if "heads"
when: "{{steps.flip-coin.outputs.result}} == heads"
- name: tails # keep flipping coins if "tails"
template: coinflip
when: "{{steps.flip-coin.outputs.result}} == tails"
- name: flip-coin
script:
image: python:alpine3.6
command: [python]
source: |
import random
result = "heads" if random.randint(0,1) == 0 else "tails"
print(result)
- name: heads
container:
image: alpine:3.6
command: [sh, -c]
args: ["echo \"it was heads\""]
在以上的步骤中，step 执行会做一个判断，如果一直是反面，会递归执行直到为正面
Exit handlers
Argo 可以定义一个出口，它会直接退出工作流，无论成功与否。
常见于:
- 工作流运行后清理
- 发送工作流状态的通知（例如，电子邮件/ Slack）
- 将通过/失败状态发布到 webhook 结果（例如 GitHub 构建结果）
- 重新提交或提交另一个工作流程
spec:
entrypoint: intentional-fail
onExit: exit-handler # invoke exit-hander template at end of the workflow
templates:
- name: exit-handler
steps:
- - name: notify
template: send-email
- name: celebrate
template: celebrate
when: "{{workflow.status}} == Succeeded"
- name: cry
template: cry
when: "{{workflow.status}} != Succeeded"
在这里使用 Onexit 参数指定了结束模板，则执行 exit-handler 这个模板时，无论是否成功，都会直接结束
timeout
Argo 定义了一个超时的限定，如果容器超时了，则直接结束 Job
- name: sleep
container:
image: alpine:latest
command: [sh, -c]
args: ["echo sleeping for 1m; sleep 60; echo done"]
activeDeadlineSeconds: 10
suspending
- name: approve
suspend: {}
- name: delay
suspend:
duration: 20
如果我们使用了 duration 关键字，它会等待 20 秒后才唤醒执行
argo resume WORKFLOW # 这里的Workflow是指workflow的pod容器名
参数操作
参数传递对于工作流也是一个比较关键的问题。对于工作流来说，不同 tempaltes 之间的传递，是通过 jinja 来定义。目前 Argo 只接受以下几种前缀
- item
- steps
- inputs
- outputs
- workflow
- tasks
parameters
通常的参数传递是通过 parameters 关键字来定义的
- name: whalesay
inputs:
parameters:
- name: message # parameter declaration
container:
# run cowsay with that message input parameter as args
image: docker/whalesay
command: [cowsay]
args: ["{{inputs.parameters.message}}"]
arguments
我们可以通过 arguments 关键字来定义一个全局参数
entrypoint: whalesay
arguments:
parameters:
- name: message
value: hello world
- name: os-list # a list of items
value: |
[
{ "image": "debian", "tag": "9.1" },
{ "image": "debian", "tag": "8.9" },
{ "image": "alpine", "tag": "3.6" },
{ "image": "ubuntu", "tag": "17.10" }
]
在启动任务时，我们可以通过-p 的参数来做实际参数，如果没有指定，则会使用默认参数（argument 中定义）
argo submit arguments.yaml -p messgae="helloworld" -p oslist=[{ "image": "ubuntu", "tag": "17.10" }]
step 间的传参
对于 step 间的传参，是通过 step 关键字来定义的
- name: test
steps:
- - name: A
template: A
- - name: B
template: B
when: "\"{{steps.A.outputs.result}}\" == \"B\""
- name: C
template: C
when: "\"{{steps.A.outputs.result}}\" == \"C\""
在以上例子中，定义了一个 step，它会首先执行 A 步骤，然后根据启动结果，如果输出是 B，则执行 B 步骤，否则执行 C 步骤
dag 间的传参
对于 Dag，其实它与 step 是十分相似的
dag:
tasks:
- name: ip
template: param
arguments:
parameters: [{name: request, value: "ip"}, {name: ip, value: "{{inputs.parameters.ip}}"}]
- name: port
template: param
arguments:
parameters: [{name: request, value: "port"}, {name: ip, value: "{{inputs.parameters.ip}}"}]
- name: username
template: param
arguments:
parameters: [{name: request, value: "username"}, {name: ip, value: "{{inputs.parameters.ip}}"}]
- name: password
template: param
arguments:
parameters: [{name: request, value: "password"}, {name: ip, value: "{{inputs.parameters.ip}}"}]
- name: server
template: server
dependencies: [ip, port, username, password]
arguments:
parameters:
- name: ip
value: "{{tasks.ip.outputs.result}}"
- name: password
value: "{{tasks.password.outputs.result}}"
- name: username
value: "{{tasks.username.outputs.result}}"
- name: port
value: "{{tasks.port.outputs.result}}"
在上面的例子中，dag 并行执行(ip, port, username, password)四个步骤，然后将执行的结果传递给 server 模块，然后 server 模块会以这四个参数来完成工作。
script result
当我们运行一个 script 时，运行的标准输出会以 result 的方式来传递
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
generateName: scripts-bash-
spec:
entrypoint: bash-script-example
templates:
- name: bash-script-example
steps:
- - name: generate
template: gen-random-int-bash
- - name: print
template: print-message
arguments:
parameters:
- name: message
value: "{{steps.generate.outputs.result}}" # The result of the here-script
- name: gen-random-int-bash
script:
image: debian:9.4
command: [bash]
source: | # Contents of the here-script
cat /dev/urandom | od -N2 -An -i | awk -v f=1 -v r=100 '{printf "%i\n", f + r * $1 / 65536}'
- name: gen-random-int-python
script:
image: python:alpine3.6
command: [python]
source: |
import random
i = random.randint(1, 100)
print(i)
- name: gen-random-int-javascript
script:
image: node:9.1-alpine
command: [node]
source: |
var rand = Math.floor(Math.random() * 100);
console.log(rand);
- name: print-message
inputs:
parameters:
- name: message
container:
image: alpine:latest
command: [sh, -c]
args: ["echo result was: {{inputs.parameters.message}}"]
比如上面的例子，generate 这个模块执行完成之后，print 模块会获取 generate 模块的输出结果作为参数来执行
output Parameters
我们可以使用步骤的输出作为参数传递
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
generateName: output-parameter-
spec:
entrypoint: output-parameter
templates:
- name: output-parameter
steps:
- - name: generate-parameter
template: whalesay
- - name: consume-parameter
template: print-message
arguments:
parameters:
# Pass the hello-param output from the generate-parameter step as the message input to print-message
- name: message
value: "{{steps.generate-parameter.outputs.parameters.hello-param}}"
- name: whalesay
container:
image: docker/whalesay:latest
command: [sh, -c]
args: ["echo -n hello world > /tmp/hello_world.txt"] # generate the content of hello_world.txt
outputs:
parameters:
- name: hello-param # name of output parameter
valueFrom:
path: /tmp/hello_world.txt # set the value of hello-param to the contents of this hello-world.txt
- name: print-message
inputs:
parameters:
- name: message
container:
image: docker/whalesay:latest
command: [cowsay]
args: ["{{inputs.parameters.message}}"]
在上面的例子中，whalesay 模块会将执行结果打印到 hello-world.txt 文本，然后将这个文本的内容定为输出结果然后 consume-parameter 模块会去获取 whalesay 模块的输出结果做为输入参数
资源操作
Argo 可以操作的资源也有很多，它不仅仅是能操作容器，kubernetes 的资源、容器资源、计算资源等也均可调配
Secret
volumes:
- name: my-secret-vol
secret:
secretName: my-secret # name of an existing k8s secret
templates:
- name: whalesay
container:
image: alpine:3.7
command: [sh, -c]
args: ['
echo "secret from env: $MYSECRETPASSWORD";
echo "secret from file: `cat /secret/mountpath/mypassword`"
']
env:
- name: MYSECRETPASSWORD # name of env var
valueFrom:
secretKeyRef:
name: my-secret # name of an existing k8s secret
key: mypassword # 'key' subcomponent of the secret
volumeMounts:
- name: my-secret-vol # mount file containing secret at /secret/mountpath
mountPath: "/secret/mountpath"
在这里，我们使用了 Secret 作为一个 volume 供模板调用
daemon
使用守护进程可以使得作业在后台运行，他们的存在使得有的工作可以跨工作流运行
- name: influxdb
daemon: true # start influxdb as a daemon
retryStrategy:
limit: 10 # retry container if it fails
container:
image: influxdb:1.2
readinessProbe: # wait for readinessProbe to succeed
httpGet:
path: /ping
port: 8086
在这里使用了 daemon: true 来开启 daemon，以保护 influxdb 持续运行
sidecar
边车模式，指著容器在同一容器中同时执行另一个容器来支持主容器的工作。Argo 也支持边车模式，可以启动一个辅助容器来协助作业的进行
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
generateName: sidecar-nginx-
spec:
entrypoint: sidecar-nginx-example
templates:
- name: sidecar-nginx-example
container:
image: appropriate/curl
command: [sh, -c]
# Try to read from nginx web server until it comes up
args: ["until `curl -G 'http://127.0.0.1/' >& /tmp/out`; do echo sleep && sleep 1; done && cat /tmp/out"]
# Create a simple nginx web server
sidecars:
- name: nginx
image: nginx:1.13
Artifacts
在 Argo 中也有集成一些比较常用的库，比如 http 和 git
templates:
- name: hardwired-artifact
inputs:
artifacts:
# Check out the master branch of the argo repo and place it at /src
# revision can be anything that git checkout accepts: branch, commit, tag, etc.
- name: argo-source
path: /src
git:
repo: https://github.com/argoproj/argo.git
revision: "master"
# Download kubectl 1.8.0 and place it at /bin/kubectl
- name: kubectl
path: /bin/kubectl
mode: 0755
http:
url: https://storage.googleapis.com/kubernetes-release/release/v1.8.0/bin/linux/amd64/kubectl
Resource
Argo 可以操作 k8s 资源 创建一个 job
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
generateName: k8s-jobs-
spec:
entrypoint: pi-tmpl
templates:
- name: pi-tmpl
resource: # indicates that this is a resource template
action: create # can be any kubectl action (e.g. create, delete, apply, patch)
successCondition: status.succeeded > 0
failureCondition: status.failed > 3
manifest: | #put your kubernetes spec here
apiVersion: batch/v1
kind: Job
metadata:
generateName: pi-job-
spec:
template:
metadata:
name: pi
spec:
containers:
- name: pi
image: perl
command: ["perl", "-Mbignum=bpi", "-wle", "print bpi(2000)"]
restartPolicy: Never
backoffLimit: 4
可以使用以下 Argo Workflow 修改此 Crontab：
apiVersion: "stable.example.com/v1"
kind: CronTab
spec:
cronSpec: "* * * * */5"
image: my-awesome-cron-image
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
generateName: k8s-patch-
spec:
entrypoint: cront-tmpl
templates:
- name: cront-tmpl
resource:
action: patch
mergeStrategy: merge # Must be one of [strategic merge json]
manifest: |
apiVersion: "stable.example.com/v1"
kind: CronTab
spec:
cronSpec: "* * * * */10"
image: my-awesome-cron-image
docker
Argo 实现了 Docker in Docker 的形式
- name: dind-sidecar-example
container:
image: docker:17.10
command: [sh, -c]
args: ["until docker ps; do sleep 3; done; docker run --rm debian:latest cat /etc/os-release"]
env:
- name: DOCKER_HOST # the docker daemon can be access on the standard port on localhost
value: 127.0.0.1
sidecars:
- name: dind
image: docker:17.10-dind # Docker already provides an image for running a Docker daemon
securityContext:
privileged: true # the Docker daemon can only run in a privileged container
# mirrorVolumeMounts will mount the same volumes specified in the main container
# to the sidecar (including artifacts), at the same mountPaths. This enables
# dind daemon to (partially) see the same filesystem as the main container in
# order to use features such as docker volume binding.
mirrorVolumeMounts: true
以上工作流实现了主容器运行等待容器创建，边车容器启动一个容器并将这个容器镜像交给主容器资源
volumes
在 Argo 中，我们也可以直接传递容器卷，方便处理大量数据
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
generateName: volumes-pvc-
spec:
entrypoint: volumes-pvc-example
volumeClaimTemplates: # define volume, same syntax as k8s Pod spec
- metadata:
name: workdir # name of volume claim
spec:
accessModes: [ "ReadWriteOnce" ]
resources:
requests:
storage: 1Gi # Gi => 1024 * 1024 * 1024
templates:
- name: volumes-pvc-example
steps:
- - name: generate
template: whalesay
- - name: print
template: print-message
- name: whalesay
container:
image: docker/whalesay:latest
command: [sh, -c]
args: ["echo generating message in volume; cowsay hello world | tee /mnt/vol/hello_world.txt"]
# Mount workdir volume at /mnt/vol before invoking docker/whalesay
volumeMounts: # same syntax as k8s Pod spec
- name: workdir
mountPath: /mnt/vol
- name: print-message
container:
image: alpine:latest
command: [sh, -c]
args: ["echo getting message from volume; find /mnt/vol; cat /mnt/vol/hello_world.txt"]
# Mount workdir volume at /mnt/vol before invoking docker/whalesay
volumeMounts: # same syntax as k8s Pod spec
- name: workdir
mountPath: /mnt/vol
在上面的例子中，workflow 初始化了一个容器卷，然后下面的 whalesay 和 print-message 模块都调用了这个容器卷
不过大多数情况下，我们会去调用一个已存在的卷
# Define Kubernetes PVC
kind: PersistentVolumeClaim
apiVersion: v1
metadata:
name: my-existing-volume
spec:
accessModes: [ "ReadWriteOnce" ]
resources:
requests:
storage: 1Gi
---
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
generateName: volumes-existing-
spec:
entrypoint: volumes-existing-example
volumes:
# Pass my-existing-volume as an argument to the volumes-existing-example template
# Same syntax as k8s Pod spec
- name: workdir
persistentVolumeClaim:
claimName: my-existing-volume
templates:
- name: volumes-existing-example
steps:
- - name: generate
template: whalesay
- - name: print
template: print-message
- name: whalesay
container:
image: docker/whalesay:latest
command: [sh, -c]
args: ["echo generating message in volume; cowsay hello world | tee /mnt/vol/hello_world.txt"]
volumeMounts:
- name: workdir
mountPath: /mnt/vol
- name: print-message
container:
image: alpine:latest
command: [sh, -c]
args: ["echo getting message from volume; find /mnt/vol; cat /mnt/vol/hello_world.txt"]
volumeMounts:
- name: workdir
mountPath: /mnt/vol
在上面的例子中，我们外部已经定义了一个 pvc，然后在 workflow 中,我们通过声明一个 pvc 为卷来调用它
总结
argo 是一个云原生的基于 k8s 的工作流引擎，如果基础环境是 k8s 的话，不管是 ci/cd 还是其他工作流用途，argo 都是非常好的选择，上手非常简单，使用 yaml 作为模板语法 与 k8s 几乎一模一样。
欢迎来到这里！
我们正在构建一个小众社区，大家在这里相互信任，以平等 • 自由 • 奔放的价值观进行分享交流。最终，希望大家能够找到与自己志同道合的伙伴，共同成长。
注册 关于