Upload a file to OSS to trigger a workflow run - Container Service for Kubernetes

Prerequisites

You have enabled the event-driven feature.
You have created OSS event notifications.
You have created a workflow cluster.
You have installed the Alibaba Cloud Argo CLI.

Step 1: Create an EventBus

Event-driven workflows in the same namespace can share an EventBus. If you have already created an EventBus, you can skip this step and go to Step 2: Create an EventSource.

Note

When you create an EventBus using Message Service (MNS), no pod is created.
To use the trigger feature, create the EventBus using NATS. The Message Service (MNS) method does not support triggers in an open source Argo Events Sensor.

Method 1: Using 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
```

Method 2: Using Message Service (MNS)

Log on to the SMQ console.
On the Topics page, create a topic named argoeventbus. On the Topic Details page, obtain the endpoint from the Access Point section.
Log on to the RAM console as a RAM administrator.
Create a RAM user, grant the AliyunMNSFullAccess permission, and obtain the user's AccessKey ID and AccessKey secret.

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. Sample configuration:

topic: Replace the value with the name of the Message Service (MNS) topic that you created in step 2.
endpoint: Replace the 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 Message Service (MNS).
    endpoint: http://165***368.mns.<region>.aliyuncs.com

Run the following command to apply the event-bus-mns.yaml file:
```
kubectl apply -f event-bus-mns.yaml
```

Step 2: Create an EventSource

Log on to the RAM console as a RAM administrator.
Create a RAM user, grant the user the AliyunMNSFullAccess permission, and obtain the user's AccessKey ID and AccessKey secret. For more information, see Create a RAM user, Grant permissions to a RAM user, Create an AccessKey pair, and View the AccessKey information of a RAM user.

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

queue: Replace the value with the name of your Message Service (MNS) queue.
endpoint: Replace the value with the endpoint of Message Service (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 # The name of the Message Service (MNS) queue.
      waitTimeSeconds: 20
      endpoint: http://165***368.mns.<region>.aliyuncs.com # The endpoint of Message Service (MNS).

Run the following command to create the EventSource:
```
kubectl apply -f event-source.yaml
```
Run the following command to verify that the EventSource pod has started:
```
kubectl get pod
```

Step 3: Create a Sensor

Create an event-sensor.yaml file and embed the workflow definition in the Sensor. Sample configuration:

Expand to view the sample code

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

Run the following command to create the Sensor:
```
kubectl apply -f event-sensor.yaml
```
Run the following command to verify that the Sensor pod has started:
```
kubectl get pod
```

Note

When you create an EventBus using Message Service (MNS), a corresponding Message Service (MNS) queue is automatically created after the Sensor is created. The queue name follows this format: ackone-argowf-<namespace>-<sensor-name>-<sensor-uid>.

Step 4: Verify the OSS-triggered workflow

Log on to the OSS console.
Upload the following two files to an OSS bucket to trigger the workflow.
- datafile: A data file in text format with custom content.
- datafile.complete: A trigger file. This can be an empty file.
Run the following command on the workflow cluster to view the status of the workflow:
```
argo list
```
Example output:
```
NAME STATUS AGE DURATION PRIORITY
process-oss-file-kmb4k Running 13s 13s 0
```

Run the following command to retrieve the workflow logs:

argo logs process-oss-file-kmb4k

Important

The workflow name in the command, process-oss-file-kmb4k, is an example. Replace it with the actual name from the previous step's output.
The message content is Base64-encoded.

Example output:

.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

Step 5: Clean up event-related resources

Run the following commands in sequence to clean up the event-related resources:

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

Run the following command to confirm that the resource pods are deleted:
```
kubectl get pod
```