go-grpc-prometheusでgRPCのmetricsをPrometeusとGrafanaでモニタリングしてみた
gRPC Ecosystemの１つにgo-grpc-prometheusがあります。今回は「gRPC Ecosystemのgo-grpc-prometheusを試してみました」エントリです。
go-grpc-prometheus
go-grpc-prometheusはgRPCのmetricsをPrometheusでモニタリングできるログ出力をサポートするインターセプターを提供します。
取得できるmetricsはレポジトリのREADMEにまとまっています。
GitHub - grpc-ecosystem/go-grpc-prometheus: Prometheus monitoring for your gRPC Go servers.
gRPC Goはインターセプターをサポートしていますので次のようにClientとServerそれぞれに設定します。
- Serverでは次のように。https://github.com/grpc-ecosystem/go-grpc-prometheus#server-side
- Clientでは次のように。https://github.com/grpc-ecosystem/go-grpc-prometheus#client-side
PrometheusでモニタリングしたmetricsをGrafanaでもモニタリングしてみる
go-grpc-prometheusでgRPCのmetricsが取得できるようになります。Prometheusを起動すればmetricsをモニタリングできるようになります。合わせてPrometheusでモニタリングしているmetricsをGrafanaでもモニタリングしてみます。
シンプルなEchoサービスを作る
unary RPCsを利用してシンプルなEchoサービスを作ります。
proto
syntax = "proto3"; option go_package = "protobuf"; package proto; service EchoService { rpc EchoService (Message) returns (Message) {} } message Message { string message = 1; }
Server Side
Server sideはgRPCのClientからのリクエストに応えるServer-side of gRPCの役割とPrometeusのためのMetricsを出力する役割の２つが必要です。
１つのPortでHTTP/2 (gRPC)とHTTP/1.1のリクエストを解釈する必要があるのでsoheilhy/cmuxを使います。
ほぼ素の使い方ですがServer-sideのソースは次のようになりました。（コード抜粋。詳細はnsoushi/go-grpc-prometheus-demoにあります。）
func main() { // Create the main listener. s, err := net.Listen("tcp", fmt.Sprintf(":%s", os.Getenv("GRPC_SERVER_PORT"))) if err != nil { log.Fatal(err) } // Create a cmux. m := cmux.New(s) // Match connections in order: grpcL := m.Match(cmux.HTTP2HeaderField("content-type", "application/grpc")) httpL := m.Match(cmux.HTTP1Fast()) // gRPC server grpcS := grpc.NewServer( grpc.UnaryInterceptor(grpc_prometheus.UnaryServerInterceptor), grpc.StreamInterceptor(grpc_prometheus.StreamServerInterceptor), ) pb.RegisterEchoServiceServer(grpcS, newGrpcServer()) // prometheus metrics server grpc_prometheus.Register(grpcS) httpS := &http.Server{ Handler: promhttp.Handler(), } go grpcS.Serve(grpcL) go httpS.Serve(httpL) m.Serve() }
※ unary RPCsのみなのでgrpc.StreamInterceptorは必要ないですがデモのため入れています。
Client Side
Client Sideはブラウザからリクエストを受け取りgRPCのServer-sideへリクエストを送ってくれるエンドポイントとPrometeusのためのmetricsを出力するエンドポイントの２つを用意します。
Client-sideのソースは次のようになりました。（コード抜粋。詳細はnsoushi/go-grpc-prometheus-demoにあります。）
func main() { //gRPC connection var err error conn, err = grpc.Dial( fmt.Sprintf("%s:%s", os.Getenv("GRPC_SERVER_HOST"), os.Getenv("GRPC_SERVER_PORT")), grpc.WithInsecure(), grpc.WithBackoffMaxDelay(time.Second), grpc.WithUnaryInterceptor(grpc_prometheus.UnaryClientInterceptor), grpc.WithStreamInterceptor(grpc_prometheus.StreamClientInterceptor), ) if err != nil { log.Error("Connection error: %v", err) } defer conn.Close() // handle http http.Handle("/metrics", promhttp.Handler()) http.HandleFunc("/echo", echoHandler) http.HandleFunc("/", indexHandler) // serve http http.ListenAndServe(fmt.Sprintf(":%s", os.Getenv("GRPC_CLIENT_PORT")), nil) }
Prometheusでmetricsを確認する
PrometheusはDockerで起動しました。Dockerで起動するとprometheus.ymlのtargetsにlocalhostとしてもgRPCのServer-sideとClient-sideのホストへはアクセスできないのでdocker-composeを使いコンテナ構成をまとめてホスト解決を行います。
version: "3" services: grpcserver: container_name: grpcserver build: ./server ports: - 8080:8080 environment: GRPC_SERVER_HOST: grpcserver GRPC_SERVER_PORT: 8080 grpcclient: container_name: grpcclient build: ./client ports: - 8081:8081 environment: GRPC_SERVER_HOST: grpcserver GRPC_SERVER_PORT: 8080 GRPC_CLIENT_HOST: grpcclient GRPC_CLIENT_PORT: 8081 prometheus: container_name: prometheus build: ./prometheus ports: - 9090:9090 depends_on: - grpcserver - grpcclient grafana: image: grafana/grafana ports: - "3000:3000" depends_on: - prometheus - grpcserver - grpcclient
Prometheusのコンテナを起動してhttp://localhost:9090/graphへアクセスするとgRPCのmetricsが insert metric at cursorのメニューに追加されているのが確認できます。
Grafanaでmetricsを確認する
Grafanaのコンテナもdocker-composeに入れましたのでhttp://localhost:3000/loginへアクセスするとGrafanaのダッシュボードを確認できます。Data Sourceにprometheusを追加してDashboardを作成します。
次のようなServer-sideのダッシュボードを作成しました。
gRPCのServer-sideのレスポンス送信数、クライアントからの受信数をGrafanaに設定しました。
※ nsoushi/go-grpc-prometheus-demoのgrafanaフォルダにServer-sideとClient-sideのダッシュボード設定をエクスポートしたJSONがあります。このJSONをインポートするとダッシュボードが簡単に作れます。詳細はレポジトリのREADMEを参照してください。
まとめ
go-grpc-prometheusをつかってgRPCのmetricsをPrometheusとGrafanaでモニタリングしました。go-grpc-prometheusの導入はインターセプターを入れるだけなので簡単ではありますが複数のClient-sideとServer-sideの条件での検証、負荷検証などサービスへの導入検証が必要。
コードを公開しています
コード全体はgitbubで確認できます。
Terraform 0.9がリリース。0.8.xから0.9.xのStateマイグレーション手順をまとめました。
HashiCorpからTerraform 0.9がリリースされました。「よし、最新バージョンにあげよう。」と作業をはじめましたがremoteコマンドが使えない。どうやら0.9からはremoteコマンドが廃止されたようです。このエントリではTerraform 0.9にバージョンアップをして0.8以前のterraform stateをマイグレーションする方法をまとめます。
remoteコマンドの廃止
remoteコマンドが廃止になりました。代わりにbackendsを利用してS3などのremoteにあるtfstateファイルの管理を行います。
remote stateがbackendsに置き換わる過程は次のPull Requestから確認できます。 github.com
0.8以前を利用している場合はbackendを有効にしたtfstateファイルを用意する必要があります。次からは0.8.xまでのリソース状態を保持したまま新機能のbackendを有効にしたtfstateファイルへのマイグレーション手順についてまとめていきます。
マイグレーション手順
次の環境のマイグレーション手順になります。
0.8.8から0.9.1へのバージョンアップする- これまではremoteに
S3をつかっていて、これからもS3を利用する - ロールバックできるように、これまでのtfstateファイルは保持して新しいtfstateファイルを用意する
0.8.8と0.9.1のterraformを使うのでtfenvを使ってterraformを切り替えながらマイグレーションを行う
１：tfファイルにterraformセクションを追加してbackends を設定する
次のように設定しました。
terraform { backend "s3" { bucket = "tfstate-bucket" // 自身のbucket名を設定します } }
- AWSの
access_key,secret_key,regionはそれぞれ環境変数のAWS_ACCESS_KEY_ID,AWS_SECRET_ACCESS_KEY,AWS_DEFAULT_REGIONを設定しているため省略しています。 - S3の
keyは必須ですが省略しています。後述するinitコマンドの-backend-configオプションで開発環境や本番環境ごとにS3のkeyを分けているためterraformセクションでは省略します。
※ その他bucketなどのS3の変数はこちらにまとまっています。
２：0.8.8のterraformをつかいremote configをしてtfstateファイルをローカルに同期する
tfenvでインストールしたバージョンリスト
terraform/dev ➤ tfenv list 0.9.1 0.8.8
0.8.8を使いremote configする
terraform/dev ➤ tfenv use 0.8.8 Terraform v0.8.8 terraform/dev ➤ terraform remote config -backend=S3 -backend-config="bucket=tfstate-bucket" -backend-config="key=dev" Initialized blank state with remote state enabled! Remote state configured and pulled.
- S3のkeyは開発環境の
devとしています
３：0.9.1のterraformをつかいinitをしてtfstateファイルをマイグレーションする
terraform/dev ➤ tfenv use 0.9.1 Terraform v0.9.1 terraform/dev ➤ terraform init -backend-config "key=dev" Initializing the backend... New backend configuration detected with legacy remote state! ・・・省略・・・
- 最初のaskで
remote stateから変更するか？と聞かれるのでyesを入力します。これをすることでtfstateファイル内のremoteがbackendに置き換わります。 - 次のaskでremoteのstateをローカルのstateにコピーする？と聞かれるのでローカルのstateを保持したければ
noを入力、コピーするのであればyesを入力します。すでにローカルにstateがあるのでnoと入力。
４：マイグレーションしたtfstateファイルをS3にアップロードする
マイグレーション後に0.8.8にロールバックするかもしれないので、0.8.8で運用したtfstateファイルを残したいです。そのため新しいS3のkeyをdev0.9と決めマイグレーションしたtfstateファイルをS3にアップロードします。
terraform/dev ➤ aws s3 cp ./.terraform/terraform.tfstate s3://tfstate-bucket/dev0.9
こうすることで開発中のtfstateファイルに影響が及ぶことはなくマイグレーションのロールバックができる状態にします。
４：最後にplanを実行して新しいtfstateファイルにリソースの差分がないか確認する
terraform/dev ➤ rm -rf ./.terraform terraform/dev ➤ tfenv use 0.9.1 Terraform v0.9.1 terraform/dev ➤ terraform init -backend-config "key=dev0.9" Initializing the backend... ・・・省略・・・ terraform/dev ➤ terraform plan --refresh=false No changes. Infrastructure is up-to-date. This means that Terraform did not detect any differences between your configuration and real physical resources that exist. As a result, Terraform doesn't need to do anything.
- 0.9.1からは
remote configを使わずinitを使いtfstateファイルをローカルに同期します
まとめ
- 0.9.1へtfstateファイルのマイグレーション手順をまとめました。
- 0.8.xまでは
remote configを利用していましたが、0.9.1からはinitを利用します。 - backendではtfstateのリソース情報がメモリ上に管理されます。0.8.xまではリソース状態がtfstateファイルを開けば確認できましたがbackendでは確認できません。リソース状態の管理がセキュアになりました。
- backendは
STATE LOCKINGを機能が有効になります。複数人でapplyを実行した場合にstateをロックし競合を防ぎます。CIなどでapplyが同時に稼働しても安心です。 - もし0.7.xからのマイグレーションの場合はリソース状態に差分が生まれているのでリソース状態を0.8系に合わせる必要があります。
参考URL
- Terraform 0.9 | HashiCorp
- core: introduce "backends" to replace "remote state" (superset) and fix UX by mitchellh · Pull Request #11286 · hashicorp/terraform · GitHub
- Backends: Migrating From 0.8.x and Earlier - Terraform by HashiCorp
- Backends - Terraform by HashiCorp
- State: Locking - Terraform by HashiCorp
gRPC streamingをつかうとマイクロサービスの責務が整理できるし省コネクションでメリットあるよね、という話
今回はgRPCをマイクロサービス間通信に導入することってメリットあるよね、というエントリです。 定期的に処理を実行してくれるバッチはよくあるものですがバッチの駆動をgRPCを使って次のような構成で動かしました。
Batch ControlとBatch ServerはBidirectional gRPC streamingでコネクションする。Batch ControlはRedisのPub/Subで特定のチャンネルを監視する。Batch Controlはチャンネルにキューが投げられたらBatch Serverへバッチ処理スタートのリクエストを送る。Batch Serviceはリクエストを受け取りバッチを動かし処理結果をBatch Controlに送る（レスポンスを送る）。- チャンネルにキューが投げられる度に上記の流れでバッチを稼働させる。
上記の構成を踏まえ次からはメリットをまとめます。
gRPCをマイクロサービスに導入するとメリットあるよね
キューのRead権限をバッチサーバから剥がせる
キュー駆動でバッチを動かしている場合、例えばAmazon SQSを導入しているとRead権限が必要です。上記の構成であればキューを監視するのはバッチサーバではなくコントロールサーバになります。そのためキューを監視する権限をコントロールサーバに集約できるメリットがあります。
ログ集約サーバへの送信責務もバッチサーバから剥がせる
図のとおりgRPCのBidirectional streamingを使えば複数のレスポンスを送信することができます。バッチ処理結果や各種ログはコントロールサーバへ送り、ログ集約サーバへの送信はコントロールサーバが行います。gRPCで各サービスをつないでおいてログを送り、受けとったクライアントにログの集約を任せる、といった構成は導入メリットの１つな気がします。（ログの送信漏れ考慮は必要ですが）
そもそものgRPCのメリット
そもそものgRPCのメリットがあります。異なる言語のマイクロサービス間の通信でもProtocol Buffersを定義することで容易に通信を確立できますし、streamingの方式を用途に合わせて選択することで省コネクションでマイクロサービス間のやり取りが行えます。
GoとJavaでBidirectional gRPC streamingをつかったデモ
上記の図の構成をもとにgRPCのクライアントをGoでサーバをJavaで通信方式はBidirectional streamingを採用してデモを作ってみました。
どのようなバッチサービス？
Bidirectional streamingを採用しているので、リクエストが複数あってレスポンスも複数、または１つのようなサービスを考えました。
結果、数値を受け取り割り算をして商と余りを返すサービスを実装しました。
Redisからキューを送信してクライアントがリクエストとレスポンスを受け取ったイメージです。
# Redis
$ redis-cli
127.0.0.1:6379> PUBLISH my_queue '{"serviceName" : "division", "numbers" : [10, 3]}'
# Client
12:27:50.452 Request : {serviceName:'division', message:'10', time:'time string'}
12:27:50.452 Request : {serviceName:'division', message:'3', time:'time string'}
12:27:50.455 Response: {serviceName:'division', message:'quotient:3', time:'time string'}
12:27:50.456 Response: {serviceName:'division', message:'remainder:1', time:'time string'}
クライアントは10と3のリクエストを２つ送り、商が3と余りが1の結果を受け取ります。（余りが0であればレスポンスは１つになる）
protoファイル
protoファイルは次のようになりました。
syntax = "proto3"; option go_package = "protobuf"; package proto; service MicroService { rpc MicroService (stream Request) returns (stream Response) {} } message Request { string name = 1; string message = 2; string time = 3; } message Response { string name = 1; string message = 2; string time = 3; }
クライアントのコード（Go）
リクエストを送信してレスポンスを受け取っている通信周りのコードの抜粋です。
※コード全体はgithubにあります。
waitc := make(chan struct{}) go func() { for { in, err := stream.Recv() if err == io.EOF { close(waitc) return } if err != nil { log.Error("Failed to receive a message : %v", err) return } responseLog.Info("{serviceName:'%s', message:'%s', time:'%s'}", in.Name, in.Message, in.Time) } }() for { message, err := pubSub.ReceiveMessage() if err != nil { panic(err) } requests, err := getRequests(message) if err != nil { panic(err) } for _, request := range requests { requestLog.Info("{serviceName:'%s', message:'%s', time:'%s'}", request.Name, request.Message, request.Time) if err := stream.Send(&request); err != nil { log.Error("Failed to send a message: %v", err) } } } stream.CloseSend() <-waitc
サーバのコード（Java）
リクエストを受け取りレスポンスを送信している通信周りのコードの抜粋です。
割り算をする数値が分けられて送られてきます。１回目のリクエストでキーを生成してリクエストを保持しながら２回目のリクエストで割った結果を送信しています。
※コード全体はgithubにあります。
return new StreamObserver<Microservice.Request>() { public void onNext(Microservice.Request req) { Long key = getTime(req); Observable.just(req) .subscribe(new Observer<Microservice.Request>() { @Override public void onSubscribe(Disposable d) { Log.i("Request", getRequestLog(req)); } @Override public void onNext(Microservice.Request request) { if (!routeNumber.containsKey(key)) { routeNumber.put(key, Arrays.asList(req)); } else if (routeNumber.get(key).size() == 1) { Microservice.Request prevRequest = routeNumber.get(key).get(0); Integer leftTerm = Integer.parseInt(prevRequest.getMessage()); Integer rightTerm = Integer.parseInt(req.getMessage()); Integer quotient = leftTerm / rightTerm; Integer remainder = leftTerm % rightTerm; if (remainder == 0) { responses.putIfAbsent(key, Arrays.asList( getResponse(req.getName(), String.format("quotient:%d", quotient)))); } else { responses.putIfAbsent(key, Arrays.asList( getResponse(req.getName(), String.format("quotient:%d", quotient)), getResponse(req.getName(), String.format("remainder:%d", remainder)))); } } else { Log.w(String.format("waring, unknown state. key:{%s}, value:{%s}", key, routeNumber.get(key))); } } @Override public void onError(Throwable e) { Log.e(String.format("onError %s", e.getMessage())); } @Override public void onComplete() { if (responses.containsKey(key)) { Observable.fromIterable(responses.get(key)) .subscribe(res -> { responseObserver.onNext(res); Log.i("Response", getResponseLog(res)); }); routeNumber.remove(key); responses.remove(key); } } }); } public void onError(Throwable t) { logger.log(Level.WARNING, "microService cancelled"); } public void onCompleted() { responseObserver.onCompleted(); } };
デモ
まとめ
- Bidirectional streamingは１回のコネクションでクライアントとサーバ間で複数回のリクエストとレスポンスを送ることができます。リクエスト／レスポンスの度にコネクションを確率しないので省コネクションのメリットがあります。
- クライアントはgRPCのコネクションを確立してからチャンネルのsubscribeを継続して行っています。キューが送られる度にgRCPのコネクションを繋いでいません。１回のgRCPコネクションを確立するだけでバッチサーバのコントロールが行うことができました。
- gRPCで考えてみましたがHTTP／API／JSONの通信形式であっても権限や責務を１つのサーバに集約させるメリットは受けられます。何よりProtocol Buffers定義による複数言語のサポートとstreaming方式の便利さが運用しているマイクロサービスに嵌まれば導入機会を検討するべきです。引き続きgRPCのメリットを受けられるようなユースケースを考えていきます。
コードを公開しています
コード全体はgitbubで確認できます。
