Executar um job do Cloud Run que processa dados de eventos salvos no Cloud Storage

É possível usar o Workflows para executar um job do Cloud Run como parte de um fluxo de trabalho que realiza processamento de dados mais complexo ou que orquestra um sistema de jobs atuais.

Este tutorial demonstra como usar o Workflows para executar um job do Cloud Run que processa dados de eventos armazenados em um bucket do Cloud Storage. Ao armazenar o payload de evento em um bucket do Cloud Storage, é possível criptografar os dados usando chaves de criptografia gerenciadas pelo cliente, o que não é possível se você estiver transmitindo os dados de eventos como variáveis de ambiente para o job do Cloud Run.

O diagrama a seguir fornece uma visão geral:

O evento do Pub/Sub é roteado pelo gatilho do Eventarc para o Workflows e salvo no bucket do Cloud Storage. O job do Cloud Run processa os dados de eventos armazenados no bucket.

Criar um job do Cloud Run

Este tutorial usa um exemplo de código que pode ser encontrado no GitHub. O script de implantação cria uma imagem de contêiner para criar um job do Cloud Run. O script também cria um bucket do Cloud Storage. O job do Cloud Run lê todos os dados de eventos armazenados no bucket do Cloud Storage e depois imprime esses dados.

  1. Se você estiver executando o script de implantação no Cloud Shell e se a conta de serviço padrão do Compute Engine não tiver o papel de editor, conceda os seguintes papéis no projeto à conta de serviço padrão do Compute Engine. Caso contrário, pule esta etapa e continue clonando o repositório do app de exemplo na próxima etapa.

    1. Conceda o papel de gravador do Artifact Registry (roles/artifactregistry.writer):

      gcloud projects add-iam-policy-binding PROJECT_ID \
    --member=serviceAccount:PROJECT_NUMBER-compute@developer.gserviceaccount.com \
    --role=roles/artifactregistry.writer

      Substitua PROJECT_NUMBER pelo número do seu projeto Google Cloud. Encontre o número do projeto na página Boas-vindas do console do Google Cloud ou executando o seguinte comando:

      gcloud projects describe PROJECT_ID --format='value(projectNumber)'

    2. Conceda o papel Usuário de objetos do Storage (roles/storage.objectUser):

      gcloud projects add-iam-policy-binding PROJECT_ID \
    --member=serviceAccount:PROJECT_NUMBER-compute@developer.gserviceaccount.com \
    --role=roles/storage.objectUser

    3. Conceda o papel de gravador de registros do Logging (roles/logging.logWriter):

      gcloud projects add-iam-policy-binding PROJECT_ID \
    --member=serviceAccount:PROJECT_NUMBER-compute@developer.gserviceaccount.com \
    --role=roles/logging.logWriter

  2. Clone o repositório do app de amostra na máquina local para acessar o código de exemplo:

    git clone https://github.com/GoogleCloudPlatform/workflows-demos.git

    Outra opção é baixar a amostra como um arquivo ZIP.

  3. Acesse o diretório que contém o exemplo de código:

    cd workflows-demos/cloud-run-jobs-payload-gcs/message-payload-job

  4. Crie o job do Cloud Run executando o script de implantação:

    ./deploy-job.sh

O script cria um bucket do Cloud Storage com o nome message-payload-PROJECT_ID, em que PROJECT_ID é o ID do seu projeto Google Cloud . Um job do Cloud Run chamado message-payload-job também é criado.

Implante um fluxo de trabalho que execute o job do Cloud Run

Defina e implante um fluxo de trabalho que execute o job do Cloud Run que você acabou de criar. Uma definição de fluxo de trabalho é composta por uma série de etapas descritas usando a sintaxe do Workflows.

O fluxo de trabalho recebe um evento, salva os dados dele em um bucket do Cloud Storage e executa um job do Cloud Run para processar os dados.

Console

  1. No console Google Cloud , acesse a página Fluxos de trabalho:

    Acessar fluxos de trabalho

  2. Clique em Criar.

  3. Insira um nome para o novo fluxo de trabalho, como message-payload-workflow.

  4. Escolha uma região apropriada, por exemplo, us-central1.

  5. No campo Conta de serviço, selecione a conta de serviço que você criou anteriormente.

    A conta de serviço serve como a identidade do fluxo de trabalho. Você já precisa ter concedido os seguintes papéis à conta de serviço:

    • Administrador do Cloud Run: para executar jobs do Cloud Run
    • Gravador de registros: para gravar entradas de registro
    • Criador de objetos do Storage: para criar objetos no Cloud Storage

  6. Clique em Próxima.

  7. No editor de fluxo de trabalho, insira a seguinte definição:

    main:
    params: [event]
    steps:
        - init:
            assign:
                - project_id: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}
                - event_bucket: ${"message-payload-" + project_id}
                - event_file: ${event.id + ".data.json"}
                - job_name: message-payload-job
                - job_location: us-central1
        - log_event:
            call: sys.log
            args:
                data: ${event}
        - write_payload_to_gcs:
            call: http.post
            args:
                url: ${"https://storage.googleapis.com/upload/storage/v1/b/" + event_bucket + "/o"}
                auth:
                    type: OAuth2
                query:
                    name: ${event_file}
                body:
                    ${event.data}
        - run_job_to_process_payload:
            call: googleapis.run.v1.namespaces.jobs.run
            args:
                name: ${"namespaces/" + project_id + "/jobs/" + job_name}
                location: ${job_location}
                body:
                    overrides:
                        containerOverrides:
                            env:
                                - name: INPUT_BUCKET
                                  value: ${event_bucket}
                                - name: INPUT_FILE
                                  value: ${event_file}
            result: job_execution
        - finish:
            return: ${job_execution}

  8. Clique em Implantar.

gcloud

  1. Crie um arquivo de código-fonte para seu fluxo de trabalho:

    touch message-payload-workflow.yaml

  2. Copie a seguinte definição de fluxo de trabalho para message-payload-workflow.yaml:

    main:
    params: [event]
    steps:
        - init:
            assign:
                - project_id: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}
                - event_bucket: ${"message-payload-" + project_id}
                - event_file: ${event.id + ".data.json"}
                - job_name: message-payload-job
                - job_location: us-central1
        - log_event:
            call: sys.log
            args:
                data: ${event}
        - write_payload_to_gcs:
            call: http.post
            args:
                url: ${"https://storage.googleapis.com/upload/storage/v1/b/" + event_bucket + "/o"}
                auth:
                    type: OAuth2
                query:
                    name: ${event_file}
                body:
                    ${event.data}
        - run_job_to_process_payload:
            call: googleapis.run.v1.namespaces.jobs.run
            args:
                name: ${"namespaces/" + project_id + "/jobs/" + job_name}
                location: ${job_location}
                body:
                    overrides:
                        containerOverrides:
                            env:
                                - name: INPUT_BUCKET
                                  value: ${event_bucket}
                                - name: INPUT_FILE
                                  value: ${event_file}
            result: job_execution
        - finish:
            return: ${job_execution}

  3. Implante o fluxo de trabalho digitando o seguinte comando:

    gcloud workflows deploy message-payload-workflow \
    --location=us-central1 \
    --source=message-payload-workflow.yaml \
    --service-account=SERVICE_ACCOUNT_NAME@PROJECT_ID.iam.gserviceaccount.com

    Substitua:

    • SERVICE_ACCOUNT_NAME: o nome da conta de serviço que você criou antes
    • PROJECT_ID: o ID do seu projetoGoogle Cloud

    A conta de serviço serve como a identidade do fluxo de trabalho. Você já precisa ter concedido os seguintes papéis à conta de serviço:

    • roles/logging.logWriter: para gravar entradas de registro
    • roles/run.admin: para executar jobs do Cloud Run
    • roles/storage.objectCreator: para criar objetos no Cloud Storage

O fluxo de trabalho faz o seguinte:

  1. Etapa init: aceita um evento como argumento e define as variáveis necessárias.

  2. A etapa log_event cria uma entrada de registro no Cloud Logging usando a função sys.log.

  3. Etapa write_payload_to_gcs: faz uma solicitação HTTP POST e grava os dados de payload do evento em um arquivo de bucket do Cloud Storage.

  4. A etapa run_job_to_process_payload usa o método do conector da API Cloud Run Admin, googleapis.run.v1.namespaces.jobs.run, para executar o job. O bucket do Cloud Storage e o nome do arquivo de dados são transmitidos como variáveis de substituição do fluxo de trabalho para o job.

  5. Etapa finish: retorna informações sobre a execução do job como resultado do fluxo de trabalho.

Criar um tópico do Pub/Sub

Crie um tópico do Pub/Sub para publicar uma mensagem nele. Um evento do Pub/Sub é usado para demonstrar como encaminhar um evento usando o Workflows e salvar o evento no Cloud Storage para que um job do Cloud Run possa processar os dados do evento.

Console

  1. No console Google Cloud , acesse a página Tópicos.

    Acesse Tópicos

  2. Clique em Criar tópico.

  3. No campo ID do tópico, insira um ID para o tópico, como message-payload-topic.

  4. Mantenha a opção Adicionar uma assinatura padrão.

  5. Não selecione as outras opções.

  6. Clique em Criar.

gcloud

Para criar um tópico com o ID message-payload-topic, execute o comando gcloud pubsub topics create:

gcloud pubsub topics create message-payload-topic

Criar um gatilho do Eventarc para rotear eventos ao fluxo de trabalho

Para executar automaticamente o fluxo de trabalho e, por sua vez, o job do Cloud Run, crie um gatilho do Eventarc que responda a eventos do Pub/Sub e os encaminhe para o fluxo de trabalho. Sempre que uma mensagem é gravada no tópico do Pub/Sub, o evento aciona uma execução do fluxo de trabalho.

Console

  1. No console Google Cloud , acesse a página Fluxos de trabalho:

    Acessar fluxos de trabalho

  2. Clique no nome do fluxo de trabalho, como message-payload-workflow.

  3. Na página Detalhes do fluxo de trabalho, clique em Editar.

  4. Na página Editar fluxo de trabalho, na seção Gatilhos, clique em Adicionar novo gatilho > Eventarc.

    O painel Gatilho do Eventarc é aberto.

  5. No campo Nome do gatilho, insira um nome para o gatilho, como message-payload-trigger.

  6. Na lista Provedor de eventos, selecione Cloud Pub/Sub.

  7. Na lista Evento, selecione google.cloud.pubsub.topic.v1.messagePublished.

  8. Na lista Selecionar um tópico do Cloud Pub/Sub, escolha o tópico do Pub/Sub criado anteriormente.

  9. No campo Conta de serviço, selecione a conta de serviço que você criou anteriormente.

    A conta de serviço serve como a identidade do gatilho. Você já precisa ter concedido os seguintes papéis à conta de serviço:

    • Receptor de eventos do Eventarc: para receber eventos
    • Chamador do Workflows: para executar fluxos de trabalho

  10. Clique em Salvar acionador.

    O gatilho do Eventarc agora aparece na seção Gatilhos da página Editar fluxo de trabalho.

  11. Clique em Próxima.

  12. Clique em Implantar.

gcloud

Crie um gatilho do Eventarc executando o seguinte comando:

gcloud eventarc triggers create message-payload-trigger \
    --location=us-central1 \
    --destination-workflow=message-payload-workflow \
    --destination-workflow-location=us-central1 \
    --event-filters="type=google.cloud.pubsub.topic.v1.messagePublished" \
    --transport-topic=projects/PROJECT_ID/topics/message-payload-topic \
    --service-account=SERVICE_ACCOUNT_NAME@PROJECT_ID.iam.gserviceaccount.com

Substitua:

  • PROJECT_ID: o ID do projeto Google Cloud
  • SERVICE_ACCOUNT_NAME: o nome da conta de serviço que você criou anteriormente.

A conta de serviço serve como a identidade do gatilho. Você já precisa ter concedido os seguintes papéis à conta de serviço:

  • roles/eventarc.eventReceiver: para receber eventos.
  • roles/workflows.invoker: para executar fluxos de trabalho

Acione o fluxo de trabalho

Teste o sistema de ponta a ponta publicando uma mensagem no tópico do Pub/Sub e gerando um evento. Para mais informações, consulte como acionar um fluxo de trabalho com eventos ou mensagens do Pub/Sub.

  1. Publique uma mensagem para o tópico do Pub/Sub a fim de gerar um evento:

    gcloud pubsub topics publish message-payload-topic --message="Hello World"

    O evento é encaminhado para o fluxo de trabalho, que registra a mensagem do evento, salva os dados do evento no bucket do Cloud Storage e executa o job do Cloud Run para processar os dados salvos no Cloud Storage. Isso pode levar um minuto.

  2. Confirme se o job do Cloud Run foi executado conforme esperado. Para isso, confira as execuções do job:

    gcloud run jobs executions list --job=message-payload-job

    Uma nova execução de job vai aparecer na saída.

  3. Para ver as entradas de registro relacionadas a eventos criadas ao acionar o fluxo de trabalho, execute o seguinte comando:

    gcloud logging read "resource.type=cloud_run_job AND textPayload:Payload"

  4. Procure uma entrada de registro semelhante a esta:

    textPayload: "Payload: {'message': {'data': 'SGVsbG8gV29ybGQ=', 'messageId': '8834108235224238',\
\ 'publishTime': '2023-09-20T17:07:52.921Z'}, 'subscription': 'projects/MY_PROJECT/subscriptions/eventarc-us-central1-message-payload-trigger-sub-741'}"
...
resource:
labels:
  job_name: message-payload-job
  location: us-central1
  project_id: MY_PROJECT
type: cloud_run_job
textPayload: Processing message payload gs://message-payload-MY_PROJECT/8254002311197919.data.json

  5. Para confirmar se os resultados estão corretos, confira os dados de eventos no objeto do bucket do Cloud Storage.

    1. Recupere o nome do bucket:

      gcloud storage ls

      O resultado será assim: 

      gs://message-payload-PROJECT_ID/

    2. Liste os objetos no seu bucket:

      gcloud storage ls gs://message-payload-PROJECT_ID/** --recursive

      A saída será semelhante a esta:

      gs://message-payload-PROJECT_ID/OBJECT_ID.data.json

      Observe o OBJECT_ID a ser usado na próxima etapa.

    3. Faça o download do objeto no bucket como um arquivo:

      gcloud storage cp gs://message-payload-PROJECT_ID/OBJECT_ID.data.json message-event.txt

      Substitua OBJECT_ID pelo ID retornado na etapa anterior.

    4. Em um editor de texto, abra o arquivo message-event.txt. O corpo do evento gravado no arquivo será semelhante a este:

      {
  "message": {
    "data": "SGVsbG8gV29ybGQ=",
    "messageId": "8254002311197919",
    "publishTime": "2023-09-20T16:54:29.312Z"
  },
  "subscription": "projects/MY_PROJECT/subscriptions/eventarc-us-central1-message-payload-trigger-sub-741"
}

      Se você decodificar o valor de dados de SGVsbG8gV29ybGQ= do formato Base64, "Hello World" será retornado.