如何分析Argo Workflows-Kubernetes的工作流引擎

魁首哥

作者

如何分析Argo Workflows-Kubernetes的工作流引擎

如何分析Argo Workflows-Kubernetes的工作流引擎，很多新手对此不是很清楚，为了帮助大家解决这个难题，下面小编将为大家详细讲解，有这方面需求的人可以来学习下，希望你能有所收获。

什么是Argo Workflows?

Argo Workflows是一个开源项目，为Kubernetes提供container-native工作流程，其主要通过Kubernetes CRD实现的。

特点如下：

工作流的每一步都是一个容器
将多步骤工作流建模为一系列任务，或者使用有向无环图(DAG)描述任务之间的依赖关系
可以在短时间内轻松运行用于机器学习或数据处理的计算密集型作业
在Kubernetes上运行CI/CD Pipeline，无需复杂的软件配置

安装

安装控制器端

Argo Wordflows的安装非常简单，直接使用以下命令安装即可。

kubectlcreatensargokubectlapply-nargo-fhttps://raw.githubusercontent.com/argoproj/argo-workflows/stable/manifests/quick-start-postgres.yaml

安装完成后，会生成以下4个pod。

#kubectlgetpo-nargoNAMEREADYSTATUSRESTARTSAGEargo-server-574ddc66b-62rjc1/1Running44h35mminio1/1Running04h35mpostgres-56fd897cf4-k8fwd1/1Running04h35mworkflow-controller-77658c77cc-p25ll1/1Running44h35m

其中：

argo-server是argo服务端
mino是进行制品仓库
postgres是数据库
workflow-controller是流程控制器

然后配置一个server端的ingress，即可访问UI，配置清单如下(我这里使用的是traefik)：

apiVersion:traefik.containo.us/v1alpha1kind:IngressRoutemetadata:name:argo-uinamespace:argospec:entryPoints:-webroutes:-match:Host(`argowork-test.coolops.cn`)kind:Ruleservices:-name:argo-serverport:2746

UI界面如下：

再配置一个minio的ingress，配置清单如下：

apiVersion:traefik.containo.us/v1alpha1kind:IngressRoutemetadata:name:minionamespace:argospec:entryPoints:-webroutes:-match:Host(`minio-test.coolops.cn`)kind:Ruleservices:-name:minioport:9000

UI界面如下（默认用户名密码是：admin:password）：

安装Client端

Argo Workflows提供Argo CLI，其安装方式也非常简单，如下：Linux系统：

#Downloadthebinarycurl-sLOhttps://github.com/argoproj/argo/releases/download/v3.0.0-rc4/argo-linux-amd64.gz#Unzipgunzipargo-linux-amd64.gz#Makebinaryexecutablechmod+xargo-linux-amd64#Movebinarytopathmv./argo-linux-amd64/usr/local/bin/argo

安装完成后，使用以下命令校验是否安装成功。

#argoversionargo:v3.0.0-rc4BuildDate:2021-03-02T21:42:55ZGitCommit:ae5587e97dad0e4806f7a230672b998fe140a767GitTreeState:cleanGitTag:v3.0.0-rc4GoVersion:go1.13Compiler:gcPlatform:linux/amd64

其主要的命令有：

list列出工作流logs查看工作流的日志submit创建工作流watch实时监听工作流get现实详细信息delete删除工作流stop停止工作流

更多命令可以使用argo --help进行查看。

然后可以使用一个简单的hello world的WorkFlow，如下：

apiVersion:argoproj.io/v1alpha1kind:Workflowmetadata:generateName:hello-world-labels:workflows.argoproj.io/archive-strategy:"false"spec:entrypoint:whalesaytemplates:-name:whalesaycontainer:image:docker/whalesay:latestcommand:[cowsay]args:["helloworld"]

使用如下命令创建并观察workflow。

$argosubmit-nargohelloworld.yaml--watch

然后可以看到以下输出。

Name:hello-world-9pw7vNamespace:argoServiceAccount:defaultStatus:SucceededConditions:CompletedTrueCreated:MonMar0814:51:35+0800(10secondsago)Started:MonMar0814:51:35+0800(10secondsago)Finished:MonMar0814:51:45+0800(now)Duration:10secondsProgress:1/1ResourcesDuration:4s*(1cpu),4s*(100Mimemory)STEPTEMPLATEPODNAMEDURATIONMESSAGE✔hello-world-9pw7vwhalesayhello-world-9pw7v5s

还可以通过argo list来查看状态，如下：

#argolist-nargoNAMESTATUSAGEDURATIONPRIORITYhello-world-9pw7vSucceeded1m10s0

使用argo logs来查看具体的日志，如下：

#argologs-nargohello-world-9pw7vhello-world-9pw7v:_____________hello-world-9pw7v:<helloworld>hello-world-9pw7v:-------------hello-world-9pw7v:\hello-world-9pw7v:\hello-world-9pw7v:\hello-world-9pw7v:##.hello-world-9pw7v:######==hello-world-9pw7v:########===hello-world-9pw7v:/""""""""""""""""___/===hello-world-9pw7v:~~~{~~~~~~~~~~~~~~~~/===-~~~hello-world-9pw7v:\______o__/hello-world-9pw7v:\\__/hello-world-9pw7v:\____\______/

核心概念

Workflow

Workflow是Argo中最重要的资源，其主要有两个重要功能：

它定义要执行的工作流
它存储工作流程的状态

要执行的工作流定义在Workflow.spec字段中，其主要包括templates和entrypoint，如下：

apiVersion:argoproj.io/v1alpha1kind:Workflowmetadata:generateName:hello-world-#Workflow的配置名称spec:entrypoint:whalesay#解析whalesaytemplatestemplates:-name:whalesay#定义whalesaytemplates，和entrypoint保持一致container:#定义一个容器，输出"helloworld"image:docker/whalesaycommand:[cowsay]args:["helloworld"]

Templates

templates是列表结构，主要分为两类：

定义具体的工作流
调用其他模板提供并行控制

定义具体的工作流

定义具体的工作流有4种类别，如下：

Container
Script
Resource
Suspend

Container

container是最常用的模板类型，它将调度一个container，其模板规范和K8S的容器规范相同，如下：

-name:whalesaycontainer:image:docker/whalesaycommand:[cowsay]args:["helloworld"]

Script

Script是Container的另一种包装实现，其定义方式和Container相同，只是增加了source字段用于自定义脚本，如下：

-name:gen-random-intscript:image:python:alpine3.6command:[python]source:|importrandomi=random.randint(1,100)print(i)

脚本的输出结果会根据调用方式自动导出到{{tasks..outputs.result}}或{{steps..outputs.result}}中。

Resource

Resource主要用于直接在K8S集群上执行集群资源操作，可以 get, create, apply, delete, replace, patch集群资源。如下在集群中创建一个ConfigMap类型资源：

-name:k8s-owner-referenceresource:action:createmanifest:|apiVersion:v1kind:ConfigMapmetadata:generateName:owned-eg-data:some:value

Suspend

Suspend主要用于暂停，可以暂停一段时间，也可以手动恢复，命令使用argo resume进行恢复。定义格式如下：

-name:delaysuspend:duration:"20s"

调用其他模板提供并行控制

调用其他模板也有两种类别：

Steps
Dag

Steps

Steps主要是通过定义一系列步骤来定义任务，其结构是"list of lists"，外部列表将顺序执行，内部列表将并行执行。如下：

-name:hello-hello-hellosteps:--name:step1template:prepare-data--name:step2atemplate:run-data-first-half-name:step2btemplate:run-data-second-half

其中step1和step2a是顺序执行，而step2a和step2b是并行执行。

还可以通过When来进行条件判断。如下：

apiVersion:argoproj.io/v1alpha1kind:Workflowmetadata:generateName:coinflip-spec:entrypoint:coinfliptemplates:-name:coinflipsteps:--name:flip-cointemplate:flip-coin--name:headstemplate:headswhen:"{{steps.flip-coin.outputs.result}}==heads"-name:tailstemplate:tailswhen:"{{steps.flip-coin.outputs.result}}==tails"-name:flip-coinscript:image:python:alpine3.6command:[python]source:|importrandomresult="heads"ifrandom.randint(0,1)==0else"tails"print(result)-name:headscontainer:image:alpine:3.6command:[sh,-c]args:["echo\"itwasheads\""]-name:tailscontainer:image:alpine:3.6command:[sh,-c]args:["echo\"itwastails\""]

提交这个Workflow，执行效果如下：

除了使用When进行条件判断，还可以进行循环操作，示例代码如下：

apiVersion:argoproj.io/v1alpha1kind:Workflowmetadata:generateName:loops-spec:entrypoint:loop-exampletemplates:-name:loop-examplesteps:--name:print-messagetemplate:whalesayarguments:parameters:-name:messagevalue:"{{item}}"withItems:-helloworld-goodbyeworld-name:whalesayinputs:parameters:-name:messagecontainer:image:docker/whalesay:latestcommand:[cowsay]args:["{{inputs.parameters.message}}"]

提交Workflow，输出结果如下：

Dag

Dag主要用于定义任务的依赖关系，可以设置开始特定任务之前必须完成其他任务，没有任何依赖关系的任务将立即执行。如下：

-name:diamonddag:tasks:-name:Atemplate:echo-name:Bdependencies:[A]template:echo-name:Cdependencies:[A]template:echo-name:Ddependencies:[B,C]template:echo

其中A会立即执行，B和C会依赖A，D依赖B和C。

然后运行一个示例看看效果，示例如下：

apiVersion:argoproj.io/v1alpha1kind:Workflowmetadata:generateName:dag-diamond-spec:entrypoint:diamondtemplates:-name:diamonddag:tasks:-name:Atemplate:echoarguments:parameters:[{name:message,value:A}]-name:Bdependencies:[A]template:echoarguments:parameters:[{name:message,value:B}]-name:Cdependencies:[A]template:echoarguments:parameters:[{name:message,value:C}]-name:Ddependencies:[B,C]template:echoarguments:parameters:[{name:message,value:D}]-name:echoinputs:parameters:-name:messagecontainer:image:alpine:3.7command:[echo,"{{inputs.parameters.message}}"]

提交workflow。

argosubmit-nargodag.yam--watch

image.png

Variables

在argo的Workflow中允许使用变量的，如下：

apiVersion:argoproj.io/v1alpha1kind:Workflowmetadata:generateName:hello-world-parameters-spec:entrypoint:whalesayarguments:parameters:-name:messagevalue:helloworldtemplates:-name:whalesayinputs:parameters:-name:messagecontainer:image:docker/whalesaycommand:[cowsay]args:["{{inputs.parameters.message}}"]

首先在spec字段定义arguments，定义变量message，其值是hello world，然后在templates字段中需要先定义一个inputs字段，用于templates的输入参数，然后在使用"{{}}"形式引用变量。

变量还可以进行一些函数运算，主要有：

filter：过滤
asInt：转换为Int
asFloat：转换为Float
string：转换为String
toJson：转换为Json

例子：

filter([1,2],{#>1})asInt(inputs.parameters["my-int-param"])asFloat(inputs.parameters["my-float-param"])string(1)toJson([1,2])

更多语法可以访问https://github.com/antonmedv/expr/blob/master/docs/Language-Definition.md进行学习。

制品库

在安装argo的时候，已经安装了mino作为制品库，那么到底该如何使用呢?

先看一个官方的例子，如下：

apiVersion:argoproj.io/v1alpha1kind:Workflowmetadata:generateName:artifact-passing-spec:entrypoint:artifact-exampletemplates:-name:artifact-examplesteps:--name:generate-artifacttemplate:whalesay--name:consume-artifacttemplate:print-messagearguments:artifacts:-name:messagefrom:"{{steps.generate-artifact.outputs.artifacts.hello-art}}"-name:whalesaycontainer:image:docker/whalesay:latestcommand:[sh,-c]args:["sleep1;cowsayhelloworld|tee/tmp/hello_world.txt"]outputs:artifacts:-name:hello-artpath:/tmp/hello_world.txt-name:print-messageinputs:artifacts:-name:messagepath:/tmp/messagecontainer:image:alpine:latestcommand:[sh,-c]args:["cat/tmp/message"]

其分为两步：

首先生成制品
然后获取制品

提交Workflow，运行结果如下：

然后在minio中可以看到生成的制品，制品经过了压缩，如下：

WorkflowTemplate

WorkflowTemplate是Workflow的模板，可以从WorkflowTemplate内部或者集群上其他Workflow和WorkflowTemplate引用它们。

WorkflowTemplate和template的区别：

template只是Workflow中templates下的一个任务，当我们定义一个Workflow时，至少需要定义一个template
WorkflowTemplate是驻留在集群中的Workflow的定义，它是Workflow的定义，因为它包含模板，可以从WorkflowTemplate内部或者集群上其他Workflow和WorkflowTemplate引用它们。

在2.7版本后，WorkflowTemplate的定义和Workflow的定义一样，我们可以简单的将kind:Workflow改成kind:WorkflowTemplate。比如：

apiVersion:argoproj.io/v1alpha1kind:WorkflowTemplatemetadata:name:workflow-template-1spec:entrypoint:whalesay-templatearguments:parameters:-name:messagevalue:helloworldtemplates:-name:whalesay-templateinputs:parameters:-name:messagecontainer:image:docker/whalesaycommand:[cowsay]args:["{{inputs.parameters.message}}"]

创建WorkflowTemplate，如下

argotemplatecreateworkflowtemplate.yaml

然后在Workflow中引用，如下：

apiVersion:argoproj.io/v1alpha1kind:Workflowmetadata:generateName:workflow-template-hello-world-spec:entrypoint:whalesaytemplates:-name:whalesaysteps:#引用模板必须在steps/dag/template下--name:call-whalesay-templatetemplateRef:#应用模板字段name:workflow-template-1#WorkflowTemplate名template:whalesay-template#具体的template名arguments:#参数parameters:-name:messagevalue:"helloworld"

ClusterWorkflowTemplate

ClusterWorkflowTemplate创建的是一个集群范围内的WorkflowTemplate，其他workflow可以引用它。

如下定义一个ClusterWorkflow。

apiVersion:argoproj.io/v1alpha1kind:ClusterWorkflowTemplatemetadata:name:cluster-workflow-template-whalesay-templatespec:templates:-name:whalesay-templateinputs:parameters:-name:messagecontainer:image:docker/whalesaycommand:[cowsay]args:["{{inputs.parameters.message}}"]

然后在workflow中使用templateRef去引用它，如下：

apiVersion:argoproj.io/v1alpha1kind:Workflowmetadata:generateName:workflow-template-hello-world-spec:entrypoint:whalesaytemplates:-name:whalesaysteps:--name:call-whalesay-templatetemplateRef:#引用模板name:cluster-workflow-template-whalesay-template#ClusterWorkflow名template:whalesay-template#具体的模板名clusterScope:true#表示是ClusterWorkflowarguments:#参数parameters:-name:messagevalue:"helloworld"

实践

上面大概叙述了一下argo的基本理论知识，更多的理论知识可以到官网去学习。

下面将使用一个简单的CI/CD实践，来了解一下用argo workflow应该如何做。

CI/CD的整个流程很简单，即：拉代码->编译->构建镜像->上传镜像->部署。

定义一个WorkflowTemplate，如下：

apiVersion:argoproj.io/v1alpha1kind:WorkflowTemplatemetadata:annotations:workflows.argoproj.io/description:|CheckoutoutfromGit,buildanddeployapplication.workflows.argoproj.io/maintainer:'@joker'workflows.argoproj.io/tags:java,gitworkflows.argoproj.io/version:'>=2.9.0'name:devops-javaspec:entrypoint:mainarguments:parameters:-name:repovalue:gitlab-test.coolops.cn/root/springboot-helloworld.git-name:branchvalue:master-name:imagevalue:registry.cn-hangzhou.aliyuncs.com/rookieops/myapp:202103101613-name:cache-imagevalue:registry.cn-hangzhou.aliyuncs.com/rookieops/myapp-name:dockerfilevalue:Dockerfile-name:devops-cd-repovalue:gitlab-test.coolops.cn/root/devops-cd.git-name:gitlabUsernamevalue:devops-name:gitlabPasswordvalue:devops123456templates:-name:mainsteps:--name:Checkouttemplate:Checkout--name:Buildtemplate:Build--name:BuildImagetemplate:BuildImage--name:Deploytemplate:Deploy#拉取代码-name:Checkoutscript:image:registry.cn-hangzhou.aliyuncs.com/rookieops/maven:3.5.0-alpineworkingDir:/workcommand:-shsource:|gitclone--branch{{workflow.parameters.branch}}http://{{workflow.parameters.gitlabUsername}}:{{workflow.parameters.gitlabPassword}}@{{workflow.parameters.repo}}.volumeMounts:-mountPath:/workname:work#编译打包-name:Buildscript:image:registry.cn-hangzhou.aliyuncs.com/rookieops/maven:3.5.0-alpineworkingDir:/workcommand:-shsource:mvn-Bcleanpackage-Dmaven.test.skip=true-Dautoconfig.skipvolumeMounts:-mountPath:/workname:work#构建镜像-name:BuildImagevolumes:-name:docker-configsecret:secretName:docker-configcontainer:image:registry.cn-hangzhou.aliyuncs.com/rookieops/kaniko-executor:v1.5.0workingDir:/workargs:---context=.---dockerfile={{workflow.parameters.dockerfile}}---destination={{workflow.parameters.image}}---skip-tls-verify---reproducible---cache=true---cache-repo={{workflow.parameters.cache-image}}volumeMounts:-mountPath:/workname:work-name:docker-configmountPath:/kaniko/.docker/#部署-name:Deployscript:image:registry.cn-hangzhou.aliyuncs.com/rookieops/kustomize:v3.8.1workingDir:/workcommand:-shsource:|gitremoteset-urloriginhttp://{{workflow.parameters.gitlabUsername}}:{{workflow.parameters.gitlabPassword}}@{{workflow.parameters.devops-cd-repo}}gitconfig--globaluser.name"Administrator"gitconfig--globaluser.email"coolops@163.com"gitclonehttp://{{workflow.parameters.gitlabUsername}}:{{workflow.parameters.gitlabPassword}}@{{workflow.parameters.devops-cd-repo}}/work/devops-cdcd/work/devops-cdgitpullcd/work/devops-cd/devops-simple-javakustomizeeditsetimage{{workflow.parameters.image}}gitcommit-am'imageupdate'gitpushoriginmastervolumeMounts:-mountPath:/workname:workvolumeClaimTemplates:-name:workmetadata:name:workspec:storageClassName:nfs-client-storageclassaccessModes:["ReadWriteOnce"]resources:requests:storage:1Gi

说明：

1、使用kaniko来创建镜像，不用挂载docker.sock，但是push镜像的时候需要config.json，所以首先需要创建一个secret，如下：

kubectlcreatesecretgenericdocker-config--from-file=.docker/config.json-nargo

2、准备好storageClass，当然也可以不需要，直接使用empty，不过可以将缓存文件这些持久化，可以加速构建(我上面没有做)。

3、创建WorkflowTemplate，命令如下：

argotemplatecreate-nargodevops-java.yaml