Deduplica le richieste di job

Cloud Scheduler include intestazioni HTTP di identificazione del job e di pianificazione nelle richieste HTTP e App Engine che consentono di deduplicare le esecuzioni ritentate e di rendere idempotenti i gestori di destinazione. Poiché Cloud Scheduler fornisce la distribuzione at-least-once, la deduplicazione delle richieste impedisce al gestore o alle API Google Cloud downstream di elaborare la stessa esecuzione pianificata più di una volta quando si verificano timeout o interruzioni di rete temporanee.

Come funziona la deduplicazione delle richieste

Le richieste di job duplicate per la stessa esecuzione pianificata possono verificarsi nelle seguenti situazioni:

  • Target a lunga esecuzione: il servizio di destinazione impiega più tempo a rispondere rispetto alla scadenza del tentativo del job (attemptDeadline), causando il timeout di Cloud Scheduler e il nuovo tentativo del job mentre il primo tentativo è ancora in esecuzione. Per i target HTTP, il valore predefinito di attemptDeadline è 3 minuti e può essere configurato fino a 30 minuti. Allinea attemptDeadline al timeout della destinazione o elabora il lavoro a lunga esecuzione in modo asincrono se l'esecuzione può superare i 30 minuti.
  • Conferme di rete perse o nuovi tentativi di proxy temporanei: la destinazione riceve ed elabora la richiesta, ma un'interruzione di rete temporanea o una disconnessione del proxy impedisce a Cloud Scheduler di ricevere la risposta HTTP 2xx. Anche se imposti retryCount su 0, la distribuzione di tipo "at-least-once" (almeno una volta) significa che un target può comunque ricevere richieste duplicate durante errori di trasporto temporanei.

Per rendere idempotente il gestore dei job e deduplicare le richieste ritentate, utilizza le intestazioni HTTP che Cloud Scheduler include automaticamente nelle richieste ai target HTTP e App Engine:

Intestazione Descrizione
X-CloudScheduler-JobName L'identificatore breve del job (JOB_ID) da projects/PROJECT_ID/locations/LOCATION_ID/jobs/JOB_ID.
X-CloudScheduler-ScheduleTime

Per i job specificati nel formato unix-cron, l'ora di esecuzione pianificata come offset di UTC formattato in base allo standard RFC 3339, ad esempio 2026-10-16T12:00:00Z o 2026-10-16T05:00:00-07:00. Questo valore rimane costante in tutti i tentativi di ripetizione per la stessa esecuzione pianificata.

Questa intestazione non è garantita per i job che utilizzano il formato di pianificazione groc alternativo, ad esempio every 5 minutes. Se attivi un job manualmente (ad esempio eseguendo gcloud scheduler jobs run o facendo clic su Forza l'esecuzione di un job nella console Google Cloud ), questa intestazione utilizza il timestamp della successiva esecuzione pianificata o potrebbe essere omessa. Di conseguenza, la deduplicazione basata su intestazione salta l'esecuzione pianificata imminente e tutte le esecuzioni manuali aggiuntive attivate prima dell'ora pianificata.

Richiedi l'autenticazione sull'endpoint di destinazione

Le intestazioni X-CloudScheduler-* non sono autenticate e possono essere impostate da qualsiasi chiamante. Non utilizzare queste intestazioni per l'autenticazione. Proteggi invece l'endpoint in base al tipo di target:

Utilizza ricerche di intestazioni senza distinzione tra maiuscole e minuscole

L'uso delle maiuscole nelle intestazioni può variare a seconda delle versioni e dell'infrastruttura HTTP (ad esempio, X-Cloudscheduler-Jobname in HTTP/1.1 o x-cloudscheduler-jobname e x-cloudscheduler-scheduletime in minuscolo negli ambienti HTTP/2 come Cloud Run). Utilizza sempre ricerche di intestazioni senza distinzione tra maiuscole e minuscole nel gestore.

Elimina l'ambiguità degli ID job tra progetti o regioni

Cloud Scheduler non invia automaticamente le intestazioni dell'ID progetto o della regione. Un singolo endpoint del gestore potrebbe ricevere richieste da più progetti o regioni che condividono lo stesso JOB_ID. In questo caso, passa l'ID progetto o la regione in HttpTarget.headers, AppEngineHttpTarget.headers, o i parametri di ricerca o il percorso dell'URL di destinazione e includi questi valori nella chiave di deduplicazione.

Deduplica le richieste in un gestore HTTP

Prima di implementare la deduplicazione delle richieste nel gestore, completa i seguenti prerequisiti di configurazione:

  • Configura la pianificazione del job nel formato unix-cron (ad esempio, */5 * * * * anziché every 5 minutes) in modo che Cloud Scheduler invii l'intestazione X-CloudScheduler-ScheduleTime.
  • Configura un datastore permanente che supporti le transazioni o le scritture condizionali atomiche, come Firestore, Spanner o Memorystore. Configura una policy di durata (TTL) sulla raccolta o sulla tabella (ad esempio sul campo expire_at) per eliminare i record di deduplicazione precedenti dopo la finestra di ripetizione del job (maxRetryDuration). Il TTL di Firestore elimina i record scaduti in modo asincrono, in genere entro 24-72 ore. Per consentire ai nuovi tentativi di recuperare le esecuzioni dalle istanze che hanno subito un arresto anomalo prima della pulizia del TTL, memorizza anche un timestamp di scadenza del lease (ad esempio lease_expires, impostato almeno sul timeout della richiesta della destinazione e su attemptDeadline) e valutalo in una transazione.

Per elaborare ogni esecuzione pianificata una sola volta nel gestore HTTP, completa i seguenti passaggi:

  1. Leggi X-CloudScheduler-JobName e X-CloudScheduler-ScheduleTime dalle intestazioni della richiesta in entrata ed esegui l'hashing in una chiave di deduplicazione di lunghezza fissa (run_id).
  2. Esegui una transazione nel datastore per controllare e rivendicare il record per la chiave di deduplicazione:
    • Se il record esiste e status è completed, restituisci immediatamente un codice di stato HTTP 200 OK (o 204 No Content) senza eseguire la logica del job, in modo che Cloud Scheduler interrompa i nuovi tentativi.
    • Se il record esiste, status è in_progress e il lease (lease_expires) non è scaduto, restituisci un codice di stato di errore ripetibile come 409 Conflict in modo che Cloud Scheduler mantenga attiva la policy di ripetizione nel caso in cui il tentativo in corso non vada a buon fine. Evita di restituire 503 Service Unavailable dai target App Engine, perché App Engine tratta 503 come un errore di sistema e riprova la richiesta senza conteggiarla in retryCount.
    • Se il record non esiste o il suo lease è scaduto, scrivi il record con status impostato su in_progress, un ID tentativo univoco (attempt_id), un timestamp di scadenza del lease (lease_expires) e un timestamp di scadenza del TTL (expire_at).

  3. Dopo aver rivendicato il record, esegui la logica del job. In una transazione, verifica che il valore attempt_id del record corrisponda ancora al tentativo corrente prima di modificare il record:

    • Se la logica del job ha esito positivo, aggiorna status in completed e restituisci un codice di stato HTTP 200 OK.
    • Se la logica del job non va a buon fine con un errore per cui puoi riprovare, elimina il record di deduplicazione prima di restituire un codice di stato 5xx in modo che Cloud Scheduler possa riprovare l'esecuzione senza attendere la scadenza del lease.

Esempio di gestore HTTP

Il seguente esempio Python (per Cloud Run o Cloud Run Functions) utilizza una transazione Firestore con un lease per rivendicare le esecuzioni e deduplicare le richieste di job ritentate.

Questo esempio non è pronto per la produzione. Esaminalo e adattalo alle tue esigenze prima del deployment:

import datetime
import hashlib
import uuid

import flask
import functions_framework
from google.cloud import firestore

# Longer than the work takes, and at least the job's attempt deadline.
LEASE = datetime.timedelta(minutes=5)
# Keep records past the job's retry window. A TTL policy on expire_at deletes them.
RETENTION = datetime.timedelta(days=1)

db = firestore.Client()


def run_job_logic() -> None:
    """Does the scheduled work. Replace the body with your own task."""


@firestore.transactional
def claim_run(
    transaction: firestore.Transaction,
    run_ref: firestore.DocumentReference,
    attempt_id: str,
    now: datetime.datetime,
) -> str:
    """Claims the run unless it's done or another attempt holds the lease."""
    snapshot = run_ref.get(transaction=transaction)
    if snapshot.exists:
        run = snapshot.to_dict()
        if run["status"] == "completed":
            return "completed"
        if run["lease_expires"] > now:
            return "in_progress"
    transaction.set(
        run_ref,
        {
            "status": "in_progress",
            "attempt_id": attempt_id,
            "lease_expires": now + LEASE,
            "expire_at": now + RETENTION,
        },
    )
    return "claimed"


@firestore.transactional
def finish_run(
    transaction: firestore.Transaction,
    run_ref: firestore.DocumentReference,
    attempt_id: str,
    succeeded: bool,
) -> None:
    """Completes or releases the run, if this attempt still holds the claim."""
    snapshot = run_ref.get(transaction=transaction)
    if not snapshot.exists or snapshot.to_dict().get("attempt_id") != attempt_id:
        return  # A retry claimed the run after this attempt's lease expired.
    if succeeded:
        transaction.update(run_ref, {"status": "completed"})
    else:
        transaction.delete(run_ref)


@functions_framework.http
def handle_scheduled_job(request: flask.Request) -> tuple[str, int]:
    """Runs a job's work once per scheduled run, even if the run is delivered twice."""
    job_name = request.headers.get("X-CloudScheduler-JobName")
    schedule_time = request.headers.get("X-CloudScheduler-ScheduleTime")
    if not job_name or not schedule_time:
        return "error: missing Cloud Scheduler headers", 400

    # Retries keep the same schedule time. A forced run uses the next
    # scheduled time, so that scheduled run is then skipped.
    run_id = hashlib.sha256(f"{job_name}/{schedule_time}".encode()).hexdigest()
    run_ref = db.collection("scheduler_runs").document(run_id)
    now = datetime.datetime.now(datetime.timezone.utc)
    attempt_id = uuid.uuid4().hex

    status = claim_run(db.transaction(), run_ref, attempt_id, now)
    if status == "completed":
        return "Duplicate run skipped", 200
    if status == "in_progress":
        # A non-2xx response keeps retries going in case that attempt fails.
        return "Run in progress", 409

    try:
        run_job_logic()
    except Exception:
        # Releases the claim so that a retry can process the run.
        finish_run(db.transaction(), run_ref, attempt_id, succeeded=False)
        raise
    finish_run(db.transaction(), run_ref, attempt_id, succeeded=True)
    return "Run processed", 200

Evita richieste duplicate alle destinazioni API Google Cloud

Quando Cloud Scheduler richiama un endpoint API Google Cloud (*.googleapis.com) direttamente tramite HTTP, l'API di destinazione ignora le intestazioni X-CloudScheduler-JobName e X-CloudScheduler-ScheduleTime. Ad esempio, la chiamata di workflowexecutions.googleapis.com per avviare un'esecuzione di Workflows, di run.googleapis.com per eseguire un job Cloud Run o di cloudbuild.googleapis.com per eseguire un trigger di Cloud Build avvia una nuova esecuzione a ogni tentativo di ripetizione.

Per l'affidabilità della produzione, applica l'idempotenza nel servizio più a valle che modifica lo stato. La deduplicazione delle richieste solo a un livello upstream restringe la finestra di nuovi tentativi, ma se una richiesta downstream va a buon fine e la risposta HTTP viene persa sulla rete, il chiamante upstream non può determinare se l'operazione è stata completata e potrebbe comunque riprovare la richiesta.

Per evitare esecuzioni duplicate durante la pianificazione di target API Google Cloud , utilizza i seguenti approcci insieme per l'idempotenza end-to-end:

  • Instrada le richieste tramite un proxy idempotente: anziché chiamare l'endpoint *.googleapis.com direttamente da Cloud Scheduler, esegui il deployment del tuo servizio Cloud Run o del proxy Cloud Run Functions e configura Cloud Scheduler per richiamare quel proxy. Nel proxy, implementa il pattern di rivendicazione transazionale atomica di Python e Firestore descritto in questo documento, in Deduplica le richieste in un gestore HTTP.

    Il proxy legge X-CloudScheduler-JobName e X-CloudScheduler-ScheduleTime, deduplica le richieste ripetute e passa la chiave di deduplicazione come token di idempotenza nel payload della richiesta (ad esempio nel campo argument quando avvia un'esecuzione di Workflows).

    Richiedi l'autenticazione sul proxy (ad esempio, esegui il deployment con --no-allow-unauthenticated) e configura il job Cloud Scheduler per eseguire l'autenticazione con un token OIDC. Esegui il proxy in un account di servizio dedicato con solo i ruoli Identity and Access Management (IAM) minimi richiesti (ad esempio roles/workflows.invoker nel workflow di destinazione e roles/datastore.user per i record di deduplicazione).

  • Applica l'idempotenza nei servizi downstream: utilizza il token di idempotenza per eseguire un controllo atomico o transazionale in un datastore come Firestore prima di eseguire i passaggi di modifica dello stato. Passa il token a qualsiasi endpoint HTTP downstream o Cloud Run Functions richiamato dal tuo servizio, in modo che anche questi target possano deduplicare i nuovi tentativi. Per un esempio end-to-end che utilizza Workflows, le transazioni Firestore e Cloud Run Functions, consulta il blog post di Google Cloud, Effettuare chiamate API esattamente una volta quando utilizzi Workflows.