push サブスクリプションの使用

Pub/Sub は push と pull の両方のメッセージ配信をサポートします。pull サブスクリプションと push サブスクリプションの概要と比較については、サブスクライバーの概要をご覧ください。このドキュメントでは、push 配信について説明します。pull 配信の説明については、pull サブスクライバー ガイドをご覧ください。

サブスクリプションが push 配信を使用する場合、Pub/Sub サービスが push エンドポイントにメッセージを配信します。push エンドポイントは、一般公開された HTTPS アドレスである必要があります。push エンドポイントのサーバーには、認証局が署名した有効な SSL 証明書が必要です。

Pub/Sub サービスでは、Pub/Sub サービスがメッセージを格納するのと同じ Google Cloud リージョンから push エンドポイントにメッセージを配信します。Pub/Sub サービスでは、同じ Google Cloud リージョンからベストエフォート方式でメッセージを配信します。

さらに、エンドポイントでリクエストを認証する際に承認ヘッダーを使用するように push サブスクリプションを構成できます。サブスクリプションと同じプロジェクトでホストされる App Engine Standard エンドポイントと Cloud Functions エンドポイントでは、自動認証と自動承認のメカニズムを使用できます。

メッセージの受信

Pub/Sub が push エンドポイントにメッセージを配信すると、Pub/Sub は POST リクエストの本文にメッセージを送信します。リクエストの本文は JSON オブジェクトであり、メッセージ データは message.data フィールドにあります。メッセージ データは base64 でエンコードされています。

push エンドポイントへの POST リクエストの本文の例を次に示します。

{
    "message": {
        "attributes": {
            "key": "value"
        },
        "data": "SGVsbG8gQ2xvdWQgUHViL1N1YiEgSGVyZSBpcyBteSBtZXNzYWdlIQ==",
        "messageId": "136969346945"
    },
   "subscription": "projects/myproject/subscriptions/mysubscription"
}

push サブスクリプションからメッセージを受信するには、Webhook を使用して、Pub/Sub によって push エンドポイントに送信される POST リクエストを処理します。たとえば、次の Cloud Functions は、メッセージ データのデコードと出力を行う Webhook です。

Go


// Package helloworld provides a set of Cloud Functions samples.
package helloworld

import (
	"context"
	"log"
)

// PubSubMessage is the payload of a Pub/Sub event.
type PubSubMessage struct {
	Data []byte `json:"data"`
}

// HelloPubSub consumes a Pub/Sub message.
func HelloPubSub(ctx context.Context, m PubSubMessage) error {
	name := string(m.Data) // Automatically decoded from base64.
	if name == "" {
		name = "World"
	}
	log.Printf("Hello, %s!", name)
	return nil
}

Java


import com.google.cloud.functions.BackgroundFunction;
import com.google.cloud.functions.Context;
import functions.eventpojos.PubSubMessage;
import java.nio.charset.StandardCharsets;
import java.util.Base64;
import java.util.logging.Level;
import java.util.logging.Logger;

public class HelloPubSub implements BackgroundFunction<PubSubMessage> {
  private static final Logger logger = Logger.getLogger(HelloPubSub.class.getName());

  @Override
  public void accept(PubSubMessage message, Context context) {
    String name = "world";
    if (message != null && message.getData() != null) {
      name = new String(
          Base64.getDecoder().decode(message.getData().getBytes(StandardCharsets.UTF_8)),
          StandardCharsets.UTF_8);
    }
    logger.info(String.format("Hello %s!", name));
    return;
  }
}

Node.js

/**
 * Background Cloud Function to be triggered by Pub/Sub.
 * This function is exported by index.js, and executed when
 * the trigger topic receives a message.
 *
 * @param {object} message The Pub/Sub message.
 * @param {object} context The event metadata.
 */
exports.helloPubSub = (message, context) => {
  const name = message.data
    ? Buffer.from(message.data, 'base64').toString()
    : 'World';

  console.log(`Hello, ${name}!`);
};

Python

def hello_pubsub(event, context):
    """Background Cloud Function to be triggered by Pub/Sub.
    Args:
         event (dict):  The dictionary with data specific to this type of
         event. The `data` field contains the PubsubMessage message. The
         `attributes` field will contain custom attributes if there are any.
         context (google.cloud.functions.Context): The Cloud Functions event
         metadata. The `event_id` field contains the Pub/Sub message ID. The
         `timestamp` field contains the publish time.
    """
    import base64

    print("""This Function was triggered by messageId {} published at {}
    """.format(context.event_id, context.timestamp))

    if 'data' in event:
        name = base64.b64decode(event['data']).decode('utf-8')
    else:
        name = 'World'
    print('Hello {}!'.format(name))

push リクエストの受信後、HTTP ステータス コードを返します。メッセージへ確認応答するには、次のいずれかのステータス コードを返します。

  • 102
  • 200
  • 201
  • 202
  • 204

メッセージに対し否定応答を送信するには、他のステータス コードを返します。否定応答を送信するか、確認応答期限が切れた場合、Pub/Sub はメッセージを再送信します。push サブスクリプションから受信する個々のメッセージの確認応答期限は、変更できません。

認証と認可

push サブスクリプションが認証を使用する場合、Pub/Sub サービスは JSON Web Token(JWT)に署名し、push リクエストの認証ヘッダーで JWT を送信します。JWT にはクレームと署名が含まれます。

サブスクライバーは JWT をデコードし、次のことを確認できます。

  • クレームが正確である。
  • Pub/Sub サービスがクレームに署名した。

サブスクライバーがファイアウォールを使用している場合、push リクエストを受信できません。push リクエストを受信するには、ファイアウォールをオフにして JWT を検証する必要があります。

JWT の形式

JWT は、ヘッダー、クレームセット、署名で構成される OpenIDConnect JWT です。Pub/Sub サービスは、JWT をピリオド区切りの base64 文字列としてエンコードします。

たとえば、次の認証ヘッダーにはエンコードされた JWT が含まれています。

"Authorization" : "Bearer
eyJhbGciOiJSUzI1NiIsImtpZCI6IjdkNjgwZDhjNzBkNDRlOTQ3MTMzY2JkNDk5ZWJjMWE2MWMzZDVh
YmMiLCJ0eXAiOiJKV1QifQ.eyJhdWQiOiJodHRwczovL2V4YW1wbGUuY29tIiwiYXpwIjoiMTEzNzc0M
jY0NDYzMDM4MzIxOTY0IiwiZW1haWwiOiJnYWUtZ2NwQGFwcHNwb3QuZ3NlcnZpY2VhY2NvdW50LmNvb
SIsImVtYWlsX3ZlcmlmaWVkIjp0cnVlLCJleHAiOjE1NTAxODU5MzUsImlhdCI6MTU1MDE4MjMzNSwia
XNzIjoiaHR0cHM6Ly9hY2NvdW50cy5nb29nbGUuY29tIiwic3ViIjoiMTEzNzc0MjY0NDYzMDM4MzIxO
TY0In0.QVjyqpmadTyDZmlX2u3jWd1kJ68YkdwsRZDo-QxSPbxjug4ucLBwAs2QePrcgZ6hhkvdc4UHY
4YF3fz9g7XHULNVIzX5xh02qXEH8dK6PgGndIWcZQzjSYfgO-q-R2oo2hNM5HBBsQN4ARtGK_acG-NGG
WM3CQfahbEjZPAJe_B8M7HfIu_G5jOLZCw2EUcGo8BvEwGcLWB2WqEgRM0-xt5-UPzoa3-FpSPG7DHk7
z9zRUeq6eB__ldb-2o4RciJmjVwHgnYqn3VvlX9oVKEgXpNFhKuYA-mWh5o7BCwhujSMmFoBOh6mbIXF
cyf5UiVqKjpqEbqPGo_AvKvIQ9VTQ" 

ヘッダーとクレームセットは JSON 文字列です。デコードされると次の形式になります。

{"alg":"RS256","kid":"7d680d8c70d44e947133cbd499ebc1a61c3d5abc","typ":"JWT"}

{
   "aud":"https://example.com",
   "azp":"113774264463038321964",
   "email":"gae-gcp@appspot.gserviceaccount.com",
   "sub":"113774264463038321964",
   "email_verified":true,
   "exp":1550185935,
   "iat":1550182335,
   "iss":"https://accounts.google.com"
  }

push エンドポイントに送信されるリクエストに添付されるトークンは、最大 1 時間前のものである可能性があります。

Pub/Sub での push 認証の設定

サブスクリプションの認証構成は、次の 2 つのパラメータで構成されます。

  • サービス アカウント: push サブスクリプションに関連付けられている GCP サービス アカウント。push リクエストには、このサービス アカウントの ID が含まれます。 たとえば、ロールが roles/run.invoker で、特定の Cloud Run(フルマネージド)サービスにバインドされたサービスアカウントで構成された push サブスクリプションは、その Cloud Run(フルマネージド)サービスを呼び出すことができます。
  • トークン オーディエンス(省略可): Webhook で特定のトークンの対象ユーザーの検証に使用する、大文字と小文字が区別されない単一の文字列

これらのフィールドの構成に加えて、サービス アカウント用にトークンを作成する権限を Pub/Sub に付与する必要があります。Pub/Sub では、プロジェクト専用のサービス アカウント service-PROJECT_NUMBER@gcp-sa-pubsub.iam.gserviceaccount.com. が作成、維持されます。このサービス アカウントは、サービス アカウント トークン作成者の役割を必要とします。Cloud Console を使用して push 認証用のサブスクリプションを設定する場合は、この役割が自動的に付与されます。その他の場合は、アカウントに対するロールを明示的に付与する必要があります。

コマンドライン

# grant Cloud Pub/Sub the permission to create tokens
PUBSUB_SERVICE_ACCOUNT="service-${PROJECT_NUMBER}@gcp-sa-pubsub.iam.gserviceaccount.com"
gcloud projects add-iam-policy-binding ${PROJECT_ID} \
 --member="serviceAccount:${PUBSUB_SERVICE_ACCOUNT}"\
 --role='roles/iam.serviceAccountTokenCreator'

# configure the subscription push identity
gcloud pubsub subscriptions (create|update|modify-push-config) ${SUBSCRIPTION} \
 --topic=${TOPIC} \
 --push-endpoint=${PUSH_ENDPOINT_URI} \
 --push-auth-service-account=${SERVICE_ACCOUNT_EMAIL} \
 --push-auth-token-audience=${OPTIONAL_AUDIENCE_OVERRIDE}

Console

  1. Pub/Sub トピックページに移動します。

    [トピック] ページに移動

  2. トピック名をクリックします。

  3. サブスクリプションを作成または更新します。

  4. ID を入力し、任意でオーディエンスを入力します。

push エンドポイントによる認証と承認

クレーム

JWT を使用して、クレーム(email クレームと aud クレームを含む)が Google によって署名されていることを検証できます。認証と承認の両方で Google の OAuth 2.0 API を使用する方法の詳細については、OpenID Connect をご覧ください。

こうしたクレームを有用にするメカニズムが 2 つあります。まず、Pub/Sub では、サービス アカウント ID と push サブスクリプションを関連付けるユーザー アカウントまたはサービス アカウントに、プロジェクトまたはサービス アカウント用のサービス アカウント ユーザーのロール(roles/iam.serviceAccountUser)を付与する必要があります。

次に、トークンの署名に使用する証明書に対するアクセス権を厳密に管理する必要があります。トークンを作成するには、Pub/Sub が個別の署名サービス アカウント ID を使用して、Google の内部サービスを呼び出す必要があります。申請されたサービス アカウント、またはアカウントを含むプロジェクト用のトークンを作成するには、署名サービス アカウントを承認する必要があります。それには、iam.serviceAccounts.getOpenIdToken 権限またはサービス アカウント トークン作成者のロール(roles/iam.serviceAccountTokenCreator)を使用します。

このロールまたは権限は、どのアカウントにも付与できます。IAM サービスを使用して、この権限を Pub/Sub 署名アカウントに付与するように指定することもできます。具体的には、Pub/Sub では次のようなサービス アカウントを使用します。

service-{project_number}@gcp-sa-pubsub.iam.gserviceaccount.com
  • {project_number}: サブスクリプションを含む GCP プロジェクト。
  • gcp-sa-pubsub: 署名サービス アカウントを含む、Google 所有のプロジェクト。

トークンの検証

次の例は、App Engine アプリケーションに対する push リクエストを認証する方法を示しています。

プロトコル

リクエスト:

GET https://oauth2.googleapis.com/tokeninfo?id_token={BEARER_TOKEN}

レスポンス:

200 OK
{
    "alg": "RS256",
    "aud": "example.com",
    "azp": "104176025330667568672",
    "email": "{SERVICE_ACCOUNT_NAME}@{YOUR_PROJECT_NAME}.iam.gserviceaccount.com",
    "email_verified": "true",
    "exp": "1555463097",
    "iat": "1555459497",
    "iss": "https://accounts.google.com",
    "kid": "3782d3f0bc89008d9d2c01730f765cfb19d3b70e",
    "sub": "104176025330667568672",
    "typ": "JWT"
}

Java

@WebServlet(value = "/pubsub/authenticated-push")
public class PubSubAuthenticatedPush extends HttpServlet {
  private final String pubsubVerificationToken = System.getenv("PUBSUB_VERIFICATION_TOKEN");
  private final MessageRepository messageRepository;
  private final GoogleIdTokenVerifier verifier =
      new GoogleIdTokenVerifier.Builder(new NetHttpTransport(), new JacksonFactory())
          /**
           * Please change example.com to match with value you are providing while creating
           * subscription as provided in @see <a
           * href="https://github.com/GoogleCloudPlatform/java-docs-samples/tree/master/appengine-java8/pubsub">README</a>.
           */
          .setAudience(Collections.singletonList("example.com"))
          .build();
  private final Gson gson = new Gson();
  private final JsonParser jsonParser = new JsonParser();

  @Override
  public void doPost(HttpServletRequest req, HttpServletResponse resp)
      throws IOException, ServletException {

    // Verify that the request originates from the application.
    if (req.getParameter("token").compareTo(pubsubVerificationToken) != 0) {
      resp.setStatus(HttpServletResponse.SC_BAD_REQUEST);
      return;
    }
    // Get the Cloud Pub/Sub-generated JWT in the "Authorization" header.
    String authorizationHeader = req.getHeader("Authorization");
    if (authorizationHeader == null
        || authorizationHeader.isEmpty()
        || authorizationHeader.split(" ").length != 2) {
      resp.setStatus(HttpServletResponse.SC_BAD_REQUEST);
      return;
    }
    String authorization = authorizationHeader.split(" ")[1];

    try {
      // Verify and decode the JWT.
      // Note: For high volume push requests, it would save some network overhead
      // if you verify the tokens offline by decoding them using Google's Public
      // Cert; caching already seen tokens works best when a large volume of
      // messsages have prompted a singple push server to handle them, in which
      // case they would all share the same token for a limited time window.
      GoogleIdToken idToken = verifier.verify(authorization);
      messageRepository.saveToken(authorization);
      messageRepository.saveClaim(idToken.getPayload().toPrettyString());
      // parse message object from "message" field in the request body json
      // decode message data from base64
      Message message = getMessage(req);
      messageRepository.save(message);
      // 200, 201, 204, 102 status codes are interpreted as success by the Pub/Sub system
      resp.setStatus(102);
      super.doPost(req, resp);
    } catch (Exception e) {
      resp.setStatus(HttpServletResponse.SC_BAD_REQUEST);
    }
  }

  private Message getMessage(HttpServletRequest request) throws IOException {
    String requestBody = request.getReader().lines().collect(Collectors.joining("\n"));
    JsonElement jsonRoot = jsonParser.parse(requestBody);
    String messageStr = jsonRoot.getAsJsonObject().get("message").toString();
    Message message = gson.fromJson(messageStr, Message.class);
    // decode from base64
    String decoded = decode(message.getData());
    message.setData(decoded);
    return message;
  }

  private String decode(String data) {
    return new String(Base64.getDecoder().decode(data));
  }

  PubSubAuthenticatedPush(MessageRepository messageRepository) {
    this.messageRepository = messageRepository;
  }

  public PubSubAuthenticatedPush() {
    this(MessageRepositoryImpl.getInstance());
  }
}

Node.js

app.post('/pubsub/authenticated-push', jsonBodyParser, async (req, res) => {
  // Verify that the request originates from the application.
  if (req.query.token !== PUBSUB_VERIFICATION_TOKEN) {
    res.status(400).send('Invalid request');
    return;
  }

  // Verify that the push request originates from Cloud Pub/Sub.
  try {
    // Get the Cloud Pub/Sub-generated JWT in the "Authorization" header.
    const bearer = req.header('Authorization');
    const [, token] = bearer.match(/Bearer (.*)/);
    tokens.push(token);

    // Verify and decode the JWT.
    // Note: For high volume push requests, it would save some network
    // overhead if you verify the tokens offline by decoding them using
    // Google's Public Cert; caching already seen tokens works best when
    // a large volume of messages have prompted a single push server to
    // handle them, in which case they would all share the same token for
    // a limited time window.
    const ticket = await authClient.verifyIdToken({
      idToken: token,
      audience: 'example.com',
    });

    const claim = ticket.getPayload();
    claims.push(claim);
  } catch (e) {
    res.status(400).send('Invalid token');
    return;
  }

  // The message is a unicode string encoded in base64.
  const message = Buffer.from(req.body.message.data, 'base64').toString(
    'utf-8'
  );

  messages.push(message);

  res.status(200).send();
});

Python

@app.route('/push-handlers/receive_messages', methods=['POST'])
def receive_messages_handler():
    # Verify that the request originates from the application.
    if (request.args.get('token', '') !=
            current_app.config['PUBSUB_VERIFICATION_TOKEN']):
        return 'Invalid request', 400

    # Verify that the push request originates from Cloud Pub/Sub.
    try:
        # Get the Cloud Pub/Sub-generated JWT in the "Authorization" header.
        bearer_token = request.headers.get('Authorization')
        token = bearer_token.split(' ')[1]
        TOKENS.append(token)

        # Verify and decode the JWT. `verify_oauth2_token` verifies
        # the JWT signature, the `aud` claim, and the `exp` claim.
        # Note: For high volume push requests, it would save some network
        # overhead if you verify the tokens offline by downloading Google's
        # Public Cert and decode them using the `google.auth.jwt` module;
        # caching already seen tokens works best when a large volume of
        # messages have prompted a single push server to handle them, in which
        # case they would all share the same token for a limited time window.
        claim = id_token.verify_oauth2_token(token, requests.Request(),
                                             audience='example.com')
        CLAIMS.append(claim)
    except Exception as e:
        return 'Invalid token: {}\n'.format(e), 400

    envelope = json.loads(request.data.decode('utf-8'))
    payload = base64.b64decode(envelope['message']['data'])
    MESSAGES.append(payload)
    # Returning any 2xx status indicates successful receipt of the message.
    return 'OK', 200

Go

// receiveMessagesHandler validates authentication token and caches the Pub/Sub
// message received.
func (a *app) receiveMessagesHandler(w http.ResponseWriter, r *http.Request) {
	if r.Method != "POST" {
		http.Error(w, http.StatusText(http.StatusMethodNotAllowed), http.StatusMethodNotAllowed)
		return
	}

	// Verify that the request originates from the application.
	// a.pubsubVerificationToken = os.Getenv("PUBSUB_VERIFICATION_TOKEN")
	if token, ok := r.URL.Query()["token"]; !ok || len(token) != 1 || token[0] != a.pubsubVerificationToken {
		http.Error(w, "Bad token", http.StatusBadRequest)
		return
	}

	// Get the Cloud Pub/Sub-generated JWT in the "Authorization" header.
	authHeader := r.Header.Get("Authorization")
	if authHeader == "" || len(strings.Split(authHeader, " ")) != 2 {
		http.Error(w, "Missing Authorization header", http.StatusBadRequest)
		return
	}
	token := strings.Split(authHeader, " ")[1]
	// Verify and decode the JWT.
	// If you don't need to control the HTTP client used you can use the
	// convenience method idtoken.Validate instead of creating a Validator.
	v, err := idtoken.NewValidator(r.Context(), option.WithHTTPClient(a.defaultHTTPClient))
	if err != nil {
		http.Error(w, "Unable to create Validator", http.StatusBadRequest)
		return
	}
	// Please change http://example.com to match with the value you are
	// providing while creating the subscription.
	payload, err := v.Validate(r.Context(), token, "http://example.com")
	if err != nil {
		http.Error(w, fmt.Sprintf("Invalid Token: %v", err), http.StatusBadRequest)
		return
	}
	if payload.Issuer != "accounts.google.com" && payload.Issuer != "https://accounts.google.com" {
		http.Error(w, "Wrong Issuer", http.StatusBadRequest)
	}

	var pr pushRequest
	if err := json.NewDecoder(r.Body).Decode(&pr); err != nil {
		http.Error(w, fmt.Sprintf("Could not decode body: %v", err), http.StatusBadRequest)
		return
	}

	a.messagesMu.Lock()
	defer a.messagesMu.Unlock()
	// Limit to ten.
	a.messages = append(a.messages, pr.Message.Data)
	if len(a.messages) > maxMessages {
		a.messages = a.messages[len(a.messages)-maxMessages:]
	}

	fmt.Fprint(w, "OK")
}

署名なし JWT の検証方法の例は、ウェブサイト用の Google ログインガイドで確認できます。OpenID トークンの概要については、OpenID Connect ガイドをご覧ください。

Cloud Run と App Engine

Cloud Run と App Engine は、Pub/Sub で生成されたトークンを確認して、HTTP 呼び出しを自動的に認証します。ユーザーについて必要な構成は、呼び出し元アカウントに付与される必要な IAM ロールだけです。たとえば、アカウントの特定の Cloud Run エンドポイントを呼び出す権限の承認や取り消しができます。詳細については、次のチュートリアルをご覧ください。

配信の停止と再開

Pub/Sub で push エンドポイントに対するリクエストの送信を一時的に停止するには、サブスクリプションを pull に変更します。この変更が適用されるまで、数分かかることがあります。

push 配信を再開するには、URL を有効なエンドポイントにもう一度設定します。配信を完全に停止するには、サブスクリプションを削除します。

割り当て、上限、配信レート

push サブスクリプションには割り当てリソース上限があります。

Pub/Sub が success レスポンスを受信しない場合、Pub/Sub は最小 100 ミリ秒、最大 60 秒で指数バックオフを適用します。

スロースタート アルゴリズムを使用して同時 push リクエストの数を調整します。 push ウィンドウは、同時 push リクエストの最大数です。 push ウィンドウは、配信が成功すると増加し、失敗すると減少します。システムは小さなウィンドウ、N の 3 倍から開始します。N は公開リージョンの数です。

サブスクライバーがメッセージの確認応答を行うと、ウィンドウは、未処理のメッセージを N の 3,000 倍まで急激に増加します。サブスクライバーがメッセージの 99% 以上を確認応答し、push リクエストのレイテンシが平均 1 秒未満であるサブスクリプションの場合、push ウィンドウは、未処理のメッセージを N の 30,000 倍まで増やします。

push リクエストのレイテンシには、次のものがあります。

未処理のメッセージが 3,000 件を超えると、ウィンドウは比例して増加し、push エンドポイントがメッセージを過剰に受け取ることを回避できます。平均レイテンシが 1 秒を超えるか、サブスクライバーの確認応答がリクエストの 99% 未満の場合、ウィンドウは未処理のメッセージの下限 3,000 件に減少します。

push 配信のモニタリングに使用できる指標の詳細については、push サブスクリプションのモニタリングをご覧ください。