הגדרת ניסיונות חוזרים של פונקציות מבוססות-אירועים

במאמר הזה מוסבר איך להגדיר ניסיונות חוזרים לפונקציות Cloud Run מבוססות-אירועים.

כפי שמתואר במאמר ניסיון חוזר של אירועים, אם יעד של הודעת Pub/Sub לא יכול לאשר את קבלת ההודעה, תגובת ברירת המחדל של Pub/Sub היא לשלוח את ההודעה שוב עם השהיה מעריכית לפני ניסיון חוזר (exponential backoff). השהיה מעריכית לפני ניסיון חוזר (exponential backoff) מאפשרת להוסיף עיכובים ארוכים יותר ויותר בין ניסיונות חוזרים. אבל יכול להיות שזה לא ההתנהגות שאתם רוצים בהטמעה הספציפית שלכם.

המאפיין retries לא מיושם בפונקציה עצמה, אלא בגורם המפעיל Eventarc שמפעיל את הפונקציה, מה שמאפשר גמישות רבה יותר. המשמעות היא שביעדים של Cloud Run (כולל פונקציות Cloud Run שנוצרו באמצעות Cloud Run Admin API או Cloud Functions v2 API), אפשר להגדיר ניסיון מסירה יחיד ללא ניסיונות חוזרים. זוהי הגדרת ברירת המחדל כשיוצרים טריגר Eventarc במסוף Google Cloud מהדף Cloud Run. מידע נוסף זמין במאמר בנושא יצירת טריגרים באמצעות Eventarc.

למה פונקציות מבוססות-אירועים לא מצליחות להסתיים

יכול להיות שפונקציה מבוססת-אירועים לא תושלם בהצלחה בגלל שגיאות שמוחזרות בקוד הפונקציה עצמו. הסיבות לכך יכולות להיות:

• הפונקציה מכילה באג וזמן הריצה יוצר חריגה.
• הפונקציה לא יכולה להגיע לנקודת קצה של שירות, או שהזמן הקצוב לתפוגה שלה מסתיים בזמן הניסיון לעשות זאת.
• הפונקציה יוצרת בכוונה חריגה (לדוגמה, כשפרמטר לא עובר אימות).
• פונקציית Node.js מחזירה הבטחה שנדחתה, או מעבירה ערך שאינו null לפונקציית קריאה חוזרת.
• במקרים נדירים, יכול להיות שפונקציה תצא לפני הזמן בגלל שגיאה פנימית, וכברירת מחדל, יכול להיות שהמערכת תנסה להפעיל אותה מחדש באופן אוטומטי או שלא.

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

הגדרת מדיניות הניסיון החוזר

בהתאם לצרכים של פונקציית Cloud Run, יכול להיות שתרצו להגדיר את מדיניות הניסיון החוזר באמצעות מדיניות הניסיון החוזר של מינוי Pub/Sub שמשויכת לטריגר Eventarc. כך תוכלו להגדיר כל שילוב של האפשרויות הבאות:

• לקצר את חלון הניסיון החוזר מ-7 ימים ל-10 דקות בלבד.
• משנים את זמן ההשהיה המינימלי והמקסימלי באסטרטגיית הניסיון החוזר עם השהיה אקספוננציאלית.
• משנים את אסטרטגיית הניסיון החוזר לניסיון חוזר מיידי.
• מגדירים נושא להודעות ללא מוצא.
• הגדרת מספר מקסימלי ומינימלי של ניסיונות מסירה.

כדי להגדיר את מדיניות הניסיון החוזר:

1. כתיבת פונקציית HTTP.
2. משתמשים ב-Pub/Sub API כדי ליצור מינוי ל-Pub/Sub, ומציינים את כתובת ה-URL של הפונקציה כיעד.

במסמכי התיעוד של Eventarc בנושא ניסיון חוזר של אירועים מופיעות המלצות נוספות, כמו יצירת פונקציות מבוססות-אירועים שניתן לנסות שוב להפעיל אותן (אידמפוטנטיות).

במסמכי התיעוד של Pub/Sub בנושא טיפול בכשלים מופיע מידע נוסף על הגדרה ישירה של Pub/Sub.

שיטות מומלצות

בקטע הזה מתוארות שיטות מומלצות לשימוש בניסיונות חוזרים.

שימוש בניסיון חוזר לתיקון שגיאות חולפות

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

הגדרת תנאי סיום כדי למנוע לולאות אינסופיות של ניסיונות חוזרים

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

גישה יעילה היא להשליך אירועים עם חותמות זמן ישנות יותר מזמן מסוים. כך אפשר למנוע ביצועים מוגזמים אם הכשלים נמשכים או ארוכים מהצפוי.

לדוגמה, קטע הקוד הזה מבטל את כל האירועים שהתרחשו לפני יותר מ-10 שניות:

Node.js

const functions = require('@google-cloud/functions-framework');

/**
 * Cloud Event Function that only executes within
 * a certain time period after the triggering event
 *
 * @param {object} event The Cloud Functions event.
 * @param {function} callback The callback function.
 */
functions.cloudEvent('avoidInfiniteRetries', (event, callback) => {
  const eventAge = Date.now() - Date.parse(event.time);
  const eventMaxAge = 10000;

  // Ignore events that are too old
  if (eventAge > eventMaxAge) {
    console.log(`Dropping event ${event} with age ${eventAge} ms.`);
    callback();
    return;
  }

  // Do what the function is supposed to do
  console.log(`Processing event ${event} with age ${eventAge} ms.`);

  // Retry failed function executions
  const failed = false;
  if (failed) {
    callback('some error');
  } else {
    callback();
  }
});

Python

from datetime import datetime, timezone

# The 'python-dateutil' package must be included in requirements.txt.
from dateutil import parser

import functions_framework


@functions_framework.cloud_event
def avoid_infinite_retries(cloud_event):
    """Cloud Event Function that only executes within a certain
    time period after the triggering event.

    Args:
        cloud_event: The cloud event associated with the current trigger
    Returns:
        None; output is written to Cloud Logging
    """
    timestamp = cloud_event["time"]

    event_time = parser.parse(timestamp)
    event_age = (datetime.now(timezone.utc) - event_time).total_seconds()
    event_age_ms = event_age * 1000

    # Ignore events that are too old
    max_age_ms = 10000
    if event_age_ms > max_age_ms:
        print("Dropped {} (age {}ms)".format(cloud_event["id"], event_age_ms))
        return "Timeout"

    # Do what the function is supposed to do
    print("Processed {} (age {}ms)".format(cloud_event["id"], event_age_ms))
    return  # To retry the execution, raise an exception here

Go

// Package tips contains tips for writing Cloud Functions in Go.
package tips

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/GoogleCloudPlatform/functions-framework-go/functions"
	"github.com/cloudevents/sdk-go/v2/event"
)

func init() {
	functions.CloudEvent("FiniteRetryPubSub", FiniteRetryPubSub)
}

// MessagePublishedData contains the full Pub/Sub message
// See the documentation for more details:
// https://cloud.google.com/eventarc/docs/cloudevents#pubsub
type MessagePublishedData struct {
	Message PubSubMessage
}

// PubSubMessage is the payload of a Pub/Sub event.
// See the documentation for more details:
// https://cloud.google.com/pubsub/docs/reference/rest/v1/PubsubMessage
type PubSubMessage struct {
	Data []byte `json:"data"`
}

// FiniteRetryPubSub demonstrates how to avoid inifinite retries.
func FiniteRetryPubSub(ctx context.Context, e event.Event) error {
	var msg MessagePublishedData
	if err := e.DataAs(&msg); err != nil {
		return fmt.Errorf("event.DataAs: %w", err)
	}

	// Ignore events that are too old.
	expiration := e.Time().Add(10 * time.Second)
	if time.Now().After(expiration) {
		log.Printf("event timeout: halting retries for expired event '%q'", e.ID())
		return nil
	}

	// Add your message processing logic.
	return processTheMessage(msg)
}

Java

import com.google.cloud.functions.CloudEventsFunction;
import io.cloudevents.CloudEvent;
import java.time.Duration;
import java.time.ZoneOffset;
import java.time.ZonedDateTime;
import java.util.logging.Logger;

public class RetryTimeout implements CloudEventsFunction {
  private static final Logger logger = Logger.getLogger(RetryTimeout.class.getName());
  private static final long MAX_EVENT_AGE = 10_000;

  /**
   * Cloud Event Function that only executes within
   * a certain time period after the triggering event
   */
  @Override
  public void accept(CloudEvent event) throws Exception {
    ZonedDateTime utcNow = ZonedDateTime.now(ZoneOffset.UTC);
    ZonedDateTime timestamp = event.getTime().atZoneSameInstant(ZoneOffset.UTC);

    long eventAge = Duration.between(timestamp, utcNow).toMillis();

    // Ignore events that are too old
    if (eventAge > MAX_EVENT_AGE) {
      logger.info(String.format("Dropping event with timestamp %s.", timestamp));
      return;
    }

    // Process events that are recent enough
    // To retry this invocation, throw an exception here
    logger.info(String.format("Processing event with timestamp %s.", timestamp));
  }
}

C#‎

using CloudNative.CloudEvents;
using Google.Cloud.Functions.Framework;
using Google.Events.Protobuf.Cloud.PubSub.V1;
using Microsoft.Extensions.Logging;
using System;
using System.Threading;
using System.Threading.Tasks;

namespace TimeBoundedRetries;

public class Function : ICloudEventFunction<MessagePublishedData>
{
    private static readonly TimeSpan MaxEventAge = TimeSpan.FromSeconds(10);
    private readonly ILogger _logger;

    // Note: for additional testability, use an injectable clock abstraction.
    public Function(ILogger<Function> logger) =>
        _logger = logger;

    public Task HandleAsync(CloudEvent cloudEvent, MessagePublishedData data, CancellationToken cancellationToken)
    {
        string textData = data.Message.TextData;

        DateTimeOffset utcNow = DateTimeOffset.UtcNow;

        // Every PubSub CloudEvent will contain a timestamp.
        DateTimeOffset timestamp = cloudEvent.Time.Value;
        DateTimeOffset expiry = timestamp + MaxEventAge;

        // Ignore events that are too old.
        if (utcNow > expiry)
        {
            _logger.LogInformation("Dropping PubSub message '{text}'", textData);
            return Task.CompletedTask;
        }

        // Process events that are recent enough.
        // If this processing throws an exception, the message will be retried until either
        // processing succeeds or the event becomes too old and is dropped by the code above.
        _logger.LogInformation("Processing PubSub message '{text}'", textData);
        return Task.CompletedTask;
    }
}

Ruby

require "functions_framework"

FunctionsFramework.cloud_event "avoid_infinite_retries" do |event|
  # Use the event timestamp to determine the event age.
  event_age_secs = Time.now - event.time.to_time
  event_age_ms = (event_age_secs * 1000).to_i

  max_age_ms = 10_000
  if event_age_ms > max_age_ms
    # Ignore events that are too old.
    logger.info "Dropped #{event.id} (age #{event_age_ms}ms)"

  else
    # Do what the function is supposed to do.
    logger.info "Handling #{event.id} (age #{event_age_ms}ms)..."
    failed = true

    # Raise an exception to signal failure and trigger a retry.
    raise "I failed!" if failed
  end
end

PHP

/**
 * This function shows an example method for avoiding infinite retries in
 * Google Cloud Functions. By default, functions configured to automatically
 * retry execution on failure will be retried indefinitely - causing an
 * infinite loop. To avoid this, we stop retrying executions (by not throwing
 * exceptions) for any events that are older than a predefined threshold.
 */

use Google\CloudFunctions\CloudEvent;

function avoidInfiniteRetries(CloudEvent $event): void
{
    $log = fopen(getenv('LOGGER_OUTPUT') ?: 'php://stderr', 'wb');

    $eventId = $event->getId();

    // The maximum age of events to process.
    $maxAge = 10; // 10 seconds

    // The age of the event being processed.
    $eventAge = time() - strtotime($event->getTime());

    // Ignore events that are too old
    if ($eventAge > $maxAge) {
        fwrite($log, 'Dropping event ' . $eventId . ' with age ' . $eventAge . ' seconds' . PHP_EOL);
        return;
    }

    // Do what the function is supposed to do
    fwrite($log, 'Processing event: ' . $eventId . ' with age ' . $eventAge . ' seconds' . PHP_EOL);

    // infinite_retries failed function executions
    $failed = true;
    if ($failed) {
        throw new Exception('Event ' . $eventId . ' failed; retrying...');
    }
}

הבחנה בין פונקציות שאפשר לנסות שוב לבין שגיאות קריטיות

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

Node.js

const functions = require('@google-cloud/functions-framework');

/**
 * Register a Cloud Event Function that demonstrates
 * how to toggle retries using a promise
 *
 * @param {object} event The Cloud Event for the function trigger.
 */
functions.cloudEvent('retryPromise', cloudEvent => {
  // The Pub/Sub event payload is passed as the CloudEvent's data payload.
  // See the documentation for more details:
  // https://cloud.google.com/eventarc/docs/cloudevents#pubsub
  const base64PubsubMessage = cloudEvent.data.message.data;
  const jsonString = Buffer.from(base64PubsubMessage, 'base64').toString();

  const tryAgain = JSON.parse(jsonString).retry;

  if (tryAgain) {
    throw new Error('Retrying...');
  } else {
    console.error('Not retrying...');
    return Promise.resolve();
  }
});

/**
 * Cloud Event Function that demonstrates
 * how to toggle retries using a callback
 *
 * @param {object} event The Cloud Event for the function trigger.
 * @param {function} callback The callback function.
 */
functions.cloudEvent('retryCallback', (cloudEvent, callback) => {
  // The Pub/Sub event payload is passed as the CloudEvent's data payload.
  // See the documentation for more details:
  // https://cloud.google.com/eventarc/docs/cloudevents#pubsub
  const base64PubsubMessage = cloudEvent.data.message.data;
  const jsonString = Buffer.from(base64PubsubMessage, 'base64').toString();

  const tryAgain = JSON.parse(jsonString).retry;
  const err = new Error('Error!');

  if (tryAgain) {
    console.error('Retrying:', err);
    callback(err);
  } else {
    console.error('Not retrying:', err);
    callback();
  }
});

Python

import base64
import json

import functions_framework
from google.cloud import error_reporting


error_client = error_reporting.Client()


@functions_framework.cloud_event
def retry_or_not(cloud_event):
    """Cloud Event Function that demonstrates how to toggle retries.

    Args:
        cloud_event: The cloud event with a Pub/Sub data payload
    Returns:
        None; output is written to Cloud Logging
    """

    # The Pub/Sub event payload is passed as the CloudEvent's data payload.
    # See the documentation for more details:
    # https://cloud.google.com/eventarc/docs/cloudevents#pubsub
    encoded_pubsub_message = cloud_event.data["message"]["data"]

    # Retry based on a user-defined parameter
    try_again = json.loads(base64.b64decode(encoded_pubsub_message).decode())["retry"]

    try:
        raise RuntimeError("I failed you")
    except RuntimeError:
        error_client.report_exception()
        if try_again:
            raise  # Raise the exception and try again
        else:
            pass  # Swallow the exception and don't retry

Go

package tips

import (
	"context"
	"errors"
	"fmt"
	"log"

	"github.com/GoogleCloudPlatform/functions-framework-go/functions"
	"github.com/cloudevents/sdk-go/v2/event"
)

func init() {
	functions.CloudEvent("RetryPubSub", RetryPubSub)
}

// MessagePublishedData contains the full Pub/Sub message
// See the documentation for more details:
// https://cloud.google.com/eventarc/docs/cloudevents#pubsub
type MessagePublishedData struct {
	Message PubSubMessage
}

// PubSubMessage is the payload of a Pub/Sub event.
// See the documentation for more details:
// https://cloud.google.com/pubsub/docs/reference/rest/v1/PubsubMessage
type PubSubMessage struct {
	Data []byte `json:"data"`
}

// RetryPubSub demonstrates how to toggle using retries.
func RetryPubSub(ctx context.Context, e event.Event) error {
	var msg MessagePublishedData
	if err := e.DataAs(&msg); err != nil {
		return fmt.Errorf("event.DataAs: %w", err)
	}

	name := string(msg.Message.Data)
	if name == "" {
		name = "World"
	}

	// A misconfigured client will stay broken until the function is redeployed.
	client, err := MisconfiguredDataClient()
	if err != nil {
		log.Printf("MisconfiguredDataClient (retry denied):  %v", err)
		// A nil return indicates that the function does not need a retry.
		return nil
	}

	// Runtime error might be resolved with a new attempt.
	if err = FailedWriteOperation(client, name); err != nil {
		log.Printf("FailedWriteOperation (retry expected): %v", err)
		// A non-nil return indicates that a retry is needed.
		return err
	}

	return nil
}

Java

import com.google.cloud.functions.CloudEventsFunction;
import com.google.gson.Gson;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import functions.eventpojos.PubSubBody;
import io.cloudevents.CloudEvent;
import java.nio.charset.StandardCharsets;
import java.util.Base64;
import java.util.logging.Logger;

public class RetryPubSub implements CloudEventsFunction {
  private static final Logger logger = Logger.getLogger(RetryPubSub.class.getName());

  // Use Gson (https://github.com/google/gson) to parse JSON content.
  private static final Gson gson = new Gson();

  @Override
  public void accept(CloudEvent event) throws Exception {
    if (event.getData() == null) {
      logger.warning("No data found in event!");
      return;
    }

    // Extract Cloud Event data and convert to PubSubBody
    String cloudEventData = new String(event.getData().toBytes(), StandardCharsets.UTF_8);
    PubSubBody body = gson.fromJson(cloudEventData, PubSubBody.class);

    String encodedData = body.getMessage().getData();
    String decodedData =
        new String(Base64.getDecoder().decode(encodedData), StandardCharsets.UTF_8);

    // Retrieve and decode PubSubMessage data into a JsonElement.
    // Function is expecting a user-supplied JSON message which determines whether
    // to retry or not.
    JsonElement jsonPubSubMessageElement = gson.fromJson(decodedData, JsonElement.class);

    boolean retry = false;
    // Get the value of the "retry" JSON parameter, if one exists
    if (jsonPubSubMessageElement != null && jsonPubSubMessageElement.isJsonObject()) {
      JsonObject jsonPubSubMessageObject = jsonPubSubMessageElement.getAsJsonObject();

      if (jsonPubSubMessageObject.has("retry")
          && jsonPubSubMessageObject.get("retry").getAsBoolean()) {
        retry = true;
      }
    }

    // Retry if appropriate
    if (retry) {
      // Throwing an exception causes the execution to be retried
      throw new RuntimeException("Retrying...");
    } else {
      logger.info("Not retrying...");
    }
  }
}

C#‎

using CloudNative.CloudEvents;
using Google.Cloud.Functions.Framework;
using Google.Events.Protobuf.Cloud.PubSub.V1;
using Microsoft.Extensions.Logging;
using System;
using System.Text.Json;
using System.Threading;
using System.Threading.Tasks;

namespace Retry;

public class Function : ICloudEventFunction<MessagePublishedData>
{
    private readonly ILogger _logger;

    public Function(ILogger<Function> logger) =>
        _logger = logger;

    public Task HandleAsync(CloudEvent cloudEvent, MessagePublishedData data, CancellationToken cancellationToken)
    {
        bool retry = false;
        string text = data.Message?.TextData;

        // Get the value of the "retry" JSON parameter, if one exists.
        if (!string.IsNullOrEmpty(text))
        {
            JsonElement element = JsonSerializer.Deserialize<JsonElement>(data.Message.TextData);

            retry = element.TryGetProperty("retry", out var property) &&
                property.ValueKind == JsonValueKind.True;
        }

        // Throwing an exception causes the execution to be retried.
        if (retry)
        {
            throw new InvalidOperationException("Retrying...");
        }
        else
        {
            _logger.LogInformation("Not retrying...");
        }
        return Task.CompletedTask;
    }
}

Ruby

require "functions_framework"

FunctionsFramework.cloud_event "retry_or_not" do |event|
  try_again = event.data["retry"]

  begin
    # Simulate a failure
    raise "I failed!"
  rescue RuntimeError => e
    logger.warn "Caught an error: #{e}"
    if try_again
      # Raise an exception to return a 500 and trigger a retry.
      logger.info "Trying again..."
      raise ex
    else
      # Return normally to end processing of this event.
      logger.info "Giving up."
    end
  end
end

PHP

use Google\CloudFunctions\CloudEvent;

function tipsRetry(CloudEvent $event): void
{
    $cloudEventData = $event->getData();
    $pubSubData = $cloudEventData['message']['data'];

    $json = json_decode(base64_decode($pubSubData), true);

    // Determine whether to retry the invocation based on a parameter
    $tryAgain = $json['some_parameter'];

    if ($tryAgain) {
        /**
         * Functions with automatic retries enabled should throw exceptions to
         * indicate intermittent failures that a retry might fix. In this
         * case, a thrown exception will cause the original function
         * invocation to be re-sent.
         */
        throw new Exception('Intermittent failure occurred; retrying...');
    }

    /**
     * If a function with retries enabled encounters a non-retriable
     * failure, it should return *without* throwing an exception.
     */
    $log = fopen(getenv('LOGGER_OUTPUT') ?: 'php://stderr', 'wb');
    fwrite($log, 'Not retrying' . PHP_EOL);
}

השלבים הבאים

• פריסת פונקציה ב-Cloud Run
• יצירת טריגרים מאירועים של Pub/Sub
• יצירת טריגרים מאירועים ב-Cloud Storage
• הפעלת פונקציות מ-Pub/Sub באמצעות Eventarc
• הפעלת פונקציות מ-Cloud Storage באמצעות Eventarc