Как ставить функции в очередь с помощью Cloud Tasks

Функции очереди задач используют Google Cloud Tasks, чтобы помочь вашему приложению выполнять ресурсоемкие, требующие много времени или пропускной способности задачи асинхронно, вне основного потока приложения.

Предположим, вам нужно создать резервные копии большого количества файлов изображений, которые сейчас хранятся в API с ограничением скорости. Чтобы ответственно использовать API, необходимо соблюдать ограничения на количество запросов. Кроме того, такие длительные задачи могут завершаться с ошибкой из-за превышения времени ожидания и ограничений памяти.

Чтобы упростить эту задачу, вы можете написать функцию очереди задач, которая задает основные параметры задачи, такие как scheduleTime и dispatchDeadline, а затем передает функцию в очередь в Cloud Tasks. Среда Cloud Tasks разработана специально для обеспечения эффективного контроля перегрузки и правил повторных попыток для таких операций.

Firebase SDK для Cloud Functions for Firebase версии 3.20.1 и более поздних взаимодействует с Firebase Admin SDK версии 10.2.0 и более поздних, чтобы поддерживать функции очереди задач.

При использовании функций очереди задач с Firebase может взиматься плата за обработку Cloud Tasks. Подробнее о ценах на Cloud Tasks…

Как создать функции очереди задач

Чтобы использовать функции очереди задач, выполните следующие действия:

  1. Напишите функцию очереди задач, используя Firebase SDK для Cloud Functions.
  2. Протестируйте функцию, активировав ее с помощью HTTP-запроса.
  3. Разверните функцию с помощью интерфейса командной строки Firebase. При первом развертывании функции очереди задач интерфейс командной строки создаст очередь задач в Cloud Tasks с параметрами (ограничение частоты запросов и повторные попытки), указанными в исходном коде.
  4. Добавьте задачи в новую очередь, передав параметры для настройки расписания выполнения, если это необходимо. Для этого напишите код с помощью Admin SDK и разверните его в Cloud Functions for Firebase.

Как писать функции очереди задач

Приведенные в этом разделе примеры кода основаны на приложении, которое настраивает сервис для резервного копирования всех изображений из Астрономической фотографии дня НАСА. Для начала импортируйте необходимые модули:

Node.js

// Dependencies for task queue functions.
const {onTaskDispatched} = require("firebase-functions/tasks");
const {onRequest, HttpsError} = require("firebase-functions/https");
const {requiresRole} = require("firebase-functions");
const {getFunctions} = require("firebase-admin/functions");
const {logger} = require("firebase-functions");

// Dependencies for image backup.
const {URL, URLSearchParams} = require("node:url");
const path = require("path");
const {initializeApp} = require("firebase-admin/app");
const {getStorage} = require("firebase-admin/storage");

Python

# Dependencies for task queue functions.
from google.cloud import tasks_v2
import requests
from firebase_functions.options import RetryConfig, RateLimits, SupportedRegion

# Dependencies for image backup.
from datetime import datetime, timedelta
import json
import pathlib
from urllib.parse import urlparse
from firebase_admin import initialize_app, storage, functions
from firebase_functions import https_fn, tasks_fn, params
import google.auth
from google.auth.transport.requests import AuthorizedSession

Для функций очереди задач используйте onTaskDispatched или on_task_dispatched. При написании функции очереди задач можно задать конфигурацию повторных попыток и ограничения скорости для каждой очереди.

Как настроить функции очереди задач

Функции очереди задач имеют множество настроек, позволяющих точно контролировать ограничения скорости и поведение при повторных попытках:

Node.js

exports.backupapod = onTaskDispatched(
    {
      retryConfig: {
        maxAttempts: 5,
        minBackoffSeconds: 60,
      },
      rateLimits: {
        maxConcurrentDispatches: 6,
      },
    }, async (req) => {

Python

@tasks_fn.on_task_dispatched(
    retry_config=RetryConfig(max_attempts=5, min_backoff_seconds=60),
    rate_limits=RateLimits(max_concurrent_dispatches=10),
)
def backupapod(req: tasks_fn.CallableRequest) -> str:
    """Grabs Astronomy Photo of the Day (APOD) using NASA's API."""

  • retryConfig.maxAttempts=5: каждая задача в очереди задач автоматически повторяется до пяти раз. Это помогает устранять временные ошибки, например сбои в сети или временное нарушение работы зависимого внешнего сервиса.

  • retryConfig.minBackoffSeconds=60 – каждая задача повторяется не раньше, чем через 60 секунд после предыдущей попытки. Это обеспечивает большой буфер между каждой попыткой, поэтому мы не спешим исчерпать пять попыток слишком быстро.

  • rateLimits.maxConcurrentDispatch=6: одновременно может быть назначено не более шести задач. Это обеспечивает стабильный поток запросов к базовой функции и помогает сократить количество активных экземпляров и холодных запусков.

Как тестировать функции очереди задач

В большинстве случаев для тестирования функций очереди задач лучше всего подходит эмулятор Cloud Functions. Чтобы узнать, как подготовить приложение к эмуляции функций очереди задач, ознакомьтесь с документацией по Emulator Suite.

Кроме того, функции очереди задач представлены в виде простых HTTP-функций в Firebase Local Emulator Suite. Вы можете протестировать эмулированную функцию задачи, отправив HTTP-запрос POST с полезной нагрузкой JSON:

 # start the Local Emulator Suite
 firebase emulators:start

 # trigger the emulated task queue function
 curl \
  -X POST                                            # An HTTP POST request...
  -H "content-type: application/json" \              # ... with a JSON body
  http://localhost:$PORT/$PROJECT_ID/$REGION/$NAME \ # ... to function url
  -d '{"data": { ... some data .... }}'              # ... with JSON encoded data

Как развернуть функции очереди задач

Чтобы развернуть функцию очереди задач с помощью интерфейса командной строки Firebase, выполните следующие действия:

$ firebase deploy --only functions:backupapod

При первом развертывании функции очереди задач CLI создает очередь задач в Cloud Tasks с параметрами (ограничение частоты запросов и повторные попытки), указанными в исходном коде.

Если при развертывании функций возникают ошибки, связанные с разрешениями, убедитесь, что пользователю, выполняющему команды развертывания, назначены подходящие роли IAM.

Как добавить в очередь функции очереди задач

Функции очереди задач можно ставить в очередь в Cloud Tasks из доверенной серверной среды, например Cloud Functions for Firebase, с помощью Firebase Admin SDK для Node.js или библиотек Google Cloud для Python. Если вы раньше не работали с Admin SDK, начните с добавления Firebase на сервер.

Обычно при создании задачи она добавляется в очередь Cloud Tasks, а затем настраивается:

Node.js

exports.enqueuebackuptasks = onRequest(
    async (_request, response) => {
      const queue = getFunctions().taskQueue("backupapod");

      const enqueues = [];
      for (let i = 0; i <= BACKUP_COUNT; i += 1) {
        const iteration = Math.floor(i / HOURLY_BATCH_SIZE);
        // Delay each batch by N * hour
        const scheduleDelaySeconds = iteration * (60 * 60);

        const backupDate = new Date(BACKUP_START_DATE);
        backupDate.setDate(BACKUP_START_DATE.getDate() + i);
        // Extract just the date portion (YYYY-MM-DD) as string.
        const date = backupDate.toISOString().substring(0, 10);
        enqueues.push(
            queue.enqueue({date}, {
              scheduleDelaySeconds,
              dispatchDeadlineSeconds: 60 * 5, // 5 minutes
            }),
        );
      }
      await Promise.all(enqueues);
      response.sendStatus(200);
    });

Python

@https_fn.on_request()
def enqueuebackuptasks(_: https_fn.Request) -> https_fn.Response:
    """Adds backup tasks to a Cloud Tasks queue."""
    task_queue = functions.task_queue("backupapod")
    target_uri = get_function_url("backupapod")

    for i in range(BACKUP_COUNT):
        batch = i // HOURLY_BATCH_SIZE

        # Delay each batch by N hours
        schedule_delay = timedelta(hours=batch)
        schedule_time = datetime.now() + schedule_delay

        dispatch_deadline_seconds = 60 * 5  # 5 minutes

        backup_date = BACKUP_START_DATE + timedelta(days=i)
        body = {"data": {"date": backup_date.isoformat()[:10]}}
        task_options = functions.TaskOptions(
            schedule_time=schedule_time,
            dispatch_deadline_seconds=dispatch_deadline_seconds,
            uri=target_uri,
        )
        task_queue.enqueue(body, task_options)
    return https_fn.Response(status=200, response=f"Enqueued {BACKUP_COUNT} tasks")

  • В примере кода выполнение задач распределено по времени: для N-й задачи задана задержка в N минут. Это означает, что будет запускаться примерно одна задача в минуту. Обратите внимание, что вы также можете использовать scheduleTime (Node.js) или schedule_time (Python), если хотите, чтобы Cloud Tasks запускал задачу в определенное время.

  • В образце кода задано максимальное время, в течение которого Cloud Tasks будет ждать завершения задачи. Cloud Tasks будет повторять задачу в соответствии с настройками очереди или до истечения срока. В примере очередь настроена на повтор задачи до пяти раз, но задача автоматически отменяется, если весь процесс (включая попытки повтора) занимает более пяти минут.

Устранение неполадок

Узнайте, как диагностировать проблемы с очередями задач с помощью ведения журналов и настроить необходимые разрешения IAM.

Как включить ведение журналов Cloud Tasks

Журналы из Cloud Tasks содержат полезную диагностическую информацию, например статус запроса, связанного с задачей. По умолчанию журналы из Cloud Tasks отключены, поскольку они могут генерировать большой объем данных в вашем проекте. Мы рекомендуем включить журналы отладки, когда вы активно разрабатываете и отлаживаете функции очереди задач. Подробнее о том, как включить ведение журнала…

Права доступа IAM

При постановке задач в очередь или при попытке Cloud Tasks вызвать функции очереди задач могут возникать ошибки PERMISSION DENIED. Убедитесь, что в вашем проекте есть следующие привязки IAM:

gcloud projects add-iam-policy-binding $PROJECT_ID \
  --member=serviceAccount:${PROJECT_ID}@appspot.gserviceaccount.com \
  --role=roles/cloudtasks.enqueuer

  • У аккаунта, используемого для добавления задач в очередь Cloud Tasks, должно быть разрешение на использование сервисного аккаунта, связанного с задачей в Cloud Tasks.

    В примере это сервисный аккаунт по умолчанию App Engine.

Инструкции по добавлению сервисного аккаунта по умолчанию App Engine в качестве пользователя сервисного аккаунта по умолчанию App Engine можно найти в документации по Cloud IAM Google Cloud.

gcloud functions add-iam-policy-binding $FUNCTION_NAME \
  --region=us-central1 \
  --member=serviceAccount:${PROJECT_ID}@appspot.gserviceaccount.com \
  --role=roles/cloudfunctions.invoker