OSSへファイルをアップロードしてワークフロー実行をトリガーする - Container Service for Kubernetes

前提条件

イベント駆動機能を有効化していること。
OSS イベント通知を作成していること。
ワークフロークラスターを作成していること。
Alibaba Cloud Argo CLI をインストールしていること。

ステップ 1: EventBus の作成

同じ名前空間内のイベント駆動型ワークフローは、EventBus を共有できます。EventBus をすでに作成している場合は、このステップをスキップしてステップ 2: EventSource の作成に進んでください。

説明

Simple Message Queue (旧称：MNS) を使用して EventBus を作成する場合、Pod は作成されません。
トリガー機能を使用するには、NATS を使用して EventBus を作成してください。Simple Message Queue (旧称：MNS) では、オープンソースの Argo Events Sensor におけるトリガーはサポートされていません。

方法 1: NATS の使用

event-bus.yaml ファイルを作成します。設定例：

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

次のコマンドを実行して EventBus を作成します：
```
kubectl apply -f event-bus.yaml
```
説明
コマンドが成功すると、default 名前空間に EventBus Pod が作成されます。以降の操作は同じ名前空間で実行する必要があります。
次のコマンドを実行して EventBus Pod が起動したことを確認します：
```
kubectl get pod
```

方法 2: Simple Message Queue (旧称：MNS) の使用

MNS コンソールにログインします。
[トピック]ページで、 argoeventbus という名前のトピックを作成します。[トピック詳細]ページで、エンドポイントセクションからエンドポイントを取得します。
RAM 管理者としてRAMコンソールにログインします。
RAM ユーザーを作成し、AliyunMNSFullAccess 権限を付与して、そのユーザーの AccessKey ID と AccessKey シークレットを取得します。
次のコマンドを実行して、AccessKey ID と AccessKey シークレットを保存する Secret を作成します：
```
kubectl create secret generic mns-secret\
  --from-literal=accesskey=*** \
  --from-literal=secretkey=***
```

event-bus-mns.yaml ファイルを作成します。設定例：

topic：ステップ2 で作成したSimple Message Queue (旧称：MNS) トピックの名前に置き換えます。
endpoint：ステップ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  # Simple Message Queue (旧称：MNS) のトピック名。
    endpoint: http://165***368.mns.<region>.aliyuncs.com

次のコマンドを実行してevent-bus-mns.yaml ファイルを適用します：
```
kubectl apply -f event-bus-mns.yaml
```

ステップ 2: EventSource の作成

RAM 管理者としてRAMコンソールにログインします。
RAM ユーザーを作成し、AliyunMNSFullAccess 権限を付与して、そのユーザーの AccessKey ID と AccessKey シークレットを取得します。詳細については、「RAMユーザーの作成」、「RAMユーザーへの権限付与」、「AccessKeyペアの作成」、および「RAMユーザーのAccessKey情報の表示」をご参照ください。

event-source.yaml ファイルを作成します。設定例：

queue：ご使用のSimple Message Queue (旧称：MNS) キューの名前に置き換えます。
endpoint：Simple Message Queue (旧称：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 # Simple Message Queue (旧称：MNS) のキュー名。
      waitTimeSeconds: 20
      endpoint: http://165***368.mns.<region>.aliyuncs.com # Simple Message Queue (旧称：MNS) のエンドポイント。

次のコマンドを実行して EventSource を作成します：
```
kubectl apply -f event-source.yaml
```
次のコマンドを実行して EventSource Pod が起動したことを確認します：
```
kubectl get pod
```

ステップ 3: Sensor の作成

event-sensor.yaml ファイルを作成し、ワークフロー定義を 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
                    # この値は上書きされます
                    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

次のコマンドを実行して Sensor を作成します：
```
kubectl apply -f event-sensor.yaml
```
次のコマンドを実行して Sensor Pod が起動したことを確認します：
```
kubectl get pod
```

説明

Simple Message Queue (旧称：MNS) を使用して EventBus を作成した場合、Sensor の作成後に対応するSimple Message Queue (旧称：MNS) キューが自動的に作成されます。キュー名は ackone-argowf-<namespace>-<sensor-name>-<sensor-uid> という形式になります。

ステップ 4: OSS トリガー型ワークフローの検証

OSSコンソールにログインします。
次の 2 つのファイルを OSS バケットにアップロードして、ワークフローをトリガーします。
- data：カスタムコンテンツを含むテキスト形式のデータファイル。
- data.complete：トリガーファイル。空のファイルでも構いません。
ワークフロークラスターで次のコマンドを実行して、ワークフローのステータスを表示します：
```
argo list
```
出力例：
```
NAME STATUS AGE DURATION PRIORITY
process-oss-file-kmb4k Running 13s 13s 0
```

次のコマンドを実行してワークフローログを取得します：

argo logs process-oss-file-kmb4k

重要

コマンド内のワークフロー名 process-oss-file-kmb4k は一例です。前の手順で出力された実際の名前に置き換えてください。
メッセージの内容は 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

ステップ 5: イベント関連リソースのクリーンアップ

次のコマンドを順番に実行して、イベント関連リソースをクリーンアップします：
```
kubectl delete sensor process-oss-file
kubectl delete eventsource ali-mns
kubectl delete eventbus default
```
次のコマンドを実行して、リソース Pod が削除されたことを確認します：
```
kubectl get pod
```