上传文件至OSS以触发工作流运行-容器服务 Kubernetes 版 ACK(ACK)-阿里云帮助中心

本文介绍如何集成阿里云对象存储OSS与阿里云轻量消息队列（原 MNS），通过将文件上传至对象存储OSS中，自动触发工作流处理文件，并生成结果。

前提条件

步骤一：创建Event Bus

Event Bus可以被命名空间中的事件驱动工作流共享。如果已经创建，可跳过该步骤，直接执行步骤二：创建Event Source。

说明

使用轻量消息队列（原 MNS）方式创建Event Bus时，不会创建Pod。
如需使用Trigger功能，请使用NATS方式创建EventBus。目前轻量消息队列（原 MNS）方式不支持开源Argo Event的Sensor Trigger。

方式一：使用NATS

创建event-bus.yaml文件。Event Bus示例代码如下所示：

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

执行以下命令，创建EventBus。
```
kubectl apply -f event-bus.yaml
```
说明
命令执行成功后，会在default命名空间下创建Event Bus Pod。后续操作需在同一命名空间下。
执行以下命令，查看Event Bus Pod是否正常启动。
```
kubectl get pod
```

方式二：使用轻量消息队列（原 MNS）

登录轻量消息队列（原 MNS）控制台。
在主题列表页面创建主题argoeventbus，并在主题详情页面的接入点区域获取Endpoint。
使用RAM管理员登录RAM控制台。
创建RAM用户，授权AliyunMNSFullAccess权限，并获取RAM用户的AK和SK。

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

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

创建event-bus-mns.yaml文件，EventBus示例代码如下所示：

topic：需替换为2中创建的轻量消息队列（原 MNS）中的主题名称。
endpoint：需替换为2中获取的Endpoint。

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  # 对应轻量消息队列（原 MNS）中的主题名称。
    endpoint: http://165***368.mns.<region>.aliyuncs.com

执行以下命令，创建event-bus-mns.yaml。
```
kubectl apply -f event-bus-mns.yaml
```

步骤二：创建Event Source

使用RAM管理员登录RAM控制台。
创建RAM用户，为其授予AliyunMNSFullAccess权限，并获取RAM用户的AK和SK。具体操作，请参见创建RAM用户、管理RAM用户的权限、创建AccessKey和查看RAM用户的AccessKey信息。

创建event-source.yaml文件，Event Source示例代码如下所示：

queue：需替换为轻量消息队列（原 MNS）的名称。
endpoint：需替换为轻量消息队列（原 MNS）的接入点。

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: oss-event-queue # 轻量消息队列（原 MNS）名称。
      waitTimeSeconds: 20
      endpoint: http://165***368.mns.<region>.aliyuncs.com # 轻量消息队列（原 MNS）接入点。

执行以下命令，创建Event Source。
```
kubectl apply -f event-source.yaml
```
执行以下命令，查看Event Source Pod是否正常启动。
```
kubectl get pod
```

步骤三：创建Event Sensor

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

展开查看示例代码

apiVersion: argoproj.io/v1alpha1
kind: Sensor
metadata:
  name: process-oss-file
spec:
  template:
    serviceAccountName: default
  dependencies:
    - name: dep1
      eventSourceName: ali-mns
      eventName: example
  triggers:
    - template:
        name: process-oss-file-workflow
        k8s:
          operation: create
          source:
            resource:
              apiVersion: argoproj.io/v1alpha1
              kind: Workflow
              metadata:
                generateName: process-oss-file-
                namespace: default
              spec:
                entrypoint: process-oss-file
                volumes:
                - name: workdir
                  persistentVolumeClaim:
                    claimName: pvc-oss
                arguments:
                  parameters:
                  - name: message
                    # this is the value that should be overridden
                    value: event message
                templates:
                - name: process-oss-file
                  steps:
                  - - name: parse-event-body
                      template: parse-event-body
                  - - name: process-file
                      template: process-file
                      arguments:
                        parameters:
                        - name: file-name
                          value: "{{steps.parse-event-body.outputs.parameters.file-name}}"
                - name: parse-event-body
                  container:
                    image: acr-multiple-clusters-registry.cn-hangzhou.cr.aliyuncs.com/ack-multiple-clusters/jq-alpine
                    command: [sh, -c]
                    args:
                    - echo "Event body:";
                      echo {{workflow.parameters.message}} | base64 -d;
                      TriggerFileName=$(echo {{workflow.parameters.message}} | base64 -d | jq .events[0].oss.object.key | cut -c2- | rev | cut -c2- |rev);
                      echo "" && echo "TriggerFileName from event is $TriggerFileName";
                      Tmp=${TriggerFileName%%.complete} && DataFileName=${Tmp##*/};
                      echo "DataFileName after cutting .complete is $DataFileName, and pass file name to next step";
                      echo $DataFileName > /tmp/file-name.txt
                  outputs:
                    parameters:
                    - name: file-name
                      valueFrom:
                        path: /tmp/file-name.txt
                - name: process-file
                  inputs:
                    parameters:
                      - name: file-name
                  container:
                    image: acr-multiple-clusters-registry.cn-hangzhou.cr.aliyuncs.com/ack-multiple-clusters/jq-alpine
                    imagePullPolicy: Always
                    command: [sh, -c]
                    args:
                    - echo "Show data-file:" && echo "";
                      ls -l /mnt/vol/{{inputs.parameters.file-name}};
                      echo "Content of data file:" && echo "";
                      cat /mnt/vol/{{inputs.parameters.file-name}} ;
                      echo "" && echo "Finished" ;
                    volumeMounts:
                    - name: workdir
                      mountPath: /mnt/vol
          parameters:
            - src:
                dependencyName: dep1
                dataKey: body
              dest: spec.arguments.parameters.0.value

执行以下命令，创建Event Sensor。
```
kubectl apply -f event-sensor.yaml
```
执行以下命令，查看Event Sensor Pod是否正常启动。
```
kubectl get pod
```

说明

使用轻量消息队列（原 MNS）方式创建Eventbus时，在Event Sensor创建完成后，会自动创建一个轻量消息队列（原 MNS）与之对应，队列命名格式为：ackone-argowf-<namespace>-<sensor-name>-<sensor-uid>。

步骤四：验证向OSS上传文件触发工作流

登录OSS管理控制台。
向OSS Bucket中上传以下2个文件（该文件需自备），触发工作流运行。
- datafile：数据文件，文本格式，内容自定义。
- datafile.complete：trigger文件，可以是空文件。
执行以下命令，在工作流集群中查看工作流运行情况。
```
argo list
```
预期输出如下：
```
NAME STATUS AGE DURATION PRIORITY
process-oss-file-kmb4k Running 13s 13s 0
```

执行以下命令，获取工作流日志。

argo logs process-oss-file-kmb4k

重要

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

预期输出如下：

.kube % argo logs process-oss-file-vtdfb
process-oss-file-vtdfb-parse-event-body-  : time="2023-12-14T08:09:25.816Z" level=info msg="capturing logs" argo=true
process-oss-file-vtdfb-parse-event-body-  : Event body:
process-oss-file-vtdfb-parse-event-body-  : {"events": [{
process-oss-file-vtdfb-parse-event-body-  :         "eventName": "ObjectCreated:PostObject",
process-oss-file-vtdfb-parse-event-body-  :         "eventSource": "acs:oss",
process-oss-file-vtdfb-parse-event-body-  :         "eventTime": "2023-12-14T08:00:21.000Z",
process-oss-file-vtdfb-parse-event-body-  :         "eventVersion": "1.0",
process-oss-file-vtdfb-parse-event-body-  :         "oss": {
process-oss-file-vtdfb-parse-event-body-  :             "bucket": {
process-oss-file-vtdfb-parse-event-body-  :                 "arn": "acs:oss:cn-hangzhou:xxx:bucket-workflow-artifact xxx",
process-oss-file-vtdfb-parse-event-body-  :                 "name": "bucket-workflow-artifactxxx",
process-oss-file-vtdfb-parse-event-body-  :                 "ownerIdentity": "xxx",
process-oss-file-vtdfb-parse-event-body-  :                 "virtualBucket": ""},
process-oss-file-vtdfb-parse-event-body-  :             "object": {
process-oss-file-vtdfb-parse-event-body-  :                 "deltaSize": 0,
process-oss-file-vtdfb-parse-event-body-  :                 "eTag": "DA6xxx",
process-oss-file-vtdfb-parse-event-body-  :                 "key": "data.complete",
process-oss-file-vtdfb-parse-event-body-  :                 "objectMeta": {"mimeType": "application/octet-stream"},
process-oss-file-vtdfb-parse-event-body-  :                 "size": 14},
process-oss-file-vtdfb-parse-event-body-  :             "ossSchemaVersion": "1.0",
process-oss-file-vtdfb-parse-event-body-  :             "ruleId": "process-data"},
process-oss-file-vtdfb-parse-event-body-  :         "region": "cn-hangzhou",
process-oss-file-vtdfb-parse-event-body-  :         "requestParameters": {"sourceIPAddress": "106.11.xxx"},
process-oss-file-vtdfb-parse-event-body-  :         "responseElements": {"requestId": "657AB6155Cxxx"},
process-oss-file-vtdfb-parse-event-body-  :         "userIdentity": {"principalId": "165304082xxx"}}]}
process-oss-file-vtdfb-parse-event-body-  : TriggerFileName from event is data.complete
process-oss-file-vtdfb-parse-event-body-  : DataFileName after cutting .complete is data, and pass file name to next step
process-oss-file-vtdfb-parse-event-body-  : time="2023-12-14T08:09:26.821Z" level=info msg="sub-process exited" argo=true error="<nil>"
process-oss-file-vtdfb-parse-event-body-  : time="2023-12-14T08:09:26.821Z" level=info msg="/tmp/file-name.txt -> /var/run/argo/outputs/parameters//tmp/file-name.txt"
  argo=true
process-oss-file-vtdfb-process-file       : time="2023-12-14T08:09:46.262Z" level=info msg="capturing logs" argo=true
process-oss-file-vtdfb-process-file       : Show data-file:
process-oss-file-vtdfb-process-file       : -rw-r------    1 root     root             3 Dec 14 08:00 /mnt/vol/data
process-oss-file-vtdfb-process-file       : Content of data file:
process-oss-file-vtdfb-process-file       : pi
process-oss-file-vtdfb-process-file       :
process-oss-file-vtdfb-process-file       : Finished
process-oss-file-vtdfb-process-file       : time="2023-12-14T08:09:47.267Z" level=info msg="sub-process exited" argo=true error="<nil>"
process-oss-file-vtdfb-sparkapp-          : time="2023-12-14T08:10:42.033Z" level=info msg="capturing logs" argo=true
process-oss-file-vtdfb-sparkapp-          : WARNING: An illegal reflective access operation has occurred
process-oss-file-vtdfb-sparkapp-          : WARNING: Illegal reflective access by org.apache.spark.unsafe.Platform (file:/opt/spark/jars/spark-unsafe_2.12-3.2.1.jar) to constructor java.nio.DirectByteBuffer
process-oss-file-vtdfb-sparkapp-          : WARNING: Please consider reporting this to the maintainers of org.apache.spark.unsafe.Platform
process-oss-file-vtdfb-sparkapp-          : WARNING: Use --illegal-access=warn to enable warnings of further illegal reflective access operations
process-oss-file-vtdfb-sparkapp-          : WARNING: All illegal access operations will be denied in a future release
process-oss-file-vtdfb-sparkapp-          : 23/12/14 08:10:45 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
process-oss-file-vtdfb-sparkapp-          : Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
process-oss-file-vtdfb-sparkapp-          : 23/12/14 08:10:45 INFO SparkKubernetesClientFactory: Auto-configuring K8S client using current context from users K8S config.file
process-oss-file-vtdfb-sparkapp-          : 23/12/14 08:10:46 INFO KerberosConfDriverFeatureStep: You have not specified a krb5.conf file locally or via a ConfigMap. Make sure that you have the krb5.conf l the driver image.
process-oss-file-vtdfb-sparkapp-          : 23/12/14 08:10:48 INFO LoggingPodStatusWatcherImpl: State changed, new state:
process-oss-file-vtdfb-sparkapp-          :   pod name: sparkapp-xxx-driver
process-oss-file-vtdfb-sparkapp-          :   namespace: default
process-oss-file-vtdfb-sparkapp-          :   labels: spark-app-selector -> spark-796bd3axxx, spark-role -> driver
process-oss-file-vtdfb-sparkapp-          :   pod uid: 3d63bb8d

步骤五：清除Event相关资源

依次执行以下命令，清除Event相关资源。

kubectl delete sensor process-oss-file
kubectl delete eventsource ali-mns
kubectl delete eventbus default

执行以下命令查看Pod，确认所有资源已清除。
```
kubectl get pod
```