הוספת פונקציות לתור באמצעות Cloud Tasks

פונקציות של תור משימות משתמשות ב-Cloud Tasks של Google כדי לעזור לאפליקציה להריץ משימות אסינכרוניות שדורשות זמן רב, משאבים רבים או רוחב פס מוגבל, מחוץ לתהליך הראשי של האפליקציה.

לדוגמה, נניח שאתם רוצים ליצור גיבויים של קבוצה גדולה של קובצי תמונות שמתארחים כרגע ב-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. כותבים פונקציה של תור משימות באמצעות ה-SDK של Firebase ל-Cloud Functions.
2. כדי לבדוק את הפונקציה, מפעילים אותה באמצעות בקשת HTTP.
3. פורסים את הפונקציה באמצעות ה-CLI של Firebase. בפריסה הראשונה של פונקציית תור המשימות, ה-CLI ייצור תור משימות ב-Cloud Tasks עם האפשרויות (הגבלת קצב של יצירת בקשות וניסיון חוזר) שצוינו בקוד המקור.
4. מוסיפים משימות לתור המשימות החדש שנוצר, ומעבירים פרמטרים כדי להגדיר לוח זמנים לביצוע לפי הצורך. כדי לעשות את זה, אתם יכולים לכתוב את הקוד באמצעות Admin SDK ולפרוס אותו ב-Cloud Functions for Firebase.

כתיבת פונקציות של תור משימות

דוגמאות הקוד בקטע הזה מבוססות על אפליקציה שמגדירה שירות לגיבוי כל התמונות מ-Astronomy Picture of the Day של נאס"א. כדי להתחיל, מייבאים את המודולים הנדרשים:

// 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");
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

משתמשים ב-onTaskDispatched או ב-on_task_dispatched לפונקציות של תור משימות. כשכותבים פונקציה של תור משימות, אפשר להגדיר לכל תור את מספר הניסיונות החוזרים ואת הגבלת הקצב של יצירת הבקשות.

הגדרת פונקציות של תור משימות

פונקציות של תור משימות כוללות קבוצה חזקה של הגדרות אישיות, שמאפשרות לשלוט באופן מדויק בהגבלת קצב של יצירת בקשות ובהתנהגות של ניסיונות חוזרים בתור משימות:

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: כל משימה בתור המשימות מופעלת מחדש באופן אוטומטי עד 5 פעמים. כך אפשר לצמצם שגיאות זמניות כמו שגיאות בחיבור לרשת או שיבושים זמניים בשירות חיצוני תלוי.

• ‫retryConfig.minBackoffSeconds=60: כל משימה מופעלת מחדש לפחות 60 שניות אחרי כל ניסיון. כך יש מרווח גדול בין כל ניסיון, ואנחנו לא מנצלים את 5 הניסיונות מהר מדי.

• ‫rateLimits.maxConcurrentDispatch=6: אפשר לשלוח עד 6 משימות בכל זמן נתון. כך אפשר להבטיח זרם קבוע של בקשות לפונקציה הבסיסית, ולצמצם את מספר המכונות הפעילות וההפעלות במצב התחלתי (cold start).

בדיקת פונקציות של תור משימות

ברוב המקרים, האמולטור של Cloud Functions הוא הדרך הטובה ביותר לבדוק פונקציות של תור משימות. במסמכי התיעוד של חבילת האמולטורים מוסבר איך להגדיר את האפליקציה לאמולציה של פונקציות של תור משימות.

בנוסף, פונקציות של תור משימות נחשפות כפונקציות HTTP פשוטות ב-Firebase Local Emulator Suite. כדי לבדוק פונקציית משימה מדומה, אפשר לשלוח בקשת HTTP POST עם מטען ייעודי (payload) של נתוני 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

פריסת פונקציות של תור משימות

פורסים את הפונקציה של תור המשימות באמצעות ה-CLI של 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 ומגדירים את המשימה:

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);
    });
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

• הקוד לדוגמה מנסה לפרוס את ביצוע המשימות על ידי שיוך של עיכוב של N דקות למשימה ה-N. כלומר, מופעלת בערך משימה אחת לדקה. שימו לב שאפשר גם להשתמש ב-scheduleTime (Node.js) או ב-schedule_time (Python) אם רוצים ש-Cloud Tasks יפעיל משימה בזמן מסוים.

• בקוד לדוגמה הגדרנו את משך הזמן המקסימלי ש-Cloud Tasks ימתין עד להשלמת המשימה. ‫Cloud Tasks ינסה שוב לבצע את המשימה בהתאם להגדרות של ניסיונות חוזרים בתור, או עד שיגיע המועד האחרון. בדוגמה, התור מוגדר לנסות שוב את המשימה עד 5 פעמים, אבל המשימה מבוטלת באופן אוטומטי אם התהליך כולו (כולל ניסיונות חוזרים) נמשך יותר מ-5 דקות.

פתרון בעיות

איך לאבחן בעיות בתורי משימות באמצעות רישום ביומן ולהגדיר את הרשאות ה-IAM הנדרשות.

הפעלת הרישום ביומן של Cloud Tasks

היומנים של Cloud Tasks מכילים מידע שימושי לאבחון, כמו סטטוס הבקשה שמשויכת למשימה. כברירת מחדל, היומנים של Cloud Tasks מושבתים בגלל נפח היומנים הגדול שהם יכולים ליצור בפרויקט. מומלץ להפעיל את יומני ניפוי הבאגים בזמן הפיתוח והניפוי של פונקציות תור המשימות. הפעלת רישום ביומן

הרשאות IAM

יכול להיות שיופיעו שגיאות PERMISSION DENIED כשמוסיפים משימות לתור או כש-Cloud Tasks מנסה להפעיל את הפונקציות של תור המשימות. מוודאים שבפרויקט יש את קישורי ה-IAM הבאים:

• הזהות שמשמשת להוספת משימות לתור ב-Cloud Tasks צריכה את הרשאת ה-IAM‏ cloudtasks.tasks.create.

  בדוגמה, זהו חשבון השירות שמוגדר כברירת מחדל של App Engine.

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.

במאמרי העזרה של Google Cloud IAM מוסבר איך להוסיף את חשבון השירות שמוגדר כברירת מחדל ב-App Engine כמשתמש של חשבון השירות שמוגדר כברירת מחדל ב-App Engine.

• הזהות שמשמשת להפעלת הפונקציה של תור המשימות צריכה את ההרשאה cloudfunctions.functions.invoke.

  בדוגמה, זהו חשבון השירות שמוגדר כברירת מחדל של App Engine

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