イベント ドリブン関数の再試行

このドキュメントでは、イベント ドリブン関数(バックグラウンド関数CloudEvent 関数)の再試行を有効にする方法について説明します。自動再試行は HTTP 関数では使用できません。

再試行のセマンティクス

Cloud Functions では、イベントソースによって生成された各イベントにつき、最低 1 回はイベント ドリブン関数が必ず実行されることになっています。ただし、デフォルトでは、関数の呼び出しがエラーによって終了した場合、その関数が再度読み出されることはなく、そのイベントはドロップします。イベント ドリブン関数の再試行を有効にすると、失敗した関数の呼び出しが正常に完了するか、再試行期間(デフォルトは 7 日間)が終了するまで再試行が行われます。

イベント ドリブン関数が完了しない理由

まれに、内部エラーのため関数が早期に終了することがあります。関数は、デフォルトで自動的に再試行される場合もあれば、再試行されない場合もあります。

よくあるのは、関数コード内でエラーがスローされるためにイベント ドリブン関数が正常に完了できないケースです。これが起こる理由としては、次のようなものがあります。

  • ファンクションにバグがあるため、ランタイムから例外がスローされる。
  • 関数がサービス エンドポイントに到達できないか、エンドポイントに到達しようとしている間にタイムアウトする。
  • 関数から意図的に例外がスローされる(たとえば、パラメータの検証に失敗した場合)。
  • Node.js に記述されている関数が、拒否された Promise を返すか、コールバックに null 以外の値を渡す場合。

上記のいずれの場合も、関数はデフォルトで実行を停止し、イベントが破棄されます。エラーが発生したときに関数を再試行する場合は、「失敗時に再試行する」プロパティを設定して、デフォルトの再試行ポリシーを変更できます。これにより、関数が正常に完了するまで数日間イベントが繰り返し再試行されます。

再試行の有効化と無効化

再試行を有効または無効にするには、gcloud コマンドライン ツールまたは Cloud Console を使用します。デフォルトでは、再試行は無効になっています。

gcloud コマンドライン ツールを使用する

gcloud コマンドライン ツールを使用して再試行を有効にするには、関数をデプロイするときに、--retry フラグを指定します。

gcloud functions deploy FUNCTION_NAME --retry FLAGS...

再試行を無効にするには、--retry フラグを指定せずに関数を再デプロイします。

gcloud functions deploy FUNCTION_NAME FLAGS...

Cloud Console を使用する

Cloud Console で再試行を有効または無効にする手順は次のとおりです。

  1. Cloud Platform Console の Cloud Functions の概要ページに移動します。

  2. [関数を作成] をクリックします。または、既存の関数をクリックしてその詳細ページに移動し、[編集] をクリックします。

  3. 関数の必須フィールドを入力します。

  4. [トリガー] フィールドがイベントベースのトリガータイプ(Cloud Pub/Sub や Cloud Storage など)に設定されていることを確認します。

  5. [詳細] をクリックして詳細設定を表示します。

  6. [失敗時に再試行する] というラベルが付いたチェックボックスをオンまたはオフにします。

ベスト プラクティス

ここでは、再試行の使用に関するベスト プラクティスを説明します。

再試行を使用して一時的なエラーを処理する

関数は成功するまで継続的に再試行されるため、テストを通じてバグなどの永続的なエラーをコードから除去してから、再試行を有効にしてください。再試行によって解決される可能性が高い断続的なエラーや一時的なエラー(サービス エンドポイントの不安定さやタイムアウトなど)を処理するには、再試行が最適です。

無限再試行ループを避けるための終了条件の設定

再試行を使用する場合は、関数が連続ループに陥らないように保護することをおすすめします。そのためには、明確に定義された終了条件を含めてから関数の処理を開始します。この手法が機能するのは、関数が正常に開始し、終了条件を評価できる場合のみです。

簡単で効果的なアプローチは、特定の時間よりも古いタイムスタンプを持つイベントを破棄することです。これにより、エラーが持続的である場合や継続時間が予想よりも長い場合に、実行時間が過度に長くなるのを回避できます。

たとえば、次のコード スニペットは 10 秒を超えるすべてのイベントを破棄します。

Node.js

/**
 * Background Cloud 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.
 */
exports.avoidInfiniteRetries = (event, callback) => {
  const eventAge = Date.now() - Date.parse(event.timestamp);
  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

def avoid_infinite_retries(data, context):
    """Background Cloud Function that only executes within a certain
    time period after the triggering event.

    Args:
        data (dict): The event payload.
        context (google.cloud.functions.Context): The event metadata.
    Returns:
        None; output is written to Stackdriver Logging
    """

    timestamp = context.timestamp

    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(context.event_id, event_age_ms))
        return 'Timeout'

    # Do what the function is supposed to do
    print('Processed {} (age {}ms)'.format(context.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"

	"cloud.google.com/go/functions/metadata"
)

// 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, m PubSubMessage) error {
	meta, err := metadata.FromContext(ctx)
	if err != nil {
		// Assume an error on the function invoker and try again.
		return fmt.Errorf("metadata.FromContext: %v", err)
	}

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

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

Java


import com.google.cloud.functions.BackgroundFunction;
import com.google.cloud.functions.Context;
import com.google.events.cloud.pubsub.v1.Message;
import com.google.gson.Gson;
import java.time.Duration;
import java.time.ZoneOffset;
import java.time.ZonedDateTime;
import java.util.logging.Logger;

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

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

  /**
   * Background Cloud Function that only executes within
   * a certain time period after the triggering event
   */
  @Override
  public void accept(Message message, Context context) {
    ZonedDateTime utcNow = ZonedDateTime.now(ZoneOffset.UTC);
    ZonedDateTime timestamp = ZonedDateTime.parse(context.timestamp());

    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;

            DateTime utcNow = DateTime.UtcNow;

            // Every PubSub CloudEvent will contain a timestamp.
            DateTime timestamp = cloudEvent.Time.Value.ToUniversalTime();
            DateTime 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

/**
 * Background Cloud Function that demonstrates
 * how to toggle retries using a promise
 *
 * @param {object} event The Cloud Functions event.
 * @param {object} event.data Data included with the event.
 * @param {object} event.data.retry User-supplied parameter that tells the function whether to retry.
 */
exports.retryPromise = event => {
  const tryAgain = !!event.data.retry;

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

/**
 * Background Cloud Function that demonstrates
 * how to toggle retries using a callback
 *
 * @param {object} event The Cloud Functions event.
 * @param {object} event.data Data included with the event.
 * @param {object} event.data.retry User-supplied parameter that tells the function whether to retry.
 * @param {function} callback The callback function.
 */
exports.retryCallback = (event, callback) => {
  const tryAgain = !!event.data.retry;
  const err = new Error('Error!');

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

Python

from google.cloud import error_reporting
error_client = error_reporting.Client()

def retry_or_not(data, context):
    """Background Cloud Function that demonstrates how to toggle retries.

    Args:
        data (dict): The event payload.
        context (google.cloud.functions.Context): The event metadata.
    Returns:
        None; output is written to Stackdriver Logging
    """

    # Retry based on a user-defined parameter
    try_again = data.data.get('retry') is not None

    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 contains tips for writing Cloud Functions in Go.
package tips

import (
	"context"
	"errors"
	"log"
)

// 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, m PubSubMessage) error {
	name := string(m.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.BackgroundFunction;
import com.google.cloud.functions.Context;
import com.google.events.cloud.pubsub.v1.Message;
import com.google.gson.Gson;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import java.nio.charset.StandardCharsets;
import java.util.Base64;
import java.util.logging.Logger;

public class RetryPubSub implements BackgroundFunction<Message> {
  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(Message message, Context context) {
    String bodyJson = new String(
        Base64.getDecoder().decode(message.getData()), StandardCharsets.UTF_8);
    JsonElement bodyElement = gson.fromJson(bodyJson, JsonElement.class);

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

      if (body.has("retry") && body.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);
}

再試行可能なイベント ドリブン関数をべき等にする

再試行可能なイベント ドリブン関数は、べき等にする必要があります。このような関数をべき等化するための一般的なガイドラインを次に示します。

  • 多くの外部 API(Stripe など)では、べき等のキーをパラメータとして指定できます。このような API を使用している場合は、イベント ID をべき等のキーとして使用します。
  • べき等では再試行が安全に行われるため、at-least-once 配信でうまく機能します。したがって、信頼性の高いコードを書くための一般的なベスト プラクティスは、べき等と再試行を組み合わせることです。
  • コードが内部でべき等であることを確認します。次に例を示します。
    • 結果が変わらずにミューテーションが 2 回以上起こることを確認する。
    • 状態を変更する前にトランザクション内のデータベース状態を照会する。
    • すべての副作用がそれ自体べき等であることを確認する。
  • コードとは関係なく、トランザクション チェックを関数の外側に置く。たとえば、指定されたイベント ID がすでに処理されたことを記録している場所の状態を保持します。
  • 重複した関数呼び出しを帯域外で処理する。たとえば、重複した関数呼び出しの後にクリーンアップする別のクリーンアップ プロセスを用意します。

次のステップ