Funktionen mit Cloud Tasks in die Warteschlange stellen

Aufgabenwarteschlangenfunktionen nutzen Google Cloud Tasks, damit Ihre App zeitaufwendige, ressourcenintensive oder bandbreitenbeschränkte Aufgaben asynchron außerhalb des Hauptanwendungsablaufs ausführen kann.

Angenommen, Sie möchten Sicherungen einer großen Anzahl von Bilddateien erstellen, die derzeit auf einer API mit einem Ratenlimit gehostet werden. Um diese API verantwortungsbewusst zu nutzen, müssen Sie die Ratenbegrenzungen einhalten. Außerdem kann es bei solchen Jobs, die lange ausgeführt werden, aufgrund von Zeitüberschreitungen und Arbeitsspeicherlimits zu Fehlern kommen.

Um diese Komplexität zu verringern, können Sie eine Aufgabenwarteschlangenfunktion schreiben, die grundlegende Task-Optionen wie scheduleTime und dispatchDeadline festlegt und die Funktion dann an eine Warteschlange in Cloud Tasks übergibt. Die Cloud Tasks-Umgebung wurde speziell entwickelt, um eine effektive Überlastungskontrolle und Wiederholungsrichtlinien für diese Art von Vorgängen zu gewährleisten.

Das Firebase SDK für Cloud Functions for Firebase v3.20.1 und höher ist mit Firebase Admin SDK v10.2.0 und höher kompatibel, um Aufgabenwarteschlangen-Funktionen zu unterstützen.

Die Verwendung von Aufgabenwarteschlangen-Funktionen mit Firebase kann zu Gebühren für die Cloud Tasks-Verarbeitung führen. Weitere Informationen finden Sie unter Cloud Tasks-Preise.

Funktionen für Aufgabenwarteschlangen erstellen

So verwenden Sie Aufgabenwarteschlangenfunktionen:

  1. Schreiben Sie eine Aufgabenwarteschlangenfunktion mit dem Firebase SDK für Cloud Functions.
  2. Testen Sie Ihre Funktion, indem Sie sie mit einer HTTP-Anfrage auslösen.
  3. Stellen Sie Ihre Funktion mit der Firebase-CLI bereit. Wenn Sie Ihre Aufgabenwarteschlangenfunktion zum ersten Mal bereitstellen, wird mit der CLI eine Aufgabenwarteschlange in Cloud Tasks erstellt, wobei die in Ihrem Quellcode angegebenen Optionen (Ratenbegrenzung und Wiederholung) verwendet werden.
  4. Fügen Sie der neu erstellten Aufgabenwarteschlange Aufgaben hinzu und übergeben Sie bei Bedarf Parameter, um einen Ausführungszeitplan festzulegen. Dazu schreiben Sie den Code mit Admin SDK und stellen ihn in Cloud Functions for Firebase bereit.

Aufgabenwarteschlangenfunktionen schreiben

Die Codebeispiele in diesem Abschnitt basieren auf einer App, die einen Dienst einrichtet, mit dem alle Bilder von Astronomy Picture of the Day der NASA gesichert werden. Importieren Sie zuerst die erforderlichen Module:

Node.js

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

// Dependencies for image backup.
const path = require("path");
const {initializeApp} = require("firebase-admin/app");
const {getStorage} = require("firebase-admin/storage");
const {GoogleAuth} = require("google-auth-library");

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

Verwenden Sie onTaskDispatched oder on_task_dispatched für Aufgabenwarteschlangenfunktionen. Wenn Sie eine Aufgabenwarteschlangenfunktion schreiben, können Sie die Konfiguration für Wiederholungen und Ratenbegrenzung für jede Warteschlange festlegen.

Aufgabenwarteschlangenfunktionen konfigurieren

Aufgabenwarteschlangenfunktionen bieten eine Reihe leistungsstarker Konfigurationseinstellungen, mit denen Sie Ratenbegrenzungen und das Wiederholungsverhalten einer Aufgabenwarteschlange genau steuern können:

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: Jede Aufgabe in der Aufgabenwarteschlange wird automatisch bis zu fünfmal wiederholt. So lassen sich vorübergehende Fehler wie Netzwerkfehler oder vorübergehende Dienstunterbrechungen eines abhängigen, externen Dienstes vermeiden.

  • retryConfig.minBackoffSeconds=60: Jeder Versuch wird mindestens 60 Sekunden nach dem vorherigen Versuch wiederholt. So bleibt zwischen den einzelnen Versuchen ein großer Puffer, damit die fünf Wiederholungsversuche nicht zu schnell aufgebraucht werden.

  • rateLimits.maxConcurrentDispatch=6: Es werden maximal 6 Aufgaben gleichzeitig zugewiesen. So wird ein stetiger Strom von Anfragen an die zugrunde liegende Funktion sichergestellt und die Anzahl der aktiven Instanzen und Kaltstarts verringert.

Aufgabenwarteschlangenfunktionen testen

In den meisten Fällen ist der Cloud Functions-Emulator die beste Möglichkeit, Taskwarteschlangenfunktionen zu testen. In der Emulator Suite-Dokumentation erfahren Sie, wie Sie Ihre App für die Emulation von Aufgabenwarteschlangen-Funktionen instrumentieren.

Außerdem werden Aufgabenwarteschlangenfunktionen als einfache HTTP-Funktionen im Firebase Local Emulator Suite verfügbar gemacht. Sie können eine emulierte Aufgabenfunktion testen, indem Sie eine HTTP-POST-Anfrage mit einer JSON-Datennutzlast senden:

 # 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

Aufgabenwarteschlangen-Funktionen bereitstellen

Aufgabenwarteschlangenfunktion mit der Firebase-CLI bereitstellen:

$ firebase deploy --only functions:backupapod

Wenn Sie eine Aufgabenwarteschlange-Funktion zum ersten Mal bereitstellen, wird mit der CLI eine Aufgabenwarteschlange in Cloud Tasks mit den in Ihrem Quellcode angegebenen Optionen (Ratenbegrenzung und Wiederholung) erstellt.

Wenn beim Bereitstellen von Funktionen Berechtigungsfehler auftreten, prüfen Sie, ob dem Nutzer, der die Bereitstellungsbefehle ausführt, die entsprechenden IAM-Rollen zugewiesen sind.

Funktionen für Aufgabenwarteschlangen in die Warteschlange stellen

Aufgabenwarteschlangenfunktionen können in Cloud Tasks aus einer vertrauenswürdigen Serverumgebung wie Cloud Functions for Firebase mit der Firebase Admin SDK für Node.js oder Google Cloud-Bibliotheken für Python in die Warteschlange gestellt werden. Wenn Sie noch nicht mit Admin SDK vertraut sind, lesen Sie den Abschnitt Firebase einem Server hinzufügen.

Bei einem typischen Ablauf wird eine neue Aufgabe erstellt, in Cloud Tasks in die Warteschlange gestellt und die Konfiguration für die Aufgabe festgelegt:

Node.js

exports.enqueuebackuptasks = onRequest(
    async (_request, response) => {
      const queue = getFunctions().taskQueue("backupapod");
      const targetUri = await getFunctionUrl("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
              uri: targetUri,
            }),
        );
      }
      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")

  • Im Beispielcode wird versucht, die Ausführung von Aufgaben zu verteilen, indem der N-ten Aufgabe eine Verzögerung von N Minuten zugewiesen wird. Das entspricht etwa einer Aufgabe pro Minute. Sie können auch scheduleTime (Node.js) oder schedule_time (Python) verwenden, wenn Sie möchten, dass Cloud Tasks eine Aufgabe zu einem bestimmten Zeitpunkt auslöst.

  • Im Beispielcode wird die maximale Zeit festgelegt, die Cloud Tasks auf den Abschluss einer Aufgabe wartet. Cloud Tasks wird die Aufgabe gemäß der Konfiguration für Wiederholungsversuche der Warteschlange oder bis zum Erreichen dieser Frist wiederholen. Im Beispiel ist die Warteschlange so konfiguriert, dass die Aufgabe bis zu fünfmal wiederholt wird. Die Aufgabe wird jedoch automatisch abgebrochen, wenn der gesamte Prozess (einschließlich der Wiederholungsversuche) länger als fünf Minuten dauert.

Fehlerbehebung

Hier erfahren Sie, wie Sie Probleme mit Aufgabenwarteschlangen mithilfe der Protokollierung diagnostizieren und die erforderlichen IAM-Berechtigungen konfigurieren.

Cloud Tasks-Protokollierung aktivieren

Logs aus Cloud Tasks enthalten nützliche Diagnoseinformationen wie den Status der Anfrage, die mit einer Aufgabe verknüpft ist. Standardmäßig sind Logs von Cloud Tasks deaktiviert, da sie in Ihrem Projekt potenziell ein großes Logvolumen generieren können. Wir empfehlen, die Debugging-Logs zu aktivieren, während Sie Ihre Aufgabenwarteschlangen-Funktionen aktiv entwickeln und debuggen. Logging aktivieren

IAM-Berechtigungen

PERMISSION DENIED-Fehler können auftreten, wenn Sie Aufgaben in die Warteschlange stellen oder wenn Cloud Tasks versucht, Ihre Aufgabenwarteschlangenfunktionen aufzurufen. Prüfen Sie, ob Ihr Projekt die folgenden IAM-Bindungen hat:

  • Die Identität, die zum Einreihen von Aufgaben in die Warteschlange für Cloud Tasks verwendet wird, benötigt die IAM-Berechtigung cloudtasks.tasks.create.

    Im Beispiel ist das das Standarddienstkonto App Engine.

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

  • Die Identität, die zum Einreihen von Aufgaben in Cloud Tasks verwendet wird, benötigt die Berechtigung, das mit einer Aufgabe in Cloud Tasks verknüpfte Dienstkonto zu verwenden.

    Im Beispiel ist das das Standarddienstkonto App Engine.

Eine Anleitung zum Hinzufügen des App Engine-Standarddienstkontos als Nutzer des App Engine-Standarddienstkontos finden Sie in der Google Cloud IAM-Dokumentation.

  • Die Identität, mit der die Aufgabenwarteschlange-Funktion ausgelöst wird, benötigt die Berechtigung cloudfunctions.functions.invoke.

    Im Beispiel ist das das App Engine-Standarddienstkonto.

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