Cloud Run for Anthos on Google Cloud で Pub/Sub を使用する

このチュートリアルでは、Cloud Run for Anthos on Google Cloud サービスを作成してデプロイし、Pub/Sub push サブスクリプションから呼び出す方法について説明します。

目標

Cloud Run for Anthos on Google Cloud でサービスを作成、ビルド、デプロイする
Pub/Sub トピックにメッセージを公開してサービスを呼び出す

料金

このチュートリアルでは、Google Cloud の課金対象となる以下のコンポーネントを使用します。

料金計算ツールを使うと、予想使用量に基づいて費用の見積もりを出すことができます。

新しい Google Cloud ユーザーは無料トライアルをご利用いただける場合があります。

始める前に

Google アカウントにログインします。
Google アカウントをまだお持ちでない場合は、新しいアカウントを登録します。

Google Cloud Console の [プロジェクトセレクタ] ページで、Google Cloud プロジェクトを選択または作成します。

注: この手順で作成するリソースをそのまま保持する予定でない場合、既存のプロジェクトを選択するのではなく、新しいプロジェクトを作成してください。チュートリアルの終了後にそのプロジェクトを削除すれば、プロジェクトに関連するすべてのリソースを削除できます。

[プロジェクトの選択] ページに移動

Cloud プロジェクトに対して課金が有効になっていることを確認します。プロジェクトに対して課金が有効になっていることを確認する方法を学習する。

Cloud Run for Anthos on Google Cloud API を有効にする
Cloud SDK をインストールして初期化します。
kubectl コンポーネントをインストールします。
```
gcloud components install kubectl
```
beta コンポーネントをインストールします。
```
gcloud components install beta
```
コンポーネントを更新します。
```
gcloud components update
```
Cloud Run for Anthos on Google Cloud を使用している場合は、Cloud Run for Anthos on Google Cloud の設定に沿って手順を行って、新しいクラスタを作成します。

gcloud のデフォルトを設定する

Cloud Run for Anthos on Google Cloud サービスのデフォルトとして gcloud を構成するには:

デフォルトプロジェクトを設定します。
```
gcloud config set project PROJECT_ID
```
PROJECT_ID は、このチュートリアルで使用するプロジェクトの名前に置き換えます。
クラスタに gcloud を構成します。
```
gcloud config set kuberun/cluster CLUSTER-NAME
gcloud config set kuberun/cluster_location REGION
```
次のように置き換えます。
- CLUSTER-NAME は、クラスタに対して使用した名前に置き換えます。
- REGION は、選択したサポート対象のクラスタの場所に置き換えます。

Pub/Sub トピックを作成する

このサンプルサービスは Pub/Sub トピックに公開されたメッセージによってトリガーされるため、Pub/Sub でトピックを作成する必要があります。

新しい Pub/Sub トピックを作成するには、次のコマンドを使用します。

gcloud pubsub topics create myRunTopic

myRunTopic を使用するか、Cloud プロジェクト内で一意のトピック名に置き換えます。

コードサンプルを取得する

使用するサンプルコードを取得するには:

ローカルマシンにサンプルアプリのリポジトリのクローンを作成します。
Node.js
```
git clone https://github.com/GoogleCloudPlatform/nodejs-docs-samples.git
```
または、zip 形式のサンプルをダウンロードし、ファイルを抽出してもかまいません。
Python
```
git clone https://github.com/GoogleCloudPlatform/python-docs-samples.git
```
または、zip 形式のサンプルをダウンロードし、ファイルを抽出してもかまいません。
Go
```
git clone https://github.com/GoogleCloudPlatform/golang-samples.git
```
または、zip 形式のサンプルをダウンロードし、ファイルを抽出してもかまいません。
Java
```
git clone https://github.com/GoogleCloudPlatform/java-docs-samples.git
```
または、zip 形式のサンプルをダウンロードし、ファイルを抽出してもかまいません。

Cloud Run for Anthos on Google Cloud のサンプルコードがあるディレクトリに移動します。

Node.js

cd nodejs-docs-samples/kuberun/pubsub/

Python

cd python-docs-samples/kuberun/pubsub/

Go

cd golang-samples/kuberun/pubsub/

Java

cd java-docs-samples/kuberun/pubsub/

コードを確認する

このチュートリアルのコードは、次のものから構成されています。

受信リクエストを処理するサーバー。

Node.js

Node.js サービスをテストしやすくするため、サーバー構成はサーバーの起動とは別になっています。

Node.js ウェブサーバーは、app.js 内で設定されています。

const express = require('express');
const bodyParser = require('body-parser');
const app = express();

app.use(bodyParser.json());

ウェブサーバーは index.js で開始します。

const app = require('./app.js');
const PORT = process.env.PORT || 8080;

app.listen(PORT, () =>
  console.log(`nodejs-pubsub-tutorial listening on port ${PORT}`)
);

Python

import base64
import os

from flask import Flask, request

app = Flask(__name__)

Go


// Sample pubsub is a KubeRun service which handles Pub/Sub messages.
package main

import (
	"encoding/json"
	"io/ioutil"
	"log"
	"net/http"
	"os"
)

func main() {
	http.HandleFunc("/", HelloPubSub)
	// Determine port for HTTP service.
	port := os.Getenv("PORT")
	if port == "" {
		port = "8080"
		log.Printf("Defaulting to port %s", port)
	}
	// Start HTTP server.
	log.Printf("Listening on port %s", port)
	if err := http.ListenAndServe(":"+port, nil); err != nil {
		log.Fatal(err)
	}
}

Java

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class PubSubApplication {
  public static void main(String[] args) {
    SpringApplication.run(PubSubApplication.class, args);
  }
}

Pub/Sub メッセージを処理し、応答メッセージをログに記録するハンドラ。

Node.js

app.post('/', (req, res) => {
  if (!req.body) {
    const msg = 'no Pub/Sub message received';
    console.error(`error: ${msg}`);
    res.status(400).send(`Bad Request: ${msg}`);
    return;
  }
  if (!req.body.message) {
    const msg = 'invalid Pub/Sub message format';
    console.error(`error: ${msg}`);
    res.status(400).send(`Bad Request: ${msg}`);
    return;
  }

  const pubSubMessage = req.body.message;
  const name = pubSubMessage.data
    ? Buffer.from(pubSubMessage.data, 'base64').toString().trim()
    : 'World';

  console.log(`Hello ${name}!`);
  res.status(204).send();
});

Python

@app.route("/", methods=["POST"])
def index():
    envelope = request.get_json()
    if not envelope:
        msg = "no Pub/Sub message received"
        print(f"error: {msg}")
        return f"Bad Request: {msg}", 400

    if not isinstance(envelope, dict) or "message" not in envelope:
        msg = "invalid Pub/Sub message format"
        print(f"error: {msg}")
        return f"Bad Request: {msg}", 400

    pubsub_message = envelope["message"]

    name = "World"
    if isinstance(pubsub_message, dict) and "data" in pubsub_message:
        name = base64.b64decode(pubsub_message["data"]).decode("utf-8").strip()

    print(f"Hello {name}!")

    return ("", 204)

Go


// PubSubMessage is the payload of a Pub/Sub event.
type PubSubMessage struct {
	Message struct {
		Data []byte `json:"data,omitempty"`
		ID   string `json:"id"`
	} `json:"message"`
	Subscription string `json:"subscription"`
}

// HelloPubSub receives and processes a Pub/Sub push message.
func HelloPubSub(w http.ResponseWriter, r *http.Request) {
	var m PubSubMessage
	body, err := ioutil.ReadAll(r.Body)
	if err != nil {
		log.Printf("ioutil.ReadAll: %v", err)
		http.Error(w, "Bad Request", http.StatusBadRequest)
		return
	}
	if err := json.Unmarshal(body, &m); err != nil {
		log.Printf("json.Unmarshal: %v", err)
		http.Error(w, "Bad Request", http.StatusBadRequest)
		return
	}

	name := string(m.Message.Data)
	if name == "" {
		name = "World"
	}
	log.Printf("Hello %s!", name)
}

Java

import java.util.Base64;
import org.apache.commons.lang3.StringUtils;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;

// PubsubController consumes a Pub/Sub message.
@RestController
public class PubSubController {
  @RequestMapping(value = "/", method = RequestMethod.POST)
  public ResponseEntity receiveMessage(@RequestBody Body body) {
    // Get PubSub message from request body.
    Body.Message message = body.getMessage();
    if (message == null) {
      String msg = "Bad Request: invalid Pub/Sub message format";
      System.out.println(msg);
      return new ResponseEntity(msg, HttpStatus.BAD_REQUEST);
    }

    String data = message.getData();
    String target =
        !StringUtils.isEmpty(data) ? new String(Base64.getDecoder().decode(data)) : "World";
    String msg = "Hello " + target + "!";

    System.out.println(msg);
    return new ResponseEntity(msg, HttpStatus.OK);
  }
}

正確な HTTP レスポンスコードを返すようにサービスをコーディングする必要があります。HTTP 200 や 204 などの成功コードは、Pub/Sub メッセージの処理の完了を意味します。HTTP 400 や 500 などのエラーコードは、push を使用したメッセージの受信で説明されているように、メッセージが再試行されることを示します。

サービスの動作環境を定義する Dockerfile。Dockerfile の内容は言語によって異なります。

Node.js


# Use the official lightweight Node.js 10 image.
# https://hub.docker.com/_/node
FROM node:12-slim

# Create and change to the app directory.
WORKDIR /usr/src/app

# Copy application dependency manifests to the container image.
# A wildcard is used to ensure both package.json AND package-lock.json are copied.
# Copying this separately prevents re-running npm install on every code change.
COPY package*.json ./

# Install dependencies.
# If you add a package-lock.json speed your build by switching to 'npm ci'.
# RUN npm ci --only=production
RUN npm install --production

# Copy local code to the container image.
COPY . .

# Run the web service on container startup.
CMD [ "npm", "start" ]

Python


# Use the official Python image.
# https://hub.docker.com/_/python
FROM python:3.9

# Allow statements and log messages to immediately appear in the KubeRun logs
ENV PYTHONUNBUFFERED True

# Copy application dependency manifests to the container image.
# Copying this separately prevents re-running pip install on every code change.
COPY requirements.txt ./

# Install production dependencies.
RUN pip install -r requirements.txt

# Copy local code to the container image.
ENV APP_HOME /app
WORKDIR $APP_HOME
COPY . ./

# Run the web service on container startup.
# Use gunicorn webserver with one worker process and 8 threads.
# For environments with multiple CPU cores, increase the number of workers
# to be equal to the cores available.
CMD exec gunicorn --bind :$PORT --workers 1 --threads 8 --timeout 0 main:app

Go


# Use the offical golang image to create a binary.
# This is based on Debian and sets the GOPATH to /go.
# https://hub.docker.com/_/golang
FROM golang:1.15-buster as builder

# Create and change to the app directory.
WORKDIR /app

# Retrieve application dependencies.
# This allows the container build to reuse cached dependencies.
# Expecting to copy go.mod and if present go.sum.
COPY go.* ./
RUN go mod download

# Copy local code to the container image.
COPY . ./

# Build the binary.
RUN go build -mod=readonly -v -o server

# Use the official Debian slim image for a lean production container.
# https://hub.docker.com/_/debian
# https://docs.docker.com/develop/develop-images/multistage-build/#use-multi-stage-builds
FROM debian:buster-slim
RUN set -x && apt-get update && DEBIAN_FRONTEND=noninteractive apt-get install -y \
    ca-certificates && \
    rm -rf /var/lib/apt/lists/*

# Copy the binary to the production image from the builder stage.
COPY --from=builder /app/server /server

# Run the web service on container startup.
CMD ["/server"]

Java

このサンプルでは、Jib を使用して一般的な Java ツールにより Docker イメージをビルドします。Jib は、Dockerfile や Docker をインストールせずにコンテナのビルドを最適化します。Jib を使用して Java コンテナを構築する方法の詳細を確認します。

<plugin>
  <groupId>com.google.cloud.tools</groupId>
  <artifactId>jib-maven-plugin</artifactId>
  <version>2.7.0</version>
  <configuration>
    <to>
      <image>gcr.io/PROJECT_ID/pubsub</image>
    </to>
  </configuration>
</plugin>

Pub/Sub リクエストの送信元を認証する方法の詳細は、以下の Pub/Sub との統合をご覧ください。

コードを配布する

コードの配布は、Cloud Build でコンテナイメージをビルドする、Container Registry にコンテナイメージをアップロードする、Cloud Run for Anthos on Google Cloud にコンテナイメージをデプロイする、という 3 つのステップで構成されます。

コードを配布するには:

コンテナをビルドして、Container Registry に公開します。
Node.js
```
gcloud builds submit --tag gcr.io/PROJECT_ID/pubsub
```
PROJECT_ID は Cloud プロジェクト ID、pubsub はサービスに付ける名前です。

ビルドが成功すると、ID、作成時間、イメージ名を含む SUCCESS メッセージが表示されます。イメージが Container Registry に保存されます。このイメージは必要に応じて再利用できます。
Python
```
gcloud builds submit --tag gcr.io/PROJECT_ID/pubsub
```
PROJECT_ID は Cloud プロジェクト ID、pubsub はサービスに付ける名前です。

ビルドが成功すると、ID、作成時間、イメージ名を含む SUCCESS メッセージが表示されます。イメージが Container Registry に保存されます。このイメージは必要に応じて再利用できます。
Go
```
gcloud builds submit --tag gcr.io/PROJECT_ID/pubsub
```
PROJECT_ID は Cloud プロジェクト ID、pubsub はサービスに付ける名前です。

ビルドが成功すると、ID、作成時間、イメージ名を含む SUCCESS メッセージが表示されます。イメージが Container Registry に保存されます。このイメージは必要に応じて再利用できます。
Java
```
mvn compile jib:build -Dimage=gcr.io/PROJECT_ID/pubsub
```
PROJECT_ID は Cloud プロジェクト ID、pubsub はサービスに付ける名前です。

ビルドが成功すると、BUILD SUCCESS メッセージが表示されます。イメージが Container Registry に保存されます。このイメージは必要に応じて再利用できます。
次のコマンドを実行して、アプリをデプロイします。
```
gcloud kuberun core services create pubsub-tutorial --image gcr.io/PROJECT_ID/pubsub
```
PROJECT_ID を Cloud プロジェクト ID に置き換えます。pubsub はコンテナ名、pubsub-tutorial はサービスの名前です。コンテナイメージは、gcloud の設定で構成したサービスとクラスタにデプロイされることに注意してください。

デプロイが完了するまで待ちます。30 秒ほどかかる場合があります。成功すると、コマンドラインにサービス URL が表示されます。この URL は、Pub/Sub サブスクリプションの構成で使用します。
サービスにコードの更新をデプロイする場合は、上記のステップを繰り返します。サービスをデプロイするたびに、リビジョンが作成されます。準備ができると、トラフィックの送信が自動的に開始します。

Pub/Sub との統合

Cloud Run for Anthos on Google Cloud サービスをデプロイしたので、メッセージを push するように Pub/Sub を構成します。

Pub/Sub とサービスを統合するには:

プロジェクトで Pub/Sub 認証トークンを作成できるようにします。
```
gcloud projects add-iam-policy-binding PROJECT_ID \
     --member=serviceAccount:service-PROJECT-NUMBER@gcp-sa-pubsub.iam.gserviceaccount.com \
     --role=roles/iam.serviceAccountTokenCreator
```
次のように置き換えます。
- PROJECT_ID を Cloud プロジェクト ID に置き換えます。
- PROJECT-NUMBER を Cloud プロジェクト番号に置き換えます。
Pub/Sub サブスクリプションの ID を表すサービスアカウントを作成または選択します。
```
gcloud iam service-accounts create kuberun-pubsub-invoker \
     --display-name "KubeRun Pub/Sub Invoker"
```
kuberun-pubsub-invoker を使用するか、Cloud プロジェクト内で一意の名前に置き換えます。
このサービスアカウントを使用して Pub/Sub サブスクリプションを作成します。
1. クラスタで自動 TLS と HTTPS を有効にし、サービスにドメインマッピングを追加します。
2. Pub/Sub のドメイン所有者を登録します。
3. Pub/Sub メッセージに添付された認証トークンを検証するコードを追加します。サンプルコードは、push エンドポイントによる認証と認可で提供されています。
  
  認証では、トークンが有効であり、予想されるサービスアカウントに関連付けられている必要があります。Cloud Run（フルマネージド）とは異なり、Cloud Run for Anthos on Google Cloud には、トークンが有効であることを示す組み込みの認証チェックがないか、サービスアカウントに Cloud Run for Anthos on Google Cloud サービスを起動する権限がありません。
4. このサービスアカウントを使用して Pub/Sub サブスクリプションを作成します。
```
gcloud pubsub subscriptions create myRunSubscription --topic myRunTopic \
     --push-endpoint=SERVICE-URL/ \
     --push-auth-service-account=kuberun-pubsub-invoker@PROJECT_ID.iam.gserviceaccount.com
```
  次のように置き換えます。
  - myRunTopic は、以前に作成したトピックに置き換えます。
  - SERVICE-URL は、カスタムサービス URL に置き換えます。プロトコルとして https を指定します。
  - PROJECT_ID を Cloud プロジェクト ID に置き換えます。
  --push-auth-service-account フラグは、認証と認可のために Pub/Sub の push 機能を有効にします。

これで、サービスが Pub/Sub と完全に統合されました。

試してみる

エンドツーエンドのソリューションをテストするには、次の手順を行います。

トピックに Pub/Sub メッセージを送信します。
```
gcloud pubsub topics publish myRunTopic --message "Runner"
```
このチュートリアルで説明したコマンドラインを使用する代わりに、プログラムでメッセージを発行することもできます。詳しくは、メッセージの公開をご覧ください。
サービスログに移動します。
1. Google Cloud Console に移動します。
2. pubsub-tutorial サービスをクリックします。
3. [ログ] タブを選択します。
  
  ログが表示されるまで少し時間がかかることがあります。すぐに表示されない場合は、しばらくしてからもう一度確認してください。
「Hello Runner!」というメッセージを見つけます。

クリーンアップ

Pub/Sub で Cloud Run for Anthos on Google Cloud を使用する場合の詳しいユースケースについては、クリーンアップをスキップして、イメージの処理のチュートリアルに進んでください。

このチュートリアル用に新規プロジェクトを作成した場合は、そのプロジェクトを削除します。既存のプロジェクトを使用し、このチュートリアルで変更を加えずに残す場合は、チュートリアル用に作成したリソースを削除します。

プロジェクトを削除する

課金をなくす最も簡単な方法は、チュートリアル用に作成したプロジェクトを削除することです。

プロジェクトを削除するには:

注意: プロジェクトを削除すると、次のような影響があります。

プロジェクト内のすべてのものが削除されます。このチュートリアルで既存のプロジェクトを使用した場合、それを削除すると、そのプロジェクトで行った他の作業もすべて削除されます。
カスタムプロジェクト ID が失われます。このプロジェクトを作成したときに、将来使用するカスタムプロジェクト ID を作成した可能性があります。そのプロジェクト ID を使用した URL（たとえば、appspot.com）を保持するには、プロジェクト全体ではなくプロジェクト内の選択したリソースだけを削除します。

複数のチュートリアルとクイックスタートを検討する予定がある場合は、プロジェクトを再利用すると、プロジェクトの割り当て制限を超えないようにできます。

Cloud Console で [リソースの管理] ページに移動します。
[リソースの管理] に移動
プロジェクトリストで、削除するプロジェクトを選択し、[削除] をクリックします。
ダイアログでプロジェクト ID を入力し、[シャットダウン] をクリックしてプロジェクトを削除します。

チュートリアルリソースを削除する

このチュートリアルでデプロイした Cloud Run for Anthos on Google Cloud サービスを削除します。
```
gcloud kuberun core services delete SERVICE-NAME
```
SERVICE-NAME は、選択したサービス名です。

Google Cloud Console から Cloud Run for Anthos on Google Cloud サービスを削除することもできます。
チュートリアルを設定したときに追加した gcloud のデフォルト構成を削除します。
```
 gcloud config unset kuberun/cluster
 gcloud config unset kuberun/cluster_location
```
プロジェクト構成を削除します。
```
 gcloud config unset project
```
このチュートリアルで作成した他の Google Cloud リソースを削除します。
- Pub/Sub トピック myRunTopic を削除します
- Pub/Sub サブスクリプション myRunSubscription を削除します。
- gcr.io/PROJECT_ID/pubsub という名前のコンテナイメージを Container Registry から削除します。
- 呼び出し元のサービスアカウント kuberun-pubsub-invoker@PROJECT_ID.iam.gserviceaccount.com を削除します。
- このチュートリアル用にクラスタを作成した場合は、クラスタを削除します。

次のステップ

このチュートリアルでデプロイされたサンプルサービスを展開して、Cloud Storage にアップロードされた画像を変更する画像処理機能を追加する。
トピックを Pub/Sub アーキテクチャに合わせる方法とトピックの管理方法を学習する。
サブスクリプションの管理における Pub/Sub サブスクリプションの詳細を学習する。
Google Cloud のその他の機能を試す。チュートリアルをご覧ください。

Cloud Run for Anthos on Google Cloud で Pub/Sub を使用する

目標

料金

始める前に

gcloud のデフォルトを設定する

Pub/Sub トピックを作成する

コードサンプルを取得する

Node.js

Python

Go

Java

Node.js

Python

Go

Java

コードを確認する

Node.js

Python

Go

Java

Node.js

Python

Go

Java

Node.js

Python

Go

Java

コードを配布する

Node.js

Python

Go

Java

Pub/Sub との統合

試してみる

クリーンアップ

プロジェクトを削除する

チュートリアル リソースを削除する

次のステップ

チュートリアルリソースを削除する