Como migrar filas push para o Cloud Tasks

Nesta página, descrevemos como migrar o código de fila push do Task Queues para o Cloud Tasks. O Cloud Tasks agora é a maneira preferencial de trabalhar com filas push do App Engine.

Visão geral

Com o Cloud Tasks, é possível acessar o mesmo serviço acessado com a API RPC Task Queues. Assim, você não precisa recriar suas filas e tarefas push atuais. No entanto, é necessário migrar o código que cria ou interage com filas ou tarefas push para usar a API Cloud Tasks.

É possível criar e interagir com filas e tarefas push usando as APIs REST e RPC Cloud Tasks, a biblioteca de cliente do Cloud Tasks para Python, a Google Cloud CLI e o Console do Google Cloud. Nesta página, apresentamos exemplos que usam a CLI gcloud e a biblioteca de cliente do Cloud Tasks para Python.

No Cloud Tasks, todas as filas operam como filas push. No restante deste guia e na documentação do Cloud Tasks, o termo fila é equivalente ao termo fila push. Da mesma forma, o termo tarefa é equivalente ao termo tarefa push.

Recursos não disponíveis no Cloud Tasks

Os recursos a seguir não estão disponíveis no Cloud Tasks:

• Enfileirar tarefas em transações do Datastore
• Usar a biblioteca de tarefas adiadas em vez de um serviço de worker
• Trabalhar com tarefas em aplicativos com vários locatários
• Simular com o servidor de desenvolvimento local
• adição de tarefas de maneira assíncrona

Preços e cotas

Migrar as filas push para o Cloud Tasks pode afetar os preços e as cotas do seu aplicativo.

Preços

O Cloud Tasks tem seus próprios preços. Assim como no Task Queues, o envio de solicitações para o aplicativo do App Engine com uma tarefa pode gerar custos.

Cotas

As cotas do Cloud Tasks são diferentes das cotas do Task Queues. Assim como no Task Queues, o envio de solicitações para o aplicativo do App Engine a partir do Cloud Tasks pode afetar as cotas de solicitação do App Engine.

Antes de migrar

Nesta seção, discutimos o que você precisa fazer antes de migrar as filas push para o Cloud Tasks.

Como migrar filas pull

Use o guia para migrar filas pull para migrá-las antes de usar este guia para migrar suas filas push. A migração de filas pull após a migração de filas push não é recomendada porque o uso obrigatório do arquivo queue.yaml provavelmente causará um comportamento inesperado com o Cloud Tasks.

Como proteger a configuração da fila

Depois de iniciar o processo de migração para o Cloud Tasks, modificar o arquivo queue.yaml pode causar comportamento inesperado e não é recomendado. Proteja a configuração da fila contra modificações pelo arquivo queue.yaml seguindo estas etapas:

1. Configure a CLI gcloud para omitir o arquivo queue.yaml em implantações futuras.

   Adicione o arquivo queue.yaml a um arquivo .gcloudignore. Para verificar se você já tem um arquivo .gcloudignore, execute o seguinte comando no terminal a partir do diretório de nível superior do aplicativo. Esse comando exibirá o nome do arquivo, se ele existir.

   ls -a | grep .gcloudignore

   Saiba mais sobre arquivos .gcloudignore lendo a referência de .gcloudignore.

2. Restrinja as permissões no arquivo queue.yaml.

   Siga as práticas recomendadas descritas no nosso guia sobre como proteger a configuração da fila.

3. Saiba mais sobre o Cloud Tasks e o arquivo queue.yaml (opcional).

   Quando você usa a API Cloud Tasks para gerenciar a configuração da fila, a implantação de um arquivo queue.yaml substitui a configuração definida pelo Cloud Tasks, o que pode causar um comportamento inesperado. Leia Como usar o gerenciamento de filas versus queue.yaml para saber mais.

Como ativar a API Cloud Tasks

Para ativar a API Cloud Tasks, clique em Ativar na API Cloud Tasks na biblioteca de APIs. Se houver um botão Gerenciar em vez de Ativar, você já ativou a API Cloud Task para o projeto e não precisa fazer isso novamente.

Como autenticar o aplicativo na API Cloud Tasks

Você precisa autenticar o aplicativo na API Cloud Tasks. Nesta seção, discutimos a autenticação para dois casos de uso diferentes.

Para desenvolver ou testar seu aplicativo localmente, recomendamos usar uma conta de serviço. Veja instruções sobre como configurar uma conta de serviço e conectá-la ao aplicativo em Como receber e fornecer credenciais da conta de serviço manualmente.

Para implantar o aplicativo no App Engine, você não precisa fornecer nenhuma nova autenticação. As Application Default Credentials (ADC, na sigla em inglês) inferem detalhes de autenticação de aplicativos do App Engine.

Como fazer o download da CLI gcloud

Faça o download e instale a CLI gcloud para usar a CLI gcloud com a API Cloud Tasks, se você não tiver instalado anteriormente. Execute o seguinte comando do seu terminal se você já tiver a CLI gcloud instalada.

gcloud components update

Como importar a biblioteca de cliente do Python

Siga as etapas abaixo para usar a biblioteca de cliente do Cloud Tasks para Python com seu aplicativo do App Engine:

1. Crie um diretório para armazenar suas bibliotecas de terceiros, como lib/:

   mkdir lib

2. Copie as bibliotecas necessárias.

   Recomendamos usar um arquivo de requisitos pip em vez de instalar as bibliotecas uma de cada vez a partir da linha de comando. Crie um arquivo requirements.txt na mesma pasta que o arquivo app.yaml, se ainda não houver um arquivo requirements.txt. Adicione as duas linhas a seguir:

   google-cloud-tasks
   

   Use pip (versão 6 ou posterior) com a sinalização -t <directory> para copiar a biblioteca do Cloud Tasks especificada no arquivo requirements.txt para a pasta criada na etapa anterior. Exemplo:

   pip install -t lib -r requirements.txt

3. Especifique as bibliotecas RPC e setuptools na seção libraries do arquivo app.yaml:

   libraries:
   - name: grpcio
     version: 1.0.0
   - name: setuptools
     version: 36.6.0
   

4. Use o módulo pkg_resources para garantir que o aplicativo use a distribuição correta da biblioteca de cliente do Cloud Tasks para Python.

   Crie um arquivo appengine_config.py na mesma pasta que o arquivo app.yaml, se você ainda não tiver um. Adicione a instrução a seguir ao arquivo appengine_config.py:

   # appengine_config.py
   import pkg_resources
   from google.appengine.ext import vendor
   
   # Set path to your libraries folder.
   path = 'lib'
   # Add libraries installed in the path folder.
   vendor.add(path)
   # Add libraries to pkg_resources working set to find the distribution.
   pkg_resources.working_set.add_entry(path)

   O arquivo appengine_config.py acima pressupõe que o diretório de trabalho atual seja o local da pasta lib. Em alguns casos, como testes de unidade, o diretório de trabalho atual pode ser diferente. Para evitar erros, transmita explicitamente o caminho completo para a pasta lib usando:

   import os
   path = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'lib')

5. Importe a biblioteca de cliente do Cloud Tasks para Python em qualquer arquivo que use filas push da API Task Queues:

   from google.cloud import tasks

   Após concluir a migração completa de todo o código que cria ou interage com filas push para o Cloud Tasks, remova as instruções que importam a API Task Queues, como from google.appengine.api import taskqueue.

Como criar e gerenciar filas

Nesta seção, descrevemos como criar e gerenciar filas usando a API Cloud Tasks.

Com o Cloud Tasks, você não precisa usar um arquivo queue.yaml para criar ou gerenciar filas. Em vez disso, use a API Cloud Tasks. Não é recomendado usar um arquivo queue.yaml e a API Cloud Tasks, mas isso pode ser inevitável na migração do Task Queues para o Cloud Tasks, dependendo do aplicativo. Leia Como usar o gerenciamento de filas versus queue.yaml para saber mais sobre as práticas recomendadas.

Como criar filas

Leia esta seção se o aplicativo criar filas de maneira programática ou se você quiser criar filas adicionais a partir da linha de comando.

No Cloud Tasks, os nomes de fila têm o formato projects/PROJECT_ID/locations/LOCATION_ID/queues/QUEUE_ID. A parte LOCATION_ID do nome da fila corresponde a uma região do Google Cloud. A parte QUEUE_ID do nome da fila é equivalente ao campo name da fila no Task Queues. O nome da fila é gerado durante a criação da fila com base no projeto, na região e no QUEUE_ID que você especificar.

Em geral, o local da fila (ou seja, a região) precisa ser o mesmo da região do aplicativo. As duas exceções a essa regra são aplicativos que usam a região europe-west e aplicativos que usam a região us-central. No Cloud Tasks, essas regiões são chamadas de europe-west1 e us-central1, respectivamente.

É possível especificar a configuração de fila opcional durante a criação da fila, mas você também pode fazer isso atualizando a fila depois de criá-la.

Você não precisa recriar filas atuais. Em vez disso, migre o código que interage com as filas atuais lendo as partes relevantes deste guia.

Como reutilizar nomes de fila

Aguarde sete dias após a exclusão de uma fila para criar uma com o mesmo ID de fila no mesmo projeto e local (ou seja, região).

O exemplo a seguir cria duas filas usando o Cloud Tasks. A primeira fila tem o ID queue-blue e está configurada para enviar todas as tarefas para a versão v2 do serviço task-module a uma taxa de 5/s. A segunda fila tem o ID queue-red e envia tarefas a uma taxa de 1/s. Ambos são criados no projeto com o ID do projeto my-project-id no local us-central1. Esse é o equivalente no Cloud Tasks à criação de filas no Task Queues.

gcloud tasks queues create queue-blue \
--max-dispatches-per-second=5 \
--routing-override=service:task-module,version:v2

gcloud tasks queues create queue-red \
--max-dispatches-per-second=1

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# queue_blue_name = 'queue-blue'
# queue_red_name = 'queue-red'

parent = f"projects/{project}/locations/{location}"

queue_blue = {
    'name': client.queue_path(project, location, queue_blue_name),
    'rate_limits': {
        'max_dispatches_per_second': 5
    },
    'app_engine_routing_override': {
        'version': 'v2',
        'service': 'task-module'
    }
}

queue_red = {
    'name': client.queue_path(project, location, queue_red_name),
    'rate_limits': {
        'max_dispatches_per_second': 1
    }
}

queues = [queue_blue, queue_red]
for queue in queues:
    response = client.create_queue(parent=parent, queue=queue)
    print(response)

Saiba mais na referência do Cloud Tasks Como criar uma fila do Cloud Tasks.

Como definir a taxa de processamento da fila

A tabela abaixo lista os campos que diferenciam o Task Queues do Cloud Tasks.

É possível definir a taxa de processamento da fila ao criá-la ou atualizá-la depois. O exemplo abaixo usa o Cloud Tasks para definir a taxa de processamento de uma fila chamada queue-blue já criada. Se queue-blue tiver sido criado ou configurado usando um arquivo queue.yaml, o exemplo abaixo redefinirá max_burst_size com base no valor max_dispatches_per_second de 20. Esse é o equivalente no Cloud Tasks à definição da taxa de processamento de filas no Task Queues.

gcloud tasks queues update queue-blue \
--max-dispatches-per-second=20 \
--max-concurrent-dispatches=10

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# queue = 'queue-blue'

# Get queue object
queue_path = client.queue_path(project, location, queue)
queue = client.get_queue(name=queue_path)

# Update queue object
queue.rate_limits.max_dispatches_per_second = 20
queue.rate_limits.max_concurrent_dispatches = 10

response = client.update_queue(queue=queue)
print(response)

Saiba mais em Definir limites de taxa.

Como desativar e retomar filas

O Cloud Tasks usa o termo pausar da mesma forma que o Task Queues usa o termo desativar. Pausar uma fila interrompe a execução das tarefas até que a fila seja retomada. No entanto, é possível continuar adicionando tarefas a uma fila pausada. O Cloud Tasks usa o termo retomar da mesma forma que o Task Queues.

O exemplo abaixo pausa uma fila com o ID queue1. Esse é o equivalente no Cloud Tasks à desativação de filas no Task Queues.

gcloud tasks queues pause queue1

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# queue = 'queue1'

queue_path = client.queue_path(project, location, queue)
response = client.pause_queue(name=queue_path)

Saiba mais na referência do Cloud Tasks Como pausar filas.

Como excluir filas

Depois de excluir uma fila, aguarde sete dias antes de criar uma fila com o mesmo nome. Considere limpar todas as tarefas de uma fila e reconfigurar a fila, se não puder esperar sete dias.

O exemplo abaixo exclui a fila com o ID queue1. Esse é o equivalente no Cloud Tasks à exclusão de filas no Task Queues.

gcloud tasks queues delete queue1

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# queue = 'queue1'

queue_path = client.queue_path(project, location, queue)
response = client.delete_queue(name=queue_path)

Saiba mais na referência do Cloud Tasks Como excluir filas.

Como criar e gerenciar tarefas

Nesta seção, descrevemos como criar e gerenciar tarefas usando a API Cloud Tasks.

Como criar tarefas

A tabela abaixo lista os campos que diferenciam o Task Queues do Cloud Tasks.

No exemplo a seguir, criamos uma tarefa que encaminha para um serviço do App Engine chamado worker com o gerenciador /update_counter . Esse é o equivalente no Cloud Tasks à criação de tarefas no Task Queues.

gcloud tasks create-app-engine-task --queue=default \
--method=POST --relative-uri=/update_counter --routing=service:worker \
--body-content=10

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# queue = 'default'
amount = 10

parent = client.queue_path(project, location, queue)

task = {
    'app_engine_http_request': {
        'http_method': tasks.HttpMethod.POST,
        'relative_uri': '/update_counter',
        'app_engine_routing': {
            'service': 'worker'
        },
        'body': str(amount).encode()
    }
}

response = client.create_task(parent=parent, task=task)
eta = response.schedule_time.strftime("%m/%d/%Y, %H:%M:%S")
print('Task {} enqueued, ETA {}.'.format(response.name, eta))

Saiba mais na referência do Cloud Tasks Como criar tarefas do App Engine.

Como especificar o serviço de destino e o roteamento

A especificação do serviço, da versão e da instância de destino do App Engine para tarefas do App Engine é opcional. Por padrão, as tarefas do App Engine são encaminhadas para o serviço, a versão e a instância que são o padrão no momento em que a tarefa é tentada.

Defina a propriedade app_engine_routing da tarefa durante a criação dela para especificar um serviço, uma versão ou uma instância diferente do App Engine.

Para encaminhar todas as tarefas de uma determinada fila para o mesmo serviço, versão e instância do App Engine, defina a propriedade app_engine_routing_override na fila.

Saiba mais na referência do Cloud Tasks Configurar roteamento.

Como passar dados ao gerenciador

Assim como no Task Queues, é possível transmitir dados para o gerenciador de duas maneiras usando o Cloud Tasks. Você pode transmitir dados como parâmetros de consulta no URI relativo ou transmiti-los no corpo da solicitação usando os métodos HTTP POST ou PUT.

O Cloud Tasks usa o termo corpo da mesma forma que o Task Queues usa o termo payload. No Cloud Tasks, o tipo de conteúdo padrão do corpo é octet-stream, e não texto simples. É possível definir o tipo de conteúdo do body especificando-o no cabeçalho.

O exemplo a seguir transmite uma chave para o gerenciador /update_counter de duas maneiras diferentes. Esse é o equivalente no Cloud Tasks à transmissão de dados para o gerenciador no Task Queues.

gcloud tasks create-app-engine-task --queue=default --method=GET  \
--relative-uri=/update_counter?key=blue --routing=service:worker

gcloud tasks create-app-engine-task --queue=default --method=POST \
--relative-uri=/update_counter --routing=service:worker \
--body-content="{'key': 'blue'}"

import json
client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# queue = 'default'

parent = client.queue_path(project, location, queue)

task1 = {
    'app_engine_http_request': {
        'http_method': tasks.HttpMethod.POST,
        'relative_uri': '/update_counter?key=blue',
        'app_engine_routing': {
            'service': 'worker'
        }
    }
}

task2 = {
    'app_engine_http_request': {
        'http_method': tasks.HttpMethod.POST,
        'relative_uri': '/update_counter',
        'app_engine_routing': {
            'service': 'worker'
        },
        'headers': {
            'Content-Type': 'application/json'
        },
        'body': json.dumps({'key': 'blue'}).encode()
    }
}

response = client.create_task(parent=parent, task=task1)
print(response)
response = client.create_task(parent=parent, task=task2)
print(response)

Como nomear tarefas

A especificação do nome da tarefa é opcional. Se você não especificar o nome da tarefa, o Cloud Tasks criará um para você, gerando um ID da tarefa e inferindo o projeto e o local (isto é, a região) com base na fila especificada durante a criação da tarefa.

Os nomes das tarefas têm o formato projects/PROJECT_ID/locations/LOCATION_ID/queues/QUEUE_ID/tasks/TASK_ID. A parte TASK_ID do nome da tarefa é equivalente ao campo name da tarefa no Task Queues.

Como reutilizar nomes de tarefas

Você precisa aguardar antes de reutilizar o nome de uma tarefa. O tempo que você precisa aguardar varia conforme a fila que envia a tarefa foi criada no Cloud Tasks ou no Task Queues.

Para tarefas em filas que foram criadas usando o Task Queues (incluindo a fila padrão), você precisa aguardar cerca de nove dias após a tarefa original ter sido excluída ou executada. Para tarefas em filas que foram criadas usando o Cloud Tasks, você precisa aguardar aproximadamente uma hora após a tarefa original ter sido excluída ou executada.

O exemplo a seguir cria uma tarefa com TASK_ID definido como first-try e o adiciona à fila padrão. Este é o equivalente no Cloud Tasks à nomeação de tarefas no Task Queues.

gcloud tasks create-app-engine-task first-try --queue=default \
--method=GET --relative-uri=/url/path

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# queue = 'default'
# task_name = 'first-try'

parent = client.queue_path(project, location, queue)

task = {
    'name': client.task_path(project, location, queue, task_name),
    'app_engine_http_request': {
        'http_method': tasks.HttpMethod.GET,
        'relative_uri': '/url/path'
    }
}
response = client.create_task(parent=parent, task=task)
print(response)

Como executar novamente tarefas com falha

É possível definir a configuração de repetição de tarefas em filas durante a criação ou atualizando a fila. A tabela abaixo lista o campo no Task Queues e o campo correspondente no Cloud Tasks.

Parâmetros de repetição específicos da tarefa

Os parâmetros de repetição específicos da tarefa configurados no Task Queues funcionam no Cloud Tasks. No entanto, não é possível editá-los ou defini-los em novas tarefas. Para alterar os parâmetros de repetição de uma tarefa que tenha parâmetros de repetição específicos, recrie a tarefa com uma fila do Cloud Tasks que tenha os parâmetros de repetição desejados.

O exemplo a seguir demonstra vários cenários de novas tentativas:

• Em fooqueue, as tarefas são repetidas até sete vezes e por até dois dias a partir da primeira tentativa de execução. Depois que ambos os limites são atingidos, elas falham permanentemente.
• Em barqueue, o Google App Engine tenta repetir tarefas, aumentando o intervalo linearmente entre cada nova tentativa até atingir a espera máxima e repetindo indefinidamente no intervalo máximo. Assim, os intervalos entre solicitações são de 10s, 20s, 30s, ..., 190s, 200s, 200s e assim por diante.
• Em bazqueue, o intervalo de repetição começa em 10s, depois duplica três vezes, aumenta linearmente e, finalmente, repete indefinidamente no intervalo máximo. Dessa forma, os intervalos entre solicitações são: 10s, 20s, 40s, 80s, 160s, 240s, 300s, 300s ...

Este é o equivalente do Cloud Tasks à repetição de tarefas no Task Queues.

gcloud tasks queues create fooqueue \
--max-attempts=7 \
--max-retry-duration=172800s  #2*60*60*24 seconds in 2 days

gcloud tasks queues create barqueue \
--min-backoff=10s \
--max-backoff=200s \
--max-doublings=0

gcloud tasks queues create bazqueue \
--min-backoff=10s \
--max-backoff=300s \
--max-doublings=3

from google.protobuf import duration_pb2

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# fooqueue = 'fooqueue'
# barqueue = 'barqueue'
# bazqueue = 'bazqueue'

parent = f"projects/{project}/locations/{location}"

max_retry = duration_pb2.Duration()
max_retry.seconds = 2*60*60*24

foo = {
    'name': client.queue_path(project, location, fooqueue),
    'rate_limits': {
        'max_dispatches_per_second': 1
    },
    'retry_config': {
        'max_attempts': 7,
        'max_retry_duration': max_retry
    }
}

min = duration_pb2.Duration()
min.seconds = 10

max = duration_pb2.Duration()
max.seconds = 200

bar = {
    'name': client.queue_path(project, location, barqueue),
    'rate_limits': {
        'max_dispatches_per_second': 1
    },
    'retry_config': {
        'min_backoff': min,
        'max_backoff': max,
        'max_doublings': 0
    }
}

max.seconds = 300
baz = {
    'name': client.queue_path(project, location, bazqueue),
    'rate_limits': {
        'max_dispatches_per_second': 1
    },
    'retry_config': {
        'min_backoff': min,
        'max_backoff': max,
        'max_doublings': 3
    }
}

queues = [foo, bar, baz]
for queue in queues:
    response = client.create_queue(parent=parent, queue=queue)
    print(response)

Saiba mais na referência do Cloud Tasks Definir parâmetros de repetição.

Como excluir tarefas de uma fila

Ao excluir uma tarefa, aguarde nove dias antes de criá-la com o mesmo nome se ela estava em uma fila criada usando um arquivo queue.yaml ou uma hora, se ela estava em uma fila criada usando o Cloud Tasks.

O exemplo a seguir exclui a tarefa com o ID de tarefa foo da fila com o ID de fila queue1. Esse é o equivalente no Cloud Tasks à exclusão de tarefas no Task Queues.

gcloud tasks delete foo --queue=queue1

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# queue = 'queue1'

task_path = client.task_path(project, location, queue, 'foo')
response = client.delete_task(name=task_path)

Saiba mais na referência do Cloud Tasks Como excluir uma tarefa de uma fila.

Como limpar tarefas

O exemplo a seguir limpa todas as tarefas da fila com o ID da fila queue1. Este é o equivalente no Cloud Tasks à limpeza de tarefas no Task Queues.

gcloud tasks queues purge queue1

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# queue = 'queue1'

queue_path = client.queue_path(project, location, queue)
response = client.purge_queue(name=queue_path)

Saiba mais na referência do Cloud Tasks Como limpar todas as tarefas de uma fila.

A seguir

• Documentação do Cloud Tasks
• Biblioteca de cliente do Cloud Tasks para Python
• Visão geral da Referência REST do Cloud Tasks
• Visão geral da referência de RPC do Cloud Tasks