Upload a file to OSS to trigger a workflow run - Container Service for Kubernetes: Central de Documentação do Alibaba Cloud

Este tópico descreve como integrar o OSS ao Message Service (MNS). Ao enviar um arquivo para o OSS, você aciona automaticamente um workflow que processa o arquivo e gera um resultado.

Pré-requisitos

O recurso orientado a eventos está ativado.
As notificações de eventos do OSS foram criadas.
Um cluster de workflow foi criado.
A Alibaba Cloud Argo CLI está instalada.

Etapa 1: Crie um EventBus

Workflows orientados a eventos no mesmo namespace podem compartilhar um EventBus. Caso já tenha criado um EventBus, pule esta etapa e acesse a Etapa 2: Criar um EventSource.

Nota

Ao criar um EventBus com o Message Service (MNS), nenhum pod é criado.
Para usar o recurso de trigger, crie o EventBus usando NATS. O método via Message Service (MNS) não oferece suporte a triggers em um Sensor do Argo Events open source.

Método 1: Usando NATS

Crie um arquivo event-bus.yaml. Exemplo de configuração:

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

Execute o comando abaixo para criar o EventBus:
```
kubectl apply -f event-bus.yaml
```
Nota
Após a execução bem-sucedida do comando, um pod do EventBus é criado no namespace default. Realize as operações subsequentes no mesmo namespace.
Verifique se o pod do EventBus foi iniciado executando o seguinte comando:
```
kubectl get pod
```

Método 2: Usando Message Service (MNS)

Faça login no console do Simple Message Queue (antigo MNS).
Na página Topics, crie um tópico chamado argoeventbus. Em seguida, na página Topic Details, obtenha o endpoint na seção Access Point.
Faça login no console do RAM como administrador do RAM.
Crie um usuário do RAM, conceda a permissão AliyunMNSFullAccess e obtenha o AccessKey ID e o AccessKey secret do usuário.

Crie um Secret para armazenar o AccessKey ID e o AccessKey secret com o comando a seguir:

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

Crie um arquivo event-bus-mns.yaml. Exemplo de configuração:

topic: Substitua o valor pelo nome do tópico do Message Service (MNS) criado na etapa 2.
endpoint: Substitua o valor pelo endpoint obtido na etapa 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

Aplique o arquivo event-bus-mns.yaml executando o comando abaixo:
```
kubectl apply -f event-bus-mns.yaml
```

Etapa 2: Crie um EventSource

Faça login no console do RAM como administrador do RAM.
Crie um usuário do RAM, conceda a ele a permissão AliyunMNSFullAccess e obtenha seu AccessKey ID e AccessKey secret. Para mais informações, consulte Create a RAM user, Grant permissions to a RAM user, Create an AccessKey pair e View the AccessKey information of a RAM user.

Crie um arquivo event-source.yaml. Exemplo de configuração:

queue: Substitua o valor pelo nome da sua fila do Message Service (MNS).
endpoint: Substitua o valor pelo endpoint do 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).

Execute o comando abaixo para criar o EventSource:
```
kubectl apply -f event-source.yaml
```
Verifique se o pod do EventSource foi iniciado executando o seguinte comando:
```
kubectl get pod
```

Etapa 3: Crie um Sensor

Crie um arquivo event-sensor.yaml e incorpore a definição do workflow no Sensor. Exemplo de configuração:

Expandir para visualizar o código de exemplo

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

Execute o comando abaixo para criar o Sensor:
```
kubectl apply -f event-sensor.yaml
```
Verifique se o pod do Sensor foi iniciado executando o seguinte comando:
```
kubectl get pod
```

Nota

Ao criar um EventBus usando o Message Service (MNS), uma fila correspondente do Message Service (MNS) é criada automaticamente após a criação do Sensor. O nome da fila segue este formato: ackone-argowf-<namespace>-<sensor-name>-<sensor-uid>.

Etapa 4: Verifique o workflow acionado pelo OSS

Faça login no console do OSS.
Envie os dois arquivos a seguir para um bucket do OSS para acionar o workflow:
- datafile: Um arquivo de dados em formato de texto com conteúdo personalizado.
- datafile.complete: Um arquivo de gatilho. Pode ser um arquivo vazio.
Visualize o status do workflow executando o comando abaixo no cluster de workflow:
```
argo list
```
Exemplo de saída:
```
NAME STATUS AGE DURATION PRIORITY
process-oss-file-kmb4k Running 13s 13s 0
```

Recupere os logs do workflow com o seguinte comando:

argo logs process-oss-file-kmb4k

Importante

O nome do workflow no comando, process-oss-file-kmb4k, é apenas um exemplo. Substitua-o pelo nome real obtido na saída da etapa anterior.
O conteúdo da mensagem está codificado em Base64.

Exemplo de saída:

.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

Etapa 5: Limpar recursos relacionados a eventos

Limpe os recursos relacionados a eventos executando os comandos abaixo em sequência:

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

Confirme a exclusão dos pods de recursos com o seguinte comando:
```
kubectl get pod
```