Cloud Storage サブスクリプションを作成する

このドキュメントでは、Cloud Storage サブスクリプションを作成する方法について説明します。Cloud Storage サブスクリプションを作成するには、Google Cloud コンソール、Google Cloud CLI、クライアント ライブラリ、または Pub/Sub API を使用できます。

準備

このドキュメントを読む前に、次の内容をよく理解しておいてください。

• Cloud Storage サブスクリプションの仕組み。
• Cloud Storage の仕組みと、Cloud Storage バケットの作成と管理方法。
• メッセージ エラーを処理するデッドレター トピックを構成する方法。

必要なロールと権限

ロールと権限に関するガイドラインは次の一覧に示すとおりです。

• サブスクリプションを作成するには、プロジェクト レベルでアクセス制御を構成する必要があります。

• このセクションで後述するように、サブスクリプションとトピックが異なるプロジェクトにある場合は、リソースレベルの権限も必要です。

• Cloud Storage サブスクリプションを作成するには、Pub/Sub サービス アカウントに、特定の Cloud Storage バケットへの書き込みとバケットのメタデータの読み取りを行う権限が付与されている必要があります。これらの権限を付与する方法について詳しくは、このドキュメントの次のセクションをご覧ください。

Cloud Storage サブスクリプションの作成に必要な権限を取得するには、管理者にプロジェクトに対する Pub/Sub 編集者（roles/pubsub.editor）IAM ロールを付与するよう依頼してください。ロールの付与については、プロジェクト、フォルダ、組織へのアクセスを管理するをご覧ください。

この事前定義ロールには、Cloud Storage サブスクリプションの作成に必要な権限が含まれています。必要とされる正確な権限については、「必要な権限」セクションを開いてご確認ください。

カスタムロールや他の事前定義ロールを使用して、これらの権限を取得することもできます。

別のプロジェクトのトピックに関連付けられているプロジェクトで Cloud Storage サブスクリプションを作成する必要がある場合は、トピック管理者にトピックに対する Pub/Sub 編集者の (roles/pubsub.editor) IAM ロールも付与するよう依頼してください。

Pub/Sub サービス アカウントに Cloud Storage のロールを割り当てる

一部の Google Cloud サービスには、Google Cloud 管理のサービス アカウントがあり、このサービスを利用してユーザーのリソースにアクセスできます。これらのサービス アカウントは、サービス エージェントとも呼ばれます。Pub/Sub では、service-PROJECT_NUMBER@gcp-sa-pubsub.iam.gserviceaccount.com 形式でのプロジェクトごとにサービス アカウントを作成し、維持します。

Cloud Storage サブスクリプションを作成するには、Pub/Sub サービス アカウントに、特定の Cloud Storage バケットへの書き込みとバケットのメタデータの読み取りを行う権限が付与されている必要があります。次のいずれかの手順のうち １ つを選択します。

• バケット レベルで権限を付与します。特定の Cloud Storage バッケトでは、Storage Object Creator (roles/storage.objectCreator) のロールと Storage Legacy Bucket Reader (roles/storage.legacyBucketReader) のロールを Pub/Sub サービス アカウントに付与します。

• プロジェクト レベルでロールを付与する必要がある場合は、代わりに Cloud Storage バケットを含むプロジェクトでストレージ管理者（roles/storage.admin）のロールを付与できます。Pub/Sub サービス アカウントに このロールを付与します。

Cloud Storage IAM の詳細については、Cloud Storage Identity and Access Management をご覧ください。

Cloud Storage サブスクリプションのプロパティ

Cloud Storage サブスクリプションを構成する場合は、すべてのサブスクリプション タイプに共通のプロパティと、追加の Cloud Storage サブスクリプション固有のプロパティを指定する必要があります。

共通のサブスクリプション プロパティ

すべてのサブスクリプションで設定できる共通のサブスクリプション プロパティについて学習します。

バケット名

Cloud Storage サブスクリプションを作成する前に、Cloud Storage バケットがすでに存在している必要があります。

メッセージはバッチとして送信され、Cloud Storage バケットに保存されます。 単一のバッチまたはファイルは、バケットにオブジェクトとして保存されます。

Cloud Storage バケットでは、リクエスト元による支払いを無効にする必要があります。

Cloud Storage バケットを作成するには、バケットの作成をご覧ください。

ファイル名の接頭辞、接尾辞、日時

Cloud Storage サブスクリプションによって生成された出力 Cloud Storage ファイルは、Cloud Storage バケットにオブジェクトとして保存されます。Cloud Storage バケットに格納されているオブジェクトの名前は、<file-prefix><UTC-date-time>_<uuid><file-suffix> の形式になります。

次のリストに、ファイル形式とカスタマイズ可能なフィールドの詳細を示します。

• <file-prefix> はカスタム ファイル名の接頭辞です。このフィールドは省略できます。

• <UTC-date-time> は、オブジェクトの作成時間に基づいて自動生成されるカスタマイズ可能な文字列です。

• <uuid> は、オブジェクトに対して自動生成されるランダムな文字列です。

• <file-suffix> はカスタム ファイル名の接尾辞です。このフィールドは省略できます。ファイル名の接尾辞を「/」で終わらせることはできません。

• ファイル名の接頭辞と接尾辞を変更できます。

  • たとえば、ファイル名接頭辞の値が prod_ で、ファイル名接尾辞の値が _archive の場合、サンプル オブジェクト名は prod_2023-09-25T04:10:00+00:00_uN1QuE_archive です。

  • ファイル名の接頭辞と接尾辞を指定しない場合、Cloud Storage バケットに保存されるオブジェクト名の形式は <UTC-date-time>_<uuid> になります。

  • Cloud Storage のオブジェクトの命名要件は、ファイル名の接頭辞と接尾辞にも適用されます。詳細については、Cloud Storage オブジェクトについてをご覧ください。

• ファイル名での日時の表示方法を変更できます。

  • 1 回だけ使用できる必須の日時マッチャー: 年（YYYY または YY）、月（MM）、日（DD）、時（hh）、分（mm）、秒（ss）。たとえば、YY-YYYY や MMM は無効です。

  • 1 回だけ使用できる任意のマッチャー: 日時区切り文字（T）とタイムゾーン オフセット（Z または +00:00）。

  • 複数回使用できる省略可能な要素: ハイフン（-）、アンダースコア（_）、コロン（:）、転送スラッシュ（/）です。

  • たとえば、ファイル名の日時形式の値が YYYY-MM-DD/hh_mm_ssZ の場合、サンプル オブジェクト名は prod_2023-09-25/04_10_00Z_uNiQuE_archive です。

  • ファイル名の日時形式が matcher ではない文字で終わる場合、その文字が <UTC-date-time> と <uuid> の間の区切り文字に置き換えられます。たとえば、ファイル名の日時形式の値が YYYY-MM-DDThh_mm_ss- の場合、サンプル オブジェクト名は prod_2023-09-25T04_10_00-uNiQuE_archive です。

ファイルのバッチ処理

Cloud Storage サブスクリプションを使用すると、Cloud Storage バケットにオブジェクトとして保存される新しい出力ファイルを作成するタイミングを決定できます。Pub/Sub は、指定されたバッチ処理条件のいずれかに一致すると、出力ファイルを書き込みます。Cloud Storage のバッチ処理条件は次のとおりです。

• ストレージのバッチ最大時間。この設定は必須です。指定した最大時間の値を超えると、Cloud Storage サブスクリプションが新しい出力ファイルを書き込みます。値を指定しない場合、デフォルト値の 5 分が適用されます。 最大時間に適用可能な値は次のとおりです。

  • 最小値: 1 分
  • デフォルト値 = 5 分
  • 最大値: 10 分。

• ストレージのバッチ最大バイト数。この設定は省略できます。指定した最大バイトの値が超過すると、Cloud Storage サブスクリプションは新しい出力ファイルを書き込みます。最大バイトに使用可能な値は次のとおりです。

  • 最小値 = 1 KB
  • 最大値 = 10 GiB

• ストレージのバッチ最大メッセージ数。この設定は省略できます。指定した最大メッセージ数が超過すると、Cloud Storage サブスクリプションは新しい出力ファイルを書き込みます。最大メッセージに使用可能な値は次のとおりです。

  • 最小値 = 1,000

たとえば、最大時間を 6 分、最大バイトを 2 GB に構成できます。4 分の時点で出力ファイルのファイルサイズが 2 GB に達すると、Pub/Sub は前のファイルをファイナライズして、新しいファイルへの書き込みを開始します。

Cloud Storage サブスクリプションでは、Cloud Storage バケット内の複数のファイルに同時に書き込む場合があります。6 分ごとに新しいファイルを作成するようにサブスクリプションを構成した場合、6 分ごとに作成される複数の Cloud Storage ファイルを目にすることがあります。

場合によっては、ファイルのバッチ処理の条件で構成されているよりも前に、Pub/Sub が新しいファイルへの書き込みを開始することがあります。 サブスクリプションが最大バイト数の値より大きいメッセージを受信した場合、ファイルが最大バイト数の値を超えることもあります。

ファイル形式

Cloud Storage サブスクリプションを作成するときに、Cloud Storage バケットに格納される出力ファイルの形式を テキストまたは Avro として指定できます。

• テキスト: メッセージは書式なしテキストとして保存されます。改行文字は、あるメッセージをファイル内の前のメッセージから区切るものです。メッセージ ペイロードのみが保存され、属性やその他のメタデータは保存されません。

• Avro: メッセージは Apache Avro バイナリ形式で保存されます。[Avro] を選択すると、次の追加プロパティを有効にできます。

  • メタデータを書き込む: このオプションを選択すると、メッセージとともにメッセージのメタデータを保存できます。subscription_name、message_id、publish_time、attributes などのメタデータ フィールドは、出力 Avro オブジェクトの最上位フィールドに書き込まれ、データ以外のすべてのメッセージプロパティ（たとえば、存在する場合 ordering_key）は、attributes マップのエントリとして追加されます。

    メタデータの書き込みが無効になっている場合、出力 Avro オブジェクトにはメッセージ ペイロードのみが書き込まれます。メタデータの書き込みを無効にした出力メッセージの Avro スキーマは次のとおりです。

    {
      "type": "record",
      "namespace": "com.google.pubsub",
      "name": "PubsubMessage",
      "fields": [
        { "name": "data", "type": "bytes" }
      ]
    }
    

    [メタデータの書き込み] を有効にした出力メッセージの Avro スキーマは次のとおりです。

    {
      "type": "record",
      "namespace": "com.google.pubsub",
      "name": "PubsubMessageWithMetadata",
      "fields": [
        { "name": "subscription_name", "type": "string" },
        { "name": "message_id", "type": "string"  },
        { "name": "publish_time", "type": {
            "type": "long",
            "logicalType": "timestamp-micros"
          }
        },
        { "name": "attributes", "type": { "type": "map", "values": "string" } },
        { "name": "data", "type": "bytes" }
      ]
    }
    

  • トピック スキーマを使用する: このオプションを使用すると、Pub/Sub は Avro ファイルの書き込み時に、サブスクリプションが接続されている Pub/Sub トピックのスキーマを使用できます。

    このオプションを使用する場合は、以下の追加要件を確認してください。

    • トピック スキーマは Apache Avro 形式にする必要があります。

    • [トピック スキーマを使用する] と [メタデータを書き込む] の両方が有効になっている場合、トピック スキーマのルートに Record オブジェクトが必要です。Pub/Sub は、レコードのフィールドのリストを拡張してメタデータ フィールドを含めます。そのため、Record には、メタデータ フィールド（subscription_name、message_id、publish_time、attributes）と同じ名前のフィールドを含めることはできません。

Cloud Storage サブスクリプションの作成

namespace pubsub = ::google::cloud::pubsub;
namespace pubsub_admin = ::google::cloud::pubsub_admin;
[](pubsub_admin::SubscriptionAdminClient client,
   std::string const& project_id, std::string const& topic_id,
   std::string const& subscription_id, std::string const& bucket) {
  google::pubsub::v1::Subscription request;
  request.set_name(
      pubsub::Subscription(project_id, subscription_id).FullName());
  request.set_topic(pubsub::Topic(project_id, topic_id).FullName());
  request.mutable_cloud_storage_config()->set_bucket(bucket);
  auto sub = client.CreateSubscription(request);
  if (!sub) {
    if (sub.status().code() == google::cloud::StatusCode::kAlreadyExists) {
      std::cout << "The subscription already exists\n";
      return;
    }
    throw std::move(sub).status();
  }

  std::cout << "The subscription was successfully created: "
            << sub->DebugString() << "\n";
}

import (
	"context"
	"fmt"
	"io"
	"time"

	"cloud.google.com/go/pubsub"
)

// createCloudStorageSubscription creates a Pub/Sub subscription that exports messages to Cloud Storage.
func createCloudStorageSubscription(w io.Writer, projectID, subID string, topic *pubsub.Topic, bucket string) error {
	// projectID := "my-project-id"
	// subID := "my-sub"
	// topic of type https://godoc.org/cloud.google.com/go/pubsub#Topic
	// note bucket should not have the gs:// prefix
	// bucket := "my-bucket"
	ctx := context.Background()
	client, err := pubsub.NewClient(ctx, projectID)
	if err != nil {
		return fmt.Errorf("pubsub.NewClient: %w", err)
	}
	defer client.Close()

	sub, err := client.CreateSubscription(ctx, subID, pubsub.SubscriptionConfig{
		Topic: topic,
		CloudStorageConfig: pubsub.CloudStorageConfig{
			Bucket:         bucket,
			FilenamePrefix: "log_events_",
			FilenameSuffix: ".avro",
			OutputFormat:   &pubsub.CloudStorageOutputFormatAvroConfig{WriteMetadata: true},
			MaxDuration:    1 * time.Minute,
			MaxBytes:       1e8,
		},
	})
	if err != nil {
		return fmt.Errorf("client.CreateSubscription: %w", err)
	}
	fmt.Fprintf(w, "Created Cloud Storage subscription: %v\n", sub)

	return nil
}

import com.google.cloud.pubsub.v1.SubscriptionAdminClient;
import com.google.protobuf.Duration;
import com.google.pubsub.v1.CloudStorageConfig;
import com.google.pubsub.v1.ProjectSubscriptionName;
import com.google.pubsub.v1.ProjectTopicName;
import com.google.pubsub.v1.Subscription;
import java.io.IOException;

public class CreateCloudStorageSubscriptionExample {
  public static void main(String... args) throws Exception {
    // TODO(developer): Replace these variables before running the sample.
    String projectId = "your-project-id";
    String topicId = "your-topic-id";
    String subscriptionId = "your-subscription-id";
    String bucket = "your-bucket";
    String filenamePrefix = "log_events_";
    String filenameSuffix = ".text";
    Duration maxDuration = Duration.newBuilder().setSeconds(300).build();

    createCloudStorageSubscription(
        projectId, topicId, subscriptionId, bucket, filenamePrefix, filenameSuffix, maxDuration);
  }

  public static void createCloudStorageSubscription(
      String projectId,
      String topicId,
      String subscriptionId,
      String bucket,
      String filenamePrefix,
      String filenameSuffix,
      Duration maxDuration)
      throws IOException {
    try (SubscriptionAdminClient subscriptionAdminClient = SubscriptionAdminClient.create()) {

      ProjectTopicName topicName = ProjectTopicName.of(projectId, topicId);
      ProjectSubscriptionName subscriptionName =
          ProjectSubscriptionName.of(projectId, subscriptionId);

      CloudStorageConfig cloudStorageConfig =
          CloudStorageConfig.newBuilder()
              .setBucket(bucket)
              .setFilenamePrefix(filenamePrefix)
              .setFilenameSuffix(filenameSuffix)
              .setMaxDuration(maxDuration)
              .build();

      Subscription subscription =
          subscriptionAdminClient.createSubscription(
              Subscription.newBuilder()
                  .setName(subscriptionName.toString())
                  .setTopic(topicName.toString())
                  .setCloudStorageConfig(cloudStorageConfig)
                  .build());

      System.out.println("Created a CloudStorage subscription: " + subscription.getAllFields());
    }
  }
}

Cloud Storage サブスクリプションをモニタリングする

Cloud Monitoring には、サブスクリプションをモニタリングするための指標がいくつか用意されています。

Pub/Sub に関連する使用可能なすべての指標とその説明については、Pub/Sub のモニタリング ドキュメントをご覧ください。

Pub/Sub 内からサブスクリプションをモニタリングすることもできます。

次のステップ

• Cloud Storage サブスクリプションのトラブルシューティングを行います。

• Cloud Storage について確認する。

• Cloud Storage サブスクリプションを含む Pub/Sub の料金を確認する。