通过MNS消息触发工作流

前提条件

• 已开通以下服务与功能

  • 开启事件驱动功能

  • 开通消息服务MNS并授权，该功能会涉及MNS相关计费，具体收费情况，请参见价格说明。

• 已创建工作流集群并下载阿里云Argo CLI。

  • 创建工作流集群

  • 获取集群KubeConfig并通过kubectl工具连接集群

  • 安装阿里云Argo CLI

步骤一：创建Event Bus

Event Bus可以被命名空间中的事件驱动工作流共享，您可以通过NATS和MNS消息队列两种方式创建，如果已经创建，请直接执行步骤二：创建Event Source。

步骤二：创建Event Source

1. 登录消息服务MNS控制台。

2. 在队列列表页面创建队列test-event-queue，并在队列详情页面的接入点区域获取Endpoint。

   说明

   若您是通过消息服务MNS创建的Event Bus，则步骤3~步骤5可直接跳过，直接执行步骤6。

3. 使用RAM管理员登录RAM控制台。

4. 创建RAM用户，为RAM用户授予AliyunMNSFullAccess权限，并获取RAM用户的AK和SK。

   具体操作，请参见创建RAM用户、为RAM用户授权、创建AccessKey和查看RAM用户的AccessKey信息。

5. 执行以下命令，创建Secret用于存储AK和SK。

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

6. 创建event-source.yaml文件，将示例中的参数修改为实际使用的参数值。

   • topic：需替换为步骤2中创建的MNS中的主题名称。

   • endpoint：需替换为步骤2中获取的Endpoint。

   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 # 对应MNS中的队列名称。
         waitTimeSeconds: 20
         endpoint: http://165***368.mns.<region>.aliyuncs.com # 对应MNS的Endpoint。

7. 执行以下命令，应用event-source.yaml文件创建Event Source。

   kubectl apply -f event-source.yaml

8. 执行以下命令，查看Event Source Pod是否正常启动。

   kubectl get pod

步骤三：创建Event Sensor

1. 创建event-sensor.yaml文件，在Event Sensor中嵌入待执行的工作流定义。Event Sensor示例代码如下所示：

   展开查看示例代码

   apiVersion: argoproj.io/v1alpha1
   kind: Sensor
   metadata:
     name: ali-mns
   spec:
     template:
       serviceAccountName: default
     dependencies:
       - name: test-dep
         eventSourceName: ali-mns    # 匹配eventsource名称。
         eventName: example          # 匹配eventsource中的事件名称定义。
     triggers:
       - template:
           name: mns-workflow
           k8s:
             operation: create
             source:
               resource:
                 apiVersion: argoproj.io/v1alpha1    # 嵌入工作流定义。
                 kind: Workflow
                 metadata:
                   generateName: ali-mns-workflow-
                 spec:
                   entrypoint: whalesay
                   arguments:
                     parameters:        # 参数传递事件内容。
                       - 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:        # 解析事件内容，传递给工作流。
                   dependencyName: test-dep
                   dataKey: body
                 dest: spec.arguments.parameters.0.value

2. 执行以下命令，应用event-sensor.yaml文件创建Event Sensor。

   kubectl apply -f event-sensor.yaml

3. 执行以下命令，查看Event Sensor Pod是否正常启动。

   kubectl get pod

步骤四：验证通过向MNS发送消息触发工作流

1. 登录消息服务MNS控制台。

2. 在队列列表页面中，找到队列test-event-queue，在其操作列选择更多 > 收发消息。

3. 在收发消息快速体验页面中，输入消息内容test trigger argo workflow，然后单击发送消息。

4. 执行以下命令，在工作流集群中查看工作流的运行情况。

   argo list

   预期输出如下：

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

5. 获取工作流日志，查看消息内容。

   argo logs ali-mns-workflow-5prz7

   重要

   • 该命令中的工作流名称必须和上一步骤中返回的工作流名称一致，ali-mns-workflow-5prz7仅为示例值，请您修改为实际环境中的返回值。

   • 消息内容使用Base64编码。

   预期输出如下：

步骤五：清除Event相关资源

1. 执行以下命令清除相关资源。

   • 清除Event Sensor

     kubectl delete sensor ali-mns

   • 清除Event Source

     kubectl delete eventsource ali-mns

   • 清除Event Bus

     kubectl delete eventbus default

2. 执行以下命令查看Pod状态，确认所有资源已清除。

   kubectl get pod