Kolejkowanie funkcji za pomocą Cloud Tasks

Funkcje kolejki zadań korzystają z usług Google Cloud Tasks, aby umożliwić aplikacji asynchroniczne wykonywanie czasochłonnych, wymagających dużej ilości zasobów lub ograniczonych pod względem przepustowości zadań poza głównym przepływem aplikacji.

Załóżmy, że chcesz utworzyć kopie zapasowe dużej ilości plików z obrazami, które są obecnie hostowane w interfejsie API z ograniczeniem szybkości. Aby być odpowiedzialnym użytkownikiem tego interfejsu API, musisz przestrzegać jego limitów szybkości. Poza tym zadania tego typu mogą być podatne na błędy z powodu limitów czasu i pamięci.

Aby uprościć ten proces, możesz napisać funkcję kolejki zadań, która ustawia podstawowe opcje zadań, takie jak scheduleTime i dispatchDeadline, a następnie przekazuje funkcję do kolejki w Cloud Tasks. Środowisko Cloud Tasks zostało zaprojektowane pod kątem skutecznego kontrolowania natężenia ruchu i zapewniania zasad ponownych prób w przypadku tego rodzaju operacji.

Pakiet SDK Firebase na Cloud Functions for Firebase w wersji 3.20.1 lub nowszej współpracuje z Firebase Admin SDK w wersji 10.2.0 lub nowszej, aby obsługiwać funkcje kolejki zadań.

Korzystanie z funkcji kolejki zadań w Firebase może spowodować obciążenia za przetwarzanieCloud Tasks. Więcej informacji znajdziesz w cenniku Cloud Tasks.

Przykłady kodu w tej sekcji są oparte na aplikacji, która konfiguruje usługę tworzącą kopię zapasową wszystkich obrazów z Astronomy Picture of the Day NASA. Aby rozpocząć, zaimportuj wymagane moduły:

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

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

# 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
main.py

W przypadku funkcji kolejki zadań używaj wartości onTaskDispatched lub on_task_dispatched. Podczas pisania funkcji kolejki zadań możesz skonfigurować ponowne próby i ograniczenie szybkości na poziomie kolejki.

Konfigurowanie funkcji kolejki zadań

Funkcje kolejki zadań są wyposażone w zaawansowany zestaw ustawień konfiguracji, który umożliwia dokładne kontrolowanie limitów szybkości i zachowania funkcji ponownego próbowania kolejki zadań:

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

@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."""
main.py

• retryConfig.maxAttempts=5: każde zadanie w kolejce zadań jest automatycznie powtarzane do 5 razy. Pomaga to ograniczyć przejściowe błędy, takie jak błędy sieciowe lub tymczasowe przerwy w działaniu zależnej usługi zewnętrznej.

• retryConfig.minBackoffSeconds=60: każde zadanie jest ponownie wykonywane co co najmniej 60 sekund od poprzedniej próby. Zapewnia to duży bufor między próbami, dzięki czemu nie wyczerpujemy zbyt szybko 5 prób.

• rateLimits.maxConcurrentDispatch=6: w danym momencie wysyłanych jest maksymalnie 6 zadań. Pomaga to zapewnić stały napływ żądań do funkcji podstawowej i ogranicza liczbę aktywnych instancji oraz uruchomień „na zimno”.

Testowanie funkcji kolejki zadań

W większości przypadków najlepszym sposobem na przetestowanie funkcji kolejki zadań jest użycie emulatora Cloud Functions. W dokumentacji pakietu emulatorów dowiesz się, jak przetestować aplikację pod kątem emulacji funkcji kolejki zadań.

Dodatkowo funkcje kolejki zadań functions_sdk są udostępniane jako proste funkcje HTTP w Firebase Local Emulator Suite. Aby przetestować emulowaną funkcję zadania, wyślij żądanie HTTP POST z ładunkiem danych 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

Wdrażanie funkcji kolejki zadań

Wdróż funkcję kolejki zadań za pomocą interfejsu wiersza poleceń Firebase:

$ firebase deploy --only functions:backupapod

Podczas pierwszego wdrażania funkcji kolejki zadań CLI tworzy kolejkę zadań w Cloud Tasks z opcjami (ograniczeniem szybkości i ponowną próbą) określonymi w kodzie źródłowym.

Jeśli podczas wdrażania funkcji wystąpią błędy uprawnień, sprawdź, czy użytkownik wykonujący polecenia wdrażania ma przypisane odpowiednie role uprawnień IAM.

Funkcja kolejkowania zadań

Funkcje kolejki zadań można umieszczać w kolei Cloud Tasks z zaufanego środowiska serwera, takiego jak Cloud Functions for Firebase, za pomocą funkcji Firebase Admin SDK w Node.js lub bibliotek Google Cloud w Pythonie. Jeśli dopiero zaczynasz korzystać z Admin SDK, zapoznaj się z artykułem Dodawanie Firebase do serwera.

Typowy przepływ pracy tworzy nowe zadanie, umieszcza je w kolejce w Cloud Tasks i ustawia jego konfigurację:

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);
    });
index.js

@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")
main.py

• Przykładowy kod stara się rozłożyć wykonywanie zadań w czasie, przypisując opóźnienie N minut do N-tego zadania. Oznacza to, że zadanie jest wykonywane mniej więcej raz na minutę. Jeśli chcesz, aby zadanie Cloud Tasks zostało uruchomione w określonym czasie, możesz użyć funkcji scheduleTime (Node.js) lub schedule_time (Python).

• Przykładowy kod określa maksymalny czas oczekiwania funkcji Cloud Tasks na wykonanie zadania. Cloud Tasks spróbuje ponownie wykonać zadanie zgodnie z konfiguracją kolejki lub do czasu upływu tego terminu. W tym przykładzie kolejka jest skonfigurowana tak, aby próbować wykonać zadanie maksymalnie 5 razy, ale zadanie jest automatycznie anulowane, jeśli cały proces (w tym próby ponownego wykonania) trwa dłużej niż 5 minut.

Pobierz i uwzględnij docelowy identyfikator URI

Ze względu na sposób, w jaki Cloud Tasks tworzy tokeny uwierzytelniania, aby uwierzytelniać żądania kierowane do funkcji kolejki zadań, musisz podać adres URL funkcji Cloud Run podczas kolejkowania zadań. Zalecamy pobranie adresu URL funkcji za pomocą kodu, jak pokazano poniżej:

/**
 * Get the URL of a given v2 cloud function.
 *
 * @param {string} name the function's name
 * @param {string} location the function's location
 * @return {Promise<string>} The URL of the function
 */
async function getFunctionUrl(name, location="us-central1") {
  if (!auth) {
    auth = new GoogleAuth({
      scopes: "https://www.googleapis.com/auth/cloud-platform",
    });
  }
  const projectId = await auth.getProjectId();
  const url = "https://cloudfunctions.googleapis.com/v2beta/" +
    `projects/${projectId}/locations/${location}/functions/${name}`;

  const client = await auth.getClient();
  const res = await client.request({url});
  const uri = res.data?.serviceConfig?.uri;
  if (!uri) {
    throw new Error(`Unable to retreive uri for function at ${url}`);
  }
  return uri;
}
index.js

def get_function_url(name: str, location: str = SupportedRegion.US_CENTRAL1) -> str:
    """Get the URL of a given v2 cloud function.

    Params:
        name: the function's name
        location: the function's location

    Returns: The URL of the function
    """
    credentials, project_id = google.auth.default(
        scopes=["https://www.googleapis.com/auth/cloud-platform"])
    authed_session = AuthorizedSession(credentials)
    url = ("https://cloudfunctions.googleapis.com/v2beta/" +
           f"projects/{project_id}/locations/{location}/functions/{name}")
    response = authed_session.get(url)
    data = response.json()
    function_url = data["serviceConfig"]["uri"]
    return function_url
main.py

Rozwiązywanie problemów

„Zadania w chmurze” > Włącz logowanie Cloud Tasks

Logi z Cloud Tasks zawierają przydatne informacje diagnostyczne, takie jak stan żądania powiązanego z zadaniem. Domyślnie logi z Cloud Tasks są wyłączone ze względu na dużą liczbę logów, które mogą potencjalnie wygenerować w Twoim projekcie. Zalecamy włączenie dzienników debugowania podczas aktywnego tworzenia i debugowania funkcji kolejki zadań. Zobacz Włączanie logowania.

Uprawnienia

Podczas kolejkowania zadań lub wywoływania funkcji kolejki zadań przez Cloud Tasks mogą wystąpić błędy PERMISSION DENIED. Upewnij się, że Twój projekt ma te atrybuty uprawnień IAM:

• Tożsamość używana do kolejkowania zadań do usługi Cloud Tasks wymaga uprawnień cloudtasks.tasks.create w IAM.

  W tym przykładzie jest to App Engine domyślne konto usługi.

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

• Tożsamość używana do kolejkowania zadań do usługi Cloud Tasks musi mieć uprawnienia do korzystania z konta usługi powiązanego z zadaniem w usłudze Cloud Tasks.

  W tym przykładzie jest to App Engine domyślne konto usługi.

Więcej informacji o dodawaniu domyślnego konta usługi App Engine jako użytkownika domyślnego konta usługi App Engine znajdziesz w dokumentacji Google Cloud IAM.

• Tożsamość użyta do wywołania funkcji kolejki zadań musi mieć uprawnienia cloudfunctions.functions.invoke.

  W tym przykładzie jest to App Engine domyślne konto usługi.

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