Usar uma fila do Cloud Tasks para armazenar as execuções do fluxo de trabalho

Este tutorial mostra como criar uma fila do Cloud Tasks que pode regular a taxa de execuções de fluxo de trabalho.

Há um número máximo de execuções de fluxo de trabalho ativas que podem ocorrer simultaneamente. Quando essa cota é esgotada e se o acúmulo de execuções está desativado ou se a cota de execuções em espera é atingida, todas as novas execuções falham com um código de status HTTP 429 Too many requests. Ao ativar uma fila do Cloud Tasks para executar fluxos de trabalho filhos em uma taxa definida por você, é possível evitar problemas relacionados à cota do Workflows e alcançar uma taxa de execução melhor.

O Cloud Tasks foi projetado para fornecer entrega "pelo menos uma vez". No entanto, o Workflows não garante o processamento "exatamente uma vez" de solicitações duplicadas do Cloud Tasks.

No diagrama a seguir, um fluxo de trabalho principal invoca fluxos de trabalho filhos que são regulamentados por uma fila do Cloud Tasks com uma taxa de envio aplicada.

Fluxo de trabalho principal que invoca iterações de um fluxo de trabalho secundário usando a fila do Cloud Tasks

Criar uma fila do Cloud Tasks

Crie uma fila do Cloud Tasks que possa ser usada no fluxo de trabalho principal e que permita regular a taxa de execuções de fluxo de trabalho.

Console

  1. No console do Google Cloud , acesse a página Cloud Tasks:

    Ir para o Cloud Tasks

  2. Clique em Criar fila push.

  3. Insira o Nome da fila, queue-workflow-child.

  4. Na lista Região, selecione us-central1 (Iowa).

  5. Clique em Criar.

gcloud

QUEUE=queue-workflow-child
LOCATION=us-central1
gcloud tasks queues create $QUEUE --location=$LOCATION

Criar e implantar um fluxo de trabalho filho

Um fluxo de trabalho filho pode receber e processar dados de um fluxo de trabalho pai. Crie e implante um fluxo de trabalho secundário que faça o seguinte:

  • Recebe um iteration como argumento
  • Coloca em suspensão por 10 segundos para simular algum processamento

  • Retorna uma string após a execução bem-sucedida.

Console

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

    Acessar fluxos de trabalho

  2. Clique em Criar.

  3. Insira o nome workflow-child para o novo fluxo de trabalho.

  4. Na lista Região, selecione us-central1 (Iowa).

  5. Na lista Conta de serviço, selecione a Conta de serviço padrão do Compute Engine.

  6. Clique em Próxima.

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

    main:
  params: [args]
  steps:
    - init:
        assign:
          - iteration : ${args.iteration}
    - wait:
        call: sys.sleep
        args:
            seconds: 10
    - return_message:
        return: ${"Hello world"+iteration}

  8. Clique em Implantar.

gcloud

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

    touch workflow-child.yaml

  2. Abra o arquivo de código-fonte em um editor de texto e copie o fluxo de trabalho a seguir para o arquivo.

    main:
  params: [args]
  steps:
    - init:
        assign:
          - iteration : ${args.iteration}
    - wait:
        call: sys.sleep
        args:
            seconds: 10
    - return_message:
        return: ${"Hello world"+iteration}

  3. Implante o fluxo de trabalho:

    gcloud workflows deploy workflow-child \
    --source=workflow-child.yaml \
    --location=us-central1 \
    --service-account=PROJECT_NUMBER-compute@developer.gserviceaccount.com

Criar e implantar o fluxo de trabalho principal

O fluxo de trabalho principal executa várias ramificações do fluxo de trabalho secundário usando um loop for.

  1. Copie o código-fonte que define o fluxo de trabalho principal:

    main:
  steps:
    - init:
        assign:
          - project_id: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}
          - project_number: ${sys.get_env("GOOGLE_CLOUD_PROJECT_NUMBER")}
          - location: ${sys.get_env("GOOGLE_CLOUD_LOCATION")}
          - workflow_child_name: "workflow-child"
          - queue_name: "queue-workflow-child"
    - enqueue_tasks_to_execute_child_workflow:
        for:
          value: iteration
          range: [1, 100]
          steps:
              - iterate:
                  assign:
                    - data:
                        iteration: ${iteration}
                    - exec:
                        # Encode object to JSON string in expression for workflow argument
                        argument: ${json.encode_to_string(data)}
              - create_task_to_execute_child_workflow:
                  call: googleapis.cloudtasks.v2.projects.locations.queues.tasks.create
                  args:
                      parent: ${"projects/" + project_id + "/locations/" + location + "/queues/" + queue_name}
                      body:
                        task:
                          httpRequest:
                            body: ${base64.encode(json.encode(exec))}
                            url: ${"https://workflowexecutions.googleapis.com/v1/projects/" + project_id + "/locations/" + location + "/workflows/" + workflow_child_name + "/executions"}
                            oauthToken:
                              serviceAccountEmail: ${project_number + "-compute@developer.gserviceaccount.com"}

    O fluxo de trabalho consiste nas seguintes partes:

    • Um mapa usado para atribuir constantes referentes ao fluxo de trabalho filho e ao nome da fila do Cloud Tasks. Para mais informações, consulte Maps.

    • Um loop for executado para invocar o fluxo de trabalho filho de forma iterativa. Para mais informações, consulte Iteração.

    • Uma etapa do fluxo de trabalho que cria e adiciona um grande número de tarefas à fila do Cloud Tasks para executar o fluxo de trabalho filho. Para mais informações, consulte Conector da API Cloud Tasks.

  2. Implante o fluxo de trabalho:

    Console

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

      Acessar fluxos de trabalho

    2. Clique em Criar.

    3. Insira o nome workflow-parent para o novo fluxo de trabalho.

    4. Na lista Região, selecione us-central1 (Iowa).

    5. Na lista Conta de serviço, selecione a Conta de serviço padrão do Compute Engine.

    6. Clique em Próxima.

    7. No editor de fluxo de trabalho, cole a definição do fluxo de trabalho principal.

    8. Clique em Implantar.

    gcloud

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

      touch workflow-parent.yaml

    2. Abra o arquivo de código-fonte em um editor de texto e cole a definição do fluxo de trabalho principal.

    3. Implante o fluxo de trabalho:

      gcloud workflows deploy workflow-parent \
    --source=workflow-parent.yaml \
    --location=us-central1 \
    --service-account=PROJECT_NUMBER-compute@developer.gserviceaccount.com

Executar o fluxo de trabalho principal sem limites de taxa

Execute o fluxo de trabalho principal para invocar os fluxos de trabalho secundários pela fila do Cloud Tasks. As execuções levam cerca de 10 segundos para serem concluídas.

Console

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

    Acessar fluxos de trabalho

  2. Na página Fluxos de trabalho, clique no fluxo de trabalho workflow-parent para acessar a página de detalhes dele.

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

  4. Clique em Executar novamente.

  5. Enquanto o fluxo de trabalho principal está em execução, volte para a página Fluxos de trabalho e clique no fluxo workflow-child para acessar a página de detalhes dele.

  6. Clique na guia Executions.

    Você vai ver execuções do fluxo de trabalho filho, executadas aproximadamente ao mesmo tempo, semelhantes a esta:

    Detalhes das execuções de fluxo de trabalho secundárias em execução na mesma hora.

gcloud

  1. Execute o fluxo de trabalho:

    gcloud workflows run workflow-parent \
     --location=us-central1

  2. Para verificar se uma execução de fluxo de trabalho foi acionada, liste as últimas quatro execuções:

    gcloud workflows executions list workflow-child --limit=4

    Como o número de execuções (100) está abaixo do limite de simultaneidade do Workflows, os resultados serão semelhantes a este: Problemas de cota podem surgir se você enviar milhares de execuções ao mesmo tempo. 

    NAME: projects/620278351741/locations/us-central1/workflows/workflow-child/executions/1570d06e-d133-4536-a859-b7b6a1a85524
STATE: ACTIVE
START_TIME: 2023-07-27T00:56:15.093934448Z
END_TIME:
NAME: projects/620278351741/locations/us-central1/workflows/workflow-child/executions/82724960-7d92-4961-aa2c-a0f0be46212c
STATE: ACTIVE
START_TIME: 2023-07-27T00:56:14.903007626Z
END_TIME:
NAME: projects/620278351741/locations/us-central1/workflows/workflow-child/executions/598126fb-37f9-45bc-91d8-aea7d795d702
STATE: ACTIVE
START_TIME: 2023-07-27T00:56:14.698260524Z
END_TIME:
NAME: projects/620278351741/locations/us-central1/workflows/workflow-child/executions/d2e9960b-f93f-4df4-a594-3e7e5c2be53f
STATE: ACTIVE
START_TIME: 2023-07-27T00:56:14.503818840Z
END_TIME:

Você criou e implantou um fluxo de trabalho que invoca 100 iterações do fluxo de trabalho filho.

Executar o fluxo de trabalho principal com limites de taxa

Aplique um limite de taxa de um envio por segundo à fila do Cloud Tasks e execute o fluxo de trabalho principal.

Console

  1. No console do Google Cloud , acesse a página Cloud Tasks:

    Ir para o Cloud Tasks

  2. Clique em queue-workflow-child, a fila do Cloud Tasks que você criou, e clique em Editar fila.

  3. Na seção Limites de taxas para despachos de tarefas, no campo Despachos máximos, digite 1.

  4. Clique em Salvar.

  5. Acesse a página Fluxos de trabalho:

    Acessar fluxos de trabalho

  6. Clique no fluxo de trabalho workflow-parent para acessar a página de detalhes dele.

  7. Na página Detalhes do fluxo de trabalho, clique em Executar.

  8. Clique em Executar novamente.

  9. Enquanto o fluxo de trabalho principal está em execução, volte para a página Fluxos de trabalho e clique no fluxo workflow-child para acessar a página de detalhes dele.

  10. Clique na guia Executions.

    Você vai ver execuções do fluxo de trabalho filho, executadas em uma solicitação por segundo, semelhante a esta:

    Detalhes do fluxo de trabalho secundário em execução por solicitação por segundo.

gcloud

  1. Atualize a fila do Cloud Tasks para aplicar um limite de taxa de um envio por segundo:

    gcloud tasks queues update $QUEUE \
    --max-dispatches-per-second=1 \
    --location=us-central1

  2. Execute o fluxo de trabalho:

    gcloud workflows run workflow-parent \
   --location=us-central1

  3. Para verificar se uma execução de fluxo de trabalho foi acionada, liste as últimas quatro execuções:

    gcloud workflows executions list workflow-child --limit=4

    Os resultados serão semelhantes a este, com um fluxo de trabalho executado por segundo: 

    NAME: projects/620278351741/locations/us-central1/workflows/workflow-child/executions/becf4957-9fb2-40d9-835d-0ff2dd0c1249
STATE: ACTIVE
START_TIME: 2023-07-27T01:07:24.446361457Z
END_TIME:
NAME: projects/620278351741/locations/us-central1/workflows/workflow-child/executions/6c1e7c4b-7ac6-4121-b351-1e2d56d10903
STATE: ACTIVE
START_TIME: 2023-07-27T01:07:23.448213989Z
END_TIME:
NAME: projects/620278351741/locations/us-central1/workflows/workflow-child/executions/f2ba5027-af40-4cd3-8cd0-b8033bcc6211
STATE: ACTIVE
START_TIME: 2023-07-27T01:07:22.431485914Z
END_TIME:
NAME: projects/620278351741/locations/us-central1/workflows/workflow-child/executions/ecc61ee5-fe87-49eb-8803-89dba929f6c8
STATE: ACTIVE
START_TIME: 2023-07-27T01:07:21.443466369Z
END_TIME:

Você implantou um fluxo de trabalho que invoca 100 iterações do fluxo de trabalho filho com uma taxa de envio de uma execução por segundo.