Workflow Step Definition

Workflow Step Definition - 图1tip

Before reading this section, make sure you have understood the concept of WorkflowStepDefinition in KubeVela and learned the basic knowledge of CUE.

In this section, we will introduce how to customize the workflow step in Application by using CUE through WorkflowStepDefinition.

Deliver a simple workflow step

We can generate WorkflowStepDefinition CUE file with command vela def <def-name> --type workflow-step > <def-name>.cue.

Let’s take a custom step for sending an HTTP request as an example, first, initialize the Definition file:

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

After initialization, we can see the following in 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. }

Inside the template is the execution logic for this workflow step. We can define parameter in template to receive the parameters passed in by the user:

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

CUE provides a series of basic builtin packages, such as: regexp, json, strings, math, etc.

At the same time, KubeVela also provides the vela/op package by default, which contains a series of built-in workflow CUE actions, such as: sending HTTP requests, operating K8s resources, printing logs, etc.

Now we can import KubeVela’s built-in vela/op package and CUE’s official encoding/json, use op.#HTTPDo to send HTTP requests according to the user’s parameters, and use json.Marshal() to marshal the data.

  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. }

If the HTTP request returns a status code greater than 400, we expect this step to be failed. Use op.#Fail to fail this step, and the definition is like:

  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. }

Use vela def apply -f request.cue to deploy this Definition to the cluster, then we can use this custom step directly in the Application.

Deploy the following Application: The first step of the workflow will send an HTTP request to get the information of the KubeVela repository; at the same time, this step will use the star number of the KubeVela repository as the Output, the next step will use this Output as a parameter, and sent it as message to the Slack:

Workflow Step Definition - 图2tip

Please refer to Inputs and Outputs for more information of data passing between steps.

  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>

Custom Health Checks

If you want to wait a period of time in a workflow step until a certain condition is met, or until the status of a resource becomes ready, you can use op.#ConditionalWait.

Take the status of waiting for a Deployment as an example, use op.#Apply to deploy a Deployment, and then use op.#ConditionalWait to wait for the status of the Deployment to become 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. }

Full available context in Workflow Step

KubeVela allows you to reference some runtime data via the context keyword.

In a workflow step definition, you can use the following context data:

Context VariableDescriptionType
context.nameThe name of the Application.string
context.appNameThe name of the Application.string
context.namespaceThe namespace of the Application.string
context.appRevisionThe revision of the Application.string
context.stepNameThe name of current step.string
context.stepSessionIDThe ID of current step.string
context.spanIDThe trace ID of current step in this reconcile.string
context.workflowNameThe workflow name specified in annotation.string
context.publishVersionThe version of application instance specified in annotation.string

Kubernetes API for WorkflowStepDefinition

KubeVela is fully programmable through CUE, while it leverages Kubernetes as a control plane and is consistent with the API in YAML.

Therefore, the CUE Definition will be translated into the Kubernetes API when applied to the cluster.

Workflow step definitions will be in the following API format:

  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>

More examples to learn

You can check the following resources for more examples: