Cloud Tasks ile işlevleri kuyruğa alın


Görev kuyruğu işlevleri, uygulamanızın zaman alan, yoğun kaynak kullanan veya bant genişliği sınırlı görevleri ana uygulama akışınızın dışında eşzamansız olarak çalıştırmasına yardımcı olmak için Google Cloud Tasks'tan yararlanır.

Örneğin, şu anda bir API üzerinde barındırılan büyük bir görüntü dosyası kümesinin hız sınırıyla yedeklerini oluşturmak istediğinizi düşünün. Söz konusu API'nin sorumlu bir tüketicisi olabilmek için oran sınırlarına uymanız gerekir. Ayrıca, bu tür uzun süreli işler, zaman aşımları ve bellek sınırlamaları nedeniyle başarısızlığa açık olabilir.

Bu karmaşıklığı azaltmak için, scheduleTime ve dispatchDeadline gibi temel görev seçeneklerini ayarlayan bir görev kuyruğu işlevi yazabilir ve ardından işlevi Bulut Görevleri'ndeki bir kuyruğa devredebilirsiniz. Bulut Görevleri ortamı, bu tür işlemler için etkili tıkanıklık kontrolü ve yeniden deneme politikaları sağlamak üzere özel olarak tasarlanmıştır.

Firebase v3.20.1 ve üzeri Bulut İşlevleri için Firebase SDK'sı, görev kuyruğu işlevlerini desteklemek üzere Firebase Admin SDK v10.2.0 ve üzeri ile birlikte çalışır.

Görev kuyruğu işlevlerinin Firebase ile kullanılması, Bulut Görevlerinin işlenmesi için ücret alınmasına neden olabilir. Daha fazla bilgi için Bulut Görevleri fiyatlandırmasına bakın.

Görev kuyruğu işlevleri oluşturma

Görev kuyruğu işlevlerini kullanmak için şu iş akışını izleyin:

  1. Cloud Functions için Firebase SDK'sını kullanarak bir görev kuyruğu işlevi yazın.
  2. İşlevinizi bir HTTP isteğiyle tetikleyerek test edin.
  3. İşlevinizi Firebase CLI ile dağıtın. Görev kuyruğu işlevinizi ilk kez dağıtırken CLI, Bulut Görevleri'nde kaynak kodunuzda belirtilen seçeneklerle (hız sınırlama ve yeniden deneme) bir görev kuyruğu oluşturacaktır.
  4. Gerekirse bir yürütme zamanlaması oluşturmak için parametreleri ileterek yeni oluşturulan görev kuyruğuna görevler ekleyin. Bunu, Admin SDK'yı kullanarak kodu yazıp bunu Cloud Functions for Firebase'e dağıtarak başarabilirsiniz.

Görev kuyruğu işlevlerini yaz

Bu bölümdeki kod örnekleri, NASA'nın Astronomi Günün Fotoğrafı'ndaki tüm görüntüleri yedekleyen bir hizmet kuran bir uygulamayı temel almaktadır. Başlamak için gerekli modülleri içe aktarın:

Node.js

// 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

Piton

# 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

Görev kuyruğu işlevleri için onTaskDispatched veya on_task_dispatched kullanın. Bir görev kuyruğu işlevi yazarken, kuyruk başına yeniden denemeyi ve hız sınırlayıcı yapılandırmayı ayarlayabilirsiniz.

Görev kuyruğu işlevlerini yapılandırma

Görev kuyruğu işlevleri, hız sınırlarını hassas bir şekilde kontrol etmek ve görev kuyruğunun davranışını yeniden denemek için güçlü bir yapılandırma ayarları seti ile birlikte gelir:

Node.js

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

Piton

@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 : Görev kuyruğundaki her görev otomatik olarak 5 defaya kadar yeniden denenir. Bu, ağ hataları veya bağımlı, harici bir hizmetin geçici hizmet kesintisi gibi geçici hataların azaltılmasına yardımcı olur.

  • retryConfig.minBackoffSeconds=60 : Her görev, her denemeden en az 60 saniye arayla yeniden denenir. Bu, her deneme arasında geniş bir ara bellek sağlar, böylece 5 yeniden deneme girişimini çok hızlı tüketmek için acele etmiyoruz.

  • rateLimits.maxConcurrentDispatch=6 : Belirli bir zamanda en fazla 6 görev gönderilir. Bu, temel işleve istikrarlı bir istek akışının sağlanmasına yardımcı olur ve etkin örneklerin ve soğuk başlatmaların sayısının azaltılmasına yardımcı olur.

Görev kuyruğu işlevlerini test edin

Firebase Local Emulator Suite'teki görev kuyruğu işlevleri, basit HTTP işlevleri olarak sunulur. JSON veri yüküyle bir HTTP POST isteği göndererek öykünülmüş bir görev işlevini test edebilirsiniz:

 # 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

Görev kuyruğu işlevlerini dağıtma

Firebase CLI'yi kullanarak görev kuyruğu işlevini dağıtın:

$ firebase deploy --only functions:backupapod

Bir görev kuyruğu işlevini ilk kez dağıtırken CLI, Bulut Görevleri'nde kaynak kodunuzda belirtilen seçeneklerle (hız sınırlama ve yeniden deneme) bir görev kuyruğu oluşturur.

İşlevleri dağıtırken izin hatalarıyla karşılaşırsanız dağıtım komutlarını çalıştıran kullanıcıya uygun IAM rollerinin atandığından emin olun.

Görev kuyruğu işlevlerini sıraya alma

Görev kuyruğu işlevleri, Node.js için Firebase Yönetici SDK'sı veya Python için Google Cloud kitaplıkları kullanılarak Cloud Functions for Firebase gibi güvenilir bir sunucu ortamından Bulut Görevleri'nde kuyruğa alınabilir. Yönetici SDK'larında yeniyseniz başlamak için Firebase'i sunucuya ekleme konusuna bakın.

Tipik bir akış, yeni bir görev oluşturur, bunu Bulut Görevleri'nde sıraya koyar ve görevin yapılandırmasını ayarlar:

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

Piton

@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

  • Örnek kod, N'inci görev için N'inci dakikalık bir gecikmeyi ilişkilendirerek görevlerin yürütülmesini yaymaya çalışır. Bu ~ 1 görev/dakika tetikleme anlamına gelir. Bulut Görevleri'nin bir görevi belirli bir zamanda tetiklemesini istiyorsanız scheduleTime (Node.js) veya schedule_time (Python) komutlarını da kullanabileceğinizi unutmayın.

  • Örnek kod, Bulut Görevlerinin bir görevin tamamlanması için bekleyeceği maksimum süreyi belirler. Cloud Tasks, kuyruğun yeniden deneme yapılandırmasının ardından veya bu son tarihe ulaşılana kadar görevi yeniden deneyecektir. Örnekte, kuyruk, görevi en fazla 5 kez yeniden deneyecek şekilde yapılandırılmıştır ancak tüm süreç (yeniden deneme girişimleri dahil) 5 dakikadan fazla sürerse görev otomatik olarak iptal edilir.

Hedef URI'yi alın ve ekleyin

Cloud Tasks'ın, temel görev kuyruğu işlevlerine yönelik isteklerin kimliğini doğrulamak için kimlik doğrulama jetonları oluşturma biçimi nedeniyle, görevleri sıraya alırken işlevin Cloud Run URL'sini belirtmeniz gerekir. İşlevinizin URL'sini aşağıda gösterildiği gibi programlı bir şekilde almanızı öneririz:

Node.js

/**
 * 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

Piton

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

Sorun giderme

Bulut Görevleri günlüğünü açın

Bulut Görevlerindeki günlükler, bir görevle ilişkili isteğin durumu gibi yararlı teşhis bilgileri içerir. Bulut Görevleri'ndeki günlükler, projenizde oluşturabileceği büyük miktarda günlük nedeniyle varsayılan olarak kapalıdır. Görev kuyruğu işlevlerinizi aktif olarak geliştirirken ve hata ayıklarken hata ayıklama günlüklerini açmanızı öneririz. Bkz. Günlüğe kaydetmeyi açma .

IAM İzinleri

Görevleri kuyruğa alırken veya Cloud Tasks, görev kuyruğu işlevlerinizi çağırmaya çalıştığında PERMISSION DENIED hatalarını görebilirsiniz. Projenizin aşağıdaki IAM bağlamalarına sahip olduğundan emin olun:

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

  • Görevleri Bulut Görevleri'nde sıraya koymak için kullanılan kimliğin, Bulut Görevleri'ndeki bir görevle ilişkili hizmet hesabını kullanma iznine ihtiyacı vardır.

    Örnekte bu, App Engine varsayılan hizmet hesabıdır .

App Engine varsayılan hizmet hesabının, App Engine varsayılan hizmet hesabının kullanıcısı olarak nasıl ekleneceğine ilişkin talimatlar için Google Cloud IAM belgelerine bakın.

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