版本:v1.6

自定义工作流

自定义工作流 - 图1提示

在阅读本部分之前,请确保你已经了解 KubeVela 中 工作流节点定义(WorkflowStepDefinition) 的概念且学习掌握了 CUE 的基本知识

本节将以工作流步骤定义的例子展开说明,介绍如何使用 CUE 通过工作流步骤定义WorkflowStepDefinition 来自定义应用部署计划的工作流步骤。

我们可以通过 vela def <def-name> --type workflow-step > <def-name>.cue 来生成一个 workflow-step 类型的 Definition CUE 文件。

我们以一个发送 HTTP 请求的自定义步骤为例,首先初始化这个 Definition 文件:

  1. vela def init request --type workflow-step --desc "Send request to the url" > request.cue

初始化完成后,我们可以在 request.cue 中看到如下内容:

  1. request: {
  2. alias: ""
  3. annotations: {}
  4. attributes: {}
  5. description: "Send request to the url"
  6. labels: {}
  7. type: "workflow-step"
  8. }
  9. template: {
  10. }

template 中是这个工作流步骤的执行逻辑。我们先在 template 中定义 parameter,用于接收用户传入的参数:

  1. template: {
  2. parameter: {
  3. url: string
  4. method: *"GET" | "POST" | "PUT" | "DELETE"
  5. body?: {...}
  6. header?: [string]: string
  7. }
  8. }

CUE 提供了一系列基础内置包,如:regexp, json, strings, math 等。

同时,KubeVela 也默认提供了 vela/op 包,其中包含了一系列内置的工作流 CUE 操作符,如:发送 HTTP 请求,操作 K8s 资源,打印日志等,来帮助你更好地编写工作流步骤。

我们在这引用 KubeVela 内置的 vela/op 包以及 CUE 官方的 encoding/json,使用 op.#HTTPDo 根据用户的参数发送 HTTP 请求,并使用 json.Marshal() 来进行数据类型的转换。

  1. import (
  2. "vela/op"
  3. "encoding/json"
  4. )
  5. request: {
  6. alias: ""
  7. annotations: {}
  8. attributes: {}
  9. description: "Send request to the url"
  10. labels: {}
  11. type: "workflow-step"
  12. }
  13. template: {
  14. http: op.#HTTPDo & {
  15. method: parameter.method
  16. url: parameter.url
  17. request: {
  18. if parameter.body != _|_ {
  19. body: json.Marshal(parameter.body)
  20. }
  21. if parameter.header != _|_ {
  22. header: parameter.header
  23. }
  24. }
  25. }
  26. parameter: {
  27. url: string
  28. method: *"GET" | "POST" | "PUT" | "DELETE"
  29. body?: {...}
  30. header?: [string]: string
  31. }
  32. }

如果 HTTP 请求返回的状态码大于 400,则我们希望这个步骤的状态是失败的。使用 op.#Fail 是步骤的状态变为失败。这样一个发送请求的自定义步骤最终为:

  1. import (
  2. "vela/op"
  3. "encoding/json"
  4. )
  5. request: {
  6. alias: ""
  7. annotations: {}
  8. attributes: {}
  9. description: "Send request to the url"
  10. labels: {}
  11. type: "workflow-step"
  12. }
  13. template: {
  14. http: op.#HTTPDo & {
  15. method: parameter.method
  16. url: parameter.url
  17. request: {
  18. if parameter.body != _|_ {
  19. body: json.Marshal(parameter.body)
  20. }
  21. if parameter.header != _|_ {
  22. header: parameter.header
  23. }
  24. }
  25. }
  26. fail: op.#Steps & {
  27. if http.response.statusCode > 400 {
  28. requestFail: op.#Fail & {
  29. message: "request of \(parameter.url) is fail: \(http.response.statusCode)"
  30. }
  31. }
  32. }
  33. response: json.Unmarshal(http.response.body)
  34. parameter: {
  35. url: string
  36. method: *"GET" | "POST" | "PUT" | "DELETE"
  37. body?: {...}
  38. header?: [string]: string
  39. }
  40. }

使用 vela def apply -f request.cue 将这个 Definition 部署到集群中,接着,我们就可以直接在应用部署计划中使用这个自定义步骤了。

部署如下应用部署计划:工作流的第一步会发送一个 HTTP 请求,得到 KubeVela 仓库的信息;同时,这个步骤会将 KubeVela 仓库的 Star 数作为 Output,下一个步骤将使用这个 Output 作为参数,并将其作为消息内容发送到 Slack 消息中:

自定义工作流 - 图2提示

有关于参数传递的更多信息,请参考 工作流参数传递

  1. apiVersion: core.oam.dev/v1beta1
  2. kind: Application
  3. metadata:
  4. name: request-http
  5. namespace: default
  6. spec:
  7. components: []
  8. workflow:
  9. steps:
  10. - name: request
  11. type: request
  12. properties:
  13. url: https://api.github.com/repos/kubevela/kubevela
  14. outputs:
  15. - name: stars
  16. valueFrom: |
  17. import "strconv"
  18. "Current star count: " + strconv.FormatInt(response["stargazers_count"], 10)
  19. - name: notification
  20. type: notification
  21. inputs:
  22. - from: stars
  23. parameterKey: slack.message.text
  24. properties:
  25. slack:
  26. url:
  27. value: <your slack url>

如果你希望在工作流步骤中等待一段时间,直到某个条件满足,或者直到某个资源的状态变为 Ready,你可以使用 op.#ConditionalWait

以等待 Deployment 的状态为例,先使用 op.#Apply 部署一个 Deployment,再使用 op.#ConditionalWait 来等待这个 Deployment 的状态变为 Ready:

  1. import (
  2. "vela/op"
  3. )
  4. "apply-deployment": {
  5. alias: ""
  6. annotations: {}
  7. attributes: {}
  8. description: ""
  9. labels: {}
  10. type: "workflow-step"
  11. }
  12. template: {
  13. output: op.#Apply & {
  14. value: {
  15. apiVersion: "apps/v1"
  16. kind: "Deployment"
  17. metadata: {
  18. name: context.stepName
  19. namespace: context.namespace
  20. }
  21. spec: {
  22. selector: matchLabels: wr: context.stepName
  23. template: {
  24. metadata: labels: wr: context.stepName
  25. spec: containers: [{
  26. name: context.stepName
  27. image: parameter.image
  28. if parameter["cmd"] != _|_ {
  29. command: parameter.cmd
  30. }
  31. }]
  32. }
  33. }
  34. }
  35. }
  36. wait: op.#ConditionalWait & {
  37. continue: output.value.status.readyReplicas == 1
  38. }
  39. parameter: {
  40. image: string
  41. cmd?: [...string]
  42. }
  43. }

KubeVela 让你可以在运行时,通过 context 关键字来引用一些信息。

在工作流步骤定义中,你可以使用如下 Context 信息:

Context VariableDescriptionType
context.name应用部署计划的名称string
context.appName应用部署计划的名称string
context.namespace应用部署计划的命名空间string
context.appRevision应用部署计划的版本string
context.stepName当前步骤的名称string
context.stepSessionID当前步骤的 IDstring
context.spanID当前步骤此次执行的 Trace IDstring
context.workflowName应用部署计划 Annotation 中声明的工作流名称string
context.publishVersion应用部署计划 Annotation 中声明的版本string

KubeVela 通过 CUE 完全可编程,同时它利用 Kubernetes 作为控制平面并与 YAML 中的 API 保持一致。

因此,CUE Definition 在应用到集群时将被转换为 Kubernetes API。

工作流步骤定义将采用以下 API 格式:

  1. apiVersion: core.oam.dev/v1beta1
  2. kind: WorkflowStepDefinition
  3. metadata:
  4. annotations:
  5. definition.oam.dev/description: <Function description>
  6. spec:
  7. schematic:
  8. cue: # Details of workflow steps defined by CUE language
  9. template: <CUE format template>

你还可以在以下例子中找到更多的工作流步骤定义:

Last updated on 2022年12月1日 by Somefive