Habilitar los reintentos de funciones basadas en eventos

En este documento se describe cómo habilitar los reintentos en funciones de Cloud Run basadas en eventos. La función de reintento automático no está disponible para las funciones HTTP.

Por qué no se completan las funciones basadas en eventos

En raras ocasiones, una función puede salir prematuramente debido a un error interno. De forma predeterminada, la función puede volver a intentarse automáticamente o no.

Lo más habitual es que una función basada en eventos no se complete correctamente debido a errores en el propio código de la función. Estos son algunos de los motivos por los que puede ocurrir:

  • La función contiene un error y el tiempo de ejecución genera una excepción.
  • La función no puede acceder a un endpoint de servicio o se agota el tiempo de espera al intentar hacerlo.
  • La función genera una excepción intencionadamente (por ejemplo, cuando un parámetro no supera la validación).
  • Una función de Node.js devuelve una promesa rechazada o pasa un valor que no es null a una retrollamada.

En cualquiera de los casos anteriores, la función dejará de ejecutarse y devolverá un error. Los activadores de eventos que generan los mensajes tienen políticas de reintentos que puedes personalizar para adaptarlas a las necesidades de tu función.

Semántica de reintento

Cloud Run Functions proporciona una ejecución al menos una vez de una función basada en eventos para cada evento emitido por una fuente de eventos. La forma de configurar los reintentos depende de cómo hayas creado la función:

  • Las funciones creadas en la Google Cloud consola o con la API de administrador de Cloud Run requieren que crees y gestiones los activadores de eventos por separado. Los activadores tienen comportamientos de reintento predeterminados que puedes personalizar para adaptarlos a las necesidades de tu función.
  • Las funciones creadas con la API Cloud Functions v2 crearán implícitamente los activadores de eventos necesarios, como temas de Pub/Sub o activadores de Eventarc. De forma predeterminada, los reintentos están inhabilitados para estos activadores y se pueden volver a habilitar mediante la API Cloud Functions v2.

Funciones basadas en eventos creadas con Cloud Run

Las funciones creadas en la Google Cloud consola o con la API Admin de Cloud Run requieren que crees y gestiones los activadores de eventos por separado. Te recomendamos encarecidamente que revises el comportamiento predeterminado de cada tipo de activador:

Funciones basadas en eventos creadas con la API Cloud Functions v2

Las funciones creadas con la API Cloud Functions v2 (por ejemplo, con la CLI de gcloud de Cloud Functions, la API REST o Terraform) crearán y gestionarán activadores de eventos en tu nombre. De forma predeterminada, si una invocación de función finaliza con un error, la función no se vuelve a invocar y el evento se descarta. Si habilitas los reintentos en una función basada en eventos, Cloud Run Functions reintenta una invocación de función fallida hasta que se completa correctamente o hasta que caduca el periodo de reintento.

Si no se habilitan los reintentos de una función (que es el valor predeterminado), la función siempre informará de que se ha ejecutado correctamente y es posible que aparezcan códigos de respuesta 200 OK en sus registros. Esto ocurre aunque la función haya detectado un error. Para que quede claro cuándo se produce un error en tu función, asegúrate de informar de los errores de forma adecuada.

Habilitar o inhabilitar los reintentos

Para habilitar o inhabilitar los reintentos, puedes usar Google Cloud CLI. De forma predeterminada, los reintentos están inhabilitados.

Configurar reintentos desde Google Cloud CLI

Para habilitar los reintentos con la CLI de Google Cloud, incluye la marca --retry al desplegar la función:

gcloud functions deploy FUNCTION_NAME --retry FLAGS...

Para inhabilitar los reintentos, vuelve a desplegar la función sin la marca --retry:

gcloud functions deploy FUNCTION_NAME FLAGS...

Periodo de reintento

Esta ventana de reintento caduca al cabo de 24 horas. Cloud Run Functions vuelve a intentar las funciones basadas en eventos recién creadas mediante una estrategia de espera exponencial, con un tiempo de espera que aumenta entre 10 y 600 segundos.

Prácticas recomendadas

En esta sección se describen las prácticas recomendadas para usar reintentos.

Usar reintentos para gestionar errores transitorios

Como tu función se vuelve a intentar continuamente hasta que se ejecuta correctamente, los errores permanentes, como los errores, deben eliminarse del código mediante pruebas antes de habilitar los reintentos. Los reintentos son la mejor opción para gestionar errores intermitentes o transitorios que tienen una alta probabilidad de resolverse al volver a intentarlo, como un punto final de servicio inestable o un tiempo de espera.

Define una condición de finalización para evitar bucles de reintentos infinitos

Es una práctica recomendada proteger tu función frente a bucles continuos al usar reintentos. Para ello, incluye una condición de finalización bien definida antes de que la función empiece a procesar. Ten en cuenta que esta técnica solo funciona si tu función se inicia correctamente y puede evaluar la condición final.

Un enfoque sencillo pero eficaz es descartar los eventos con marcas de tiempo anteriores a un periodo determinado. De esta forma, se evitan ejecuciones excesivas cuando los errores son persistentes o duran más de lo esperado.

Por ejemplo, este fragmento de código descarta todos los eventos que tengan más de 10 segundos:

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 Stackdriver 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...');
    }
}

Distingue entre funciones que se pueden reintentar y errores críticos

Si tu función tiene habilitados los reintentos, cualquier error no controlado activará un reintento. Asegúrate de que tu código detecte los errores que no deberían provocar un reintento.

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

Hacer que las funciones basadas en eventos que se pueden reintentar sean idempotentes

Las funciones basadas en eventos que se pueden reintentar deben ser idempotentes. A continuación, se incluyen algunas directrices generales para que una función de este tipo sea idempotente:

  • Muchas APIs externas (como Stripe) te permiten proporcionar una clave de idempotencia como parámetro. Si usas una API de este tipo, debes usar el ID de evento como clave de idempotencia.
  • La idempotencia funciona bien con la entrega al menos una vez, ya que permite reintentar la operación de forma segura. Por lo tanto, una práctica recomendada general para escribir código fiable es combinar la idempotencia con los reintentos.
  • Asegúrate de que tu código sea idempotente internamente. Por ejemplo:
    • Asegúrate de que las mutaciones puedan producirse más de una vez sin cambiar el resultado.
    • Consulta el estado de la base de datos en una transacción antes de mutar el estado.
    • Asegúrate de que todos los efectos secundarios sean idempotentes.
  • Imponer una comprobación transaccional fuera de la función, independientemente del código. Por ejemplo, persiste el estado en algún lugar registrando que ya se ha procesado un ID de evento determinado.
  • Gestionar las llamadas de funciones duplicadas fuera de banda. Por ejemplo, puedes tener un proceso de limpieza independiente que se encargue de limpiar después de las llamadas de funciones duplicadas.

Configurar la política de reintentos

En función de las necesidades de tu función de Cloud Run, puede que quieras configurar la política de reintentos directamente. De esta forma, podrás configurar cualquier combinación de las siguientes opciones:

  • Acortar el periodo de reintento de 7 días a tan solo 10 minutos.
  • Cambia el tiempo de espera mínimo y máximo de la estrategia de reintento de espera exponencial.
  • Cambia la estrategia de reintento para que se reintente inmediatamente.
  • Configura un tema de mensajes fallidos.
  • Define un número máximo y mínimo de intentos de entrega.

Para configurar la política de reintentos, sigue estos pasos:

  1. Escribe una función HTTP.
  2. Usa la API de Pub/Sub para crear una suscripción de Pub/Sub y especifica la URL de la función como destino.

Consulta la documentación de Pub/Sub sobre cómo gestionar errores para obtener más información sobre cómo configurar Pub/Sub directamente.

Pasos siguientes