Trigger a workflow with Simple Message Queue (formerly MNS) - Container Service for Kubernetes

You can integrate a workflow cluster with Alibaba Cloud Simple Message Queue (formerly MNS). By using Simple Message Queue (formerly MNS) as a message broker to connect to various event sources, you can trigger workflows in an event-driven manner. When a new event occurs, such as an event from Object Storage Service (OSS) or EventBridge, Simple Message Queue (formerly MNS) receives a message. Argo then automatically runs the workflow if the message meets the defined event trigger conditions.

Prerequisites

The following services and features must be activated:
- Enable the event-driven feature
- Activate Simple Message Queue (formerly MNS) and grant permissions. This feature incurs charges for Simple Message Queue (formerly MNS). For more information, see Billing.
Create a workflow cluster and download the Alibaba Cloud Argo CLI.

Step 1: Create an event bus

An event bus can be shared by event-driven workflows in a namespace. You can create an event bus using NATS or Simple Message Queue (formerly MNS). If you have already created an event bus, skip to Step 2: Create an event source.

NATS

Create an event-bus.yaml file. Sample configuration:

apiVersion: argoproj.io/v1alpha1
kind: EventBus
metadata:
  name: default
spec:
  nats:
    native:
      replicas: 3
      auth: token

Run the following command to create the EventBus:
```
kubectl apply -f event-bus.yaml
```
Note
After the command succeeds, an EventBus pod is created in the default namespace. Subsequent operations must be performed in the same namespace.
Run the following command to verify that the EventBus pod has started:
```
kubectl get pod
```

MNS

Log on to the SMQ console.
On the Topics page, create a topic named argoeventbus. Then, on the Topic Details page, obtain the endpoint from the Access Point section.
Log on to the RAM console as a RAM user who has administrative rights.
Create a RAM user, grant the AliyunMNSFullAccess permission to the RAM user, and obtain the AccessKey ID and AccessKey secret for the RAM user.
For more information, see Create a RAM user, Manage RAM user permissions, Create an AccessKey, and View the AccessKey information of a RAM user.

Run the following command to create a Secret to store the AccessKey ID and AccessKey secret.

kubectl create secret generic mns-secret\
  --from-literal=accesskey=*** \
  --from-literal=secretkey=***

Create an event-bus-mns.yaml file and replace the placeholder values in the example with your own.

topic: Replace this value with the name of the Simple Message Queue (formerly MNS) topic that you created in step 2.
endpoint: Replace this value with the endpoint that you obtained in step 2.

apiVersion: argoproj.io/v1alpha1
kind: EventBus
metadata:
  name: default
spec:
  alimns:
    accessKey:
      key: accesskey
      name: mns-secret
    secretKey:
      key: secretkey
      name: mns-secret
    topic: argoeventbus  # The name of the topic in Simple Message Queue (formerly MNS).
    endpoint: http://165***368.mns.<region>.aliyuncs.com  # The endpoint for Simple Message Queue (formerly MNS).

Run the following command to apply the event-bus-mns.yaml file and create the EventBus resource.
```
kubectl apply -f event-bus-mns.yaml
```

Step 2: Create an event source

Log on to the SMQ console.
On the Queues page, create a queue named test-event-queue. Then, on the Queue Details page, obtain the endpoint from the Access Point section.
Note
If you created the event bus using Simple Message Queue (formerly MNS), you can skip steps 3 to 5 and proceed to step 6.
Log on to the RAM console as a RAM user who has administrative rights.
Create a RAM user, grant the AliyunMNSFullAccess permission to the RAM user, and obtain the AccessKey ID and AccessKey secret for the RAM user.
For more information, see Create a RAM user, Manage RAM user permissions, Create an AccessKey, and View the AccessKey information of a RAM user.

Run the following command to create a Secret to store the AccessKey ID and AccessKey secret.

kubectl create secret generic mns-secret\
  --from-literal=accesskey=*** \
  --from-literal=secretkey=***

Create an event-source.yaml file and replace the placeholder values in the example with your own.

queue: Replace this value with the name of the Simple Message Queue (formerly MNS) queue that you created in step 2.
endpoint: Replace this value with the endpoint that you obtained in step 2.

apiVersion: argoproj.io/v1alpha1
kind: EventSource
metadata:
  name: ali-mns
spec:
  mns:
    example:
      jsonBody: true
      accessKey:
        key: accesskey
        name: mns-secret
      secretKey:
        key: secretkey
        name: mns-secret
      queue: test-event-queue # The name of the queue in Simple Message Queue (formerly MNS).
      waitTimeSeconds: 20
      endpoint: http://165***368.mns.<region>.aliyuncs.com # The endpoint for Simple Message Queue (formerly MNS).

Run the following command to apply the event-source.yaml file and create the EventSource resource.
```
kubectl apply -f event-source.yaml
```
Run the following command to verify that the event source Pod started successfully.
```
kubectl get pod
```

Step 3: Create an event sensor

Create an event-sensor.yaml file and embed the workflow definition within the event sensor. The following is a sample configuration for the event sensor:

Sample code

apiVersion: argoproj.io/v1alpha1
kind: Sensor
metadata:
  name: ali-mns
spec:
  template:
    serviceAccountName: default
  dependencies:
    - name: test-dep
      eventSourceName: ali-mns    # Matches the event source name.
      eventName: example          # Matches the event name defined in the event source.
  triggers:
    - template:
        name: mns-workflow
        k8s:
          operation: create
          source:
            resource:
              apiVersion: argoproj.io/v1alpha1    # Embeds the workflow definition.
              kind: Workflow
              metadata:
                generateName: ali-mns-workflow-
              spec:
                entrypoint: whalesay
                arguments:
                  parameters:        # Passes the event content.
                    - name: message
                      # this is the value that should be overridden
                      value: hello world
                templates:
                  - name: whalesay
                    inputs:
                      parameters:
                        - name: message
                    container:
                      image: docker/whalesay:latest
                      command: [cowsay]
                      args: ["{{inputs.parameters.message}}"]
          parameters:
            - src:        # Parses the event content and passes it to the workflow.
                dependencyName: test-dep
                dataKey: body
              dest: spec.arguments.parameters.0.value

Apply the event-sensor.yaml file to create the Sensor resource.
```
kubectl apply -f event-sensor.yaml
```
Run the following command to verify that the event sensor Pod started successfully.
```
kubectl get pod
```

Note

When you create an EventBus by using Simple Message Queue (formerly MNS), a corresponding Simple Message Queue (formerly MNS) queue is automatically created after the event sensor is created. The naming format for the queue is: ackone-argowf-<namespace>-<sensor-name>-<sensor-uid>.

Step 4: Test the workflow trigger

Log on to the SMQ console.
On the Queues page, find the queue named test-event-queue and click Send Messages in the Actions column.
On the Send Message page, enter the message content test trigger argo workflow, and then click Send Message.

Check the workflow's status in the cluster.

argo list

The expected output is as follows:

NAME                     STATUS    AGE   DURATION   PRIORITY
ali-mns-workflow-5prz7   Running   6s    6s         0

View the workflow logs to inspect the message content.

argo logs ali-mns-workflow-5prz7

Important

The workflow name in the command must be the same as the name returned in the previous step. ali-mns-workflow-5prz7 is an example. Replace it with the name from your environment.
The message content is Base64-encoded.

The expected output is as follows:

ali-mns-workflow-5prz7-whalesay-2429203954: time="2023-12-14T08:33:37.964Z" level=info msg="capturing logs" argo=true
ali-mns-workflow-5prz7-whalesay-2429203954:
ali-mns-workflow-5prz7-whalesay-2429203954:  < dGVzdCB0cmlnZ2VyIGFyZ28gd29ya2Zsb3c= >
ali-mns-workflow-5prz7-whalesay-2429203954:  -----------------------------------------
ali-mns-workflow-5prz7-whalesay-2429203954:     \
ali-mns-workflow-5prz7-whalesay-2429203954:      \
ali-mns-workflow-5prz7-whalesay-2429203954:       \
ali-mns-workflow-5prz7-whalesay-2429203954:                     ##        .
ali-mns-workflow-5prz7-whalesay-2429203954:               ## ## ##       ==
ali-mns-workflow-5prz7-whalesay-2429203954:            ## ## ## ##      ===
ali-mns-workflow-5prz7-whalesay-2429203954:        /""""""""""""""""___/ ===
ali-mns-workflow-5prz7-whalesay-2429203954:   ~~~ {~~ ~~~~ ~~~ ~~~~ ~~ ~ /  ===- ~~~
ali-mns-workflow-5prz7-whalesay-2429203954:        \______ o          __/
ali-mns-workflow-5prz7-whalesay-2429203954:         \    \        __/
ali-mns-workflow-5prz7-whalesay-2429203954:           \____\______/
ali-mns-workflow-5prz7-whalesay-2429203954: time="2023-12-14T08:33:38.979Z" level=info msg="sub-process exited" argo=true error="<nil>"

Step 5: Clean up resources

Clean up the resources.

Delete the event sensor.
```
kubectl delete sensor ali-mns
```
Delete the event source.
```
kubectl delete eventsource ali-mns
```
Delete the event bus.
```
kubectl delete eventbus default
```

Check the Pod status to confirm that all related Pods have been terminated.
```
kubectl get pod
```