grpc-microservices

gRPC マイクロサービス

gRPC と Protocol Buffers を使用して、高パフォーマンスで型安全なマイクロサービスを構築するための包括的なスキルです。このスキルでは、サービス設計、すべてのストリーミングパターン、インターセプター、負荷分散、エラー処理、分散システムの本番環境へのデプロイパターンを扱います。

このスキルを使用する場面

以下の場合に、このスキルを使用してください。

• 高パフォーマンスで低レイテンシの通信が必要なマイクロサービスを構築する
• サービス間のリアルタイムデータストリーミングを実装する
• Protocol Buffers を使用して、強力な契約に基づく型安全な API を設計する
• 異なる言語でサービスを記述するポリグロットシステムを作成する
• 双方向ストリーミングが必要な分散システムを構築する
• 高度なルーティングと可観測性を備えたサービスメッシュを実装する
• 後方互換性と前方互換性を保ちながら進化させる必要がある API を設計する
• パフォーマンスと型安全性が重要な内部 API を作成する
• ストリーミングデータパイプラインを備えたイベント駆動型アーキテクチャを構築する
• プッシュ機能（サーバーストリーミング）を備えたクライアント・サーバーシステムを実装する
• 効率的なバイナリーシリアライゼーションが必要なシステムを設計する
• 複数の言語向けに自動コード生成が必要なマイクロサービスを構築する

基本概念

gRPC の基礎

gRPC は、どこでも実行できる最新のオープンソース RPC フレームワークです。クライアントアプリケーションとサーバーアプリケーションが透過的に通信できるようにし、接続されたシステムを容易に構築できます。

主な特性：

• HTTP/2 ベース：多重化、サーバープッシュ、ヘッダー圧縮
• Protocol Buffers：効率的なバイナリーシリアライゼーション形式
• ストリーミング：双方向ストリーミングを標準でサポート
• コード生成：10 以上の言語でクライアント／サーバーコードを自動生成
• 期限／タイムアウト：タイムアウトを第一級の機能としてサポート
• キャンセル：サービス間でキャンセルを伝播
• インターセプター：横断的関心事に対応するミドルウェアパターン

Protocol Buffers（protobuf）

Protocol Buffers は、構造化データをシリアライズするための、言語やプラットフォームに依存しない拡張可能な仕組みです。

利点：

• コンパクト：JSON の 3～10 倍小さい
• 高速：シリアライズ／デシリアライズが JSON より 20～100 倍高速
• 型安全：検証機能を備えた強い型付けのスキーマ
• 後方互換性／前方互換性：スキーマを安全に進化させられる
• 言語サポート：10 以上の言語を公式にサポート
• 自己文書化：スキーマがドキュメントとして機能

基本構文：

syntax = "proto3";

message User {
  int32 id = 1;
  string name = 2;
  string email = 3;
}

サービス定義

gRPC サービスは .proto ファイルで定義し、利用できるメソッドとその入出力の型を指定します。

基本的なサービス：

service UserService {
  rpc GetUser(GetUserRequest) returns (GetUserResponse);
  rpc CreateUser(CreateUserRequest) returns (CreateUserResponse);
}

4 種類の RPC メソッド

1. 単項 RPC（リクエスト・レスポンス）

従来の REST API 呼び出しのような、シンプルなリクエスト・レスポンスパターンです。

rpc GetUser(GetUserRequest) returns (GetUserResponse);

ユースケース：

• CRUD 操作
• シンプルなクエリ
• 同期操作
• 従来のリクエスト・レスポンスパターン

2. サーバーストリーミング RPC

クライアントが 1 つのリクエストを送信し、サーバーがレスポンスのストリームを返します。

rpc ListUsers(ListUsersRequest) returns (stream User);

ユースケース：

• ページ分割された結果
• リアルタイム更新
• サーバー側からのイベントプッシュ
• 大規模なデータセットのダウンロード

3. クライアントストリーミング RPC

クライアントがリクエストのストリームを送信し、サーバーが 1 つのレスポンスを返します。

rpc CreateUsers(stream CreateUserRequest) returns (CreateUsersResponse);

ユースケース：

• 一括アップロード
• バッチ処理
• クライアント側での集計
• チャンク単位のファイルアップロード

4. 双方向ストリーミング RPC

クライアントとサーバーが、それぞれ独立してメッセージのストリームを送信します。

rpc Chat(stream ChatMessage) returns (stream ChatMessage);

ユースケース：

• リアルタイムチャットアプリケーション
• ライブコラボレーション
• ゲーム（リアルタイムの状態同期）
• IoT の双方向通信

Protobuf スキーマ設計

メッセージ設計のベストプラクティス

1. 明示的なフィールド番号を使用する

フィールド番号は後方互換性にとって非常に重要であり、再利用してはなりません。

message User {
  int32 id = 1;           // Never change this number
  string name = 2;        // Never change this number
  string email = 3;       // Never change this number
  // int32 age = 4;       // DEPRECATED - don't reuse 4
  string phone = 5;       // New field - use next available
}

2. 固定された値の集合には列挙型を使用する

enum UserRole {
  USER_ROLE_UNSPECIFIED = 0;  // Always have a zero value
  USER_ROLE_ADMIN = 1;
  USER_ROLE_MODERATOR = 2;
  USER_ROLE_MEMBER = 3;
}

message User {
  int32 id = 1;
  string name = 2;
  UserRole role = 3;
}

3. 複雑な型にはネストしたメッセージを使用する

message User {
  int32 id = 1;
  string name = 2;

  message Address {
    string street = 1;
    string city = 2;
    string state = 3;
    string zip = 4;
  }

  Address address = 3;
  repeated Address additional_addresses = 4;
}

4. 配列には repeated を使用する

message UserList {
  repeated User users = 1;
}

message User {
  int32 id = 1;
  string name = 2;
  repeated string tags = 3;
}

5. ユニオン型には oneof を使用する

message SearchRequest {
  string query = 1;

  oneof filter {
    string category = 2;
    int32 user_id = 3;
    string tag = 4;
  }
}

6. google.protobuf のよく知られた型を使用する

import "google/protobuf/timestamp.proto";
import "google/protobuf/duration.proto";
import "google/protobuf/empty.proto";
import "google/protobuf/wrappers.proto";

message Event {
  string id = 1;
  string name = 2;
  google.protobuf.Timestamp created_at = 3;
  google.protobuf.Duration duration = 4;
  google.protobuf.Int32Value optional_count = 5;
}

サービス設計パターン

1. リソース指向設計

RPC 向けに適応した RESTful の原則に従います。

service UserService {
  // Get single resource
  rpc GetUser(GetUserRequest) returns (User);

  // List resources
  rpc ListUsers(ListUsersRequest) returns (ListUsersResponse);

  // Create resource
  rpc CreateUser(CreateUserRequest) returns (User);

  // Update resource
  rpc UpdateUser(UpdateUserRequest) returns (User);

  // Delete resource
  rpc DeleteUser(DeleteUserRequest) returns (google.protobuf.Empty);
}

2. ページネーションパターン

message ListUsersRequest {
  int32 page_size = 1;
  string page_token = 2;
  string filter = 3;
}

message ListUsersResponse {
  repeated User users = 1;
  string next_page_token = 2;
  int32 total_count = 3;
}

3. バッチ操作パターン

message BatchGetUsersRequest {
  repeated int32 user_ids = 1;
}

message BatchGetUsersResponse {
  map<int32, User> users = 1;
  repeated int32 not_found = 2;
}

4. 長時間実行操作パターン

import "google/longrunning/operations.proto";

service BatchJobService {
  rpc ProcessBatch(BatchRequest) returns (google.longrunning.Operation);
  rpc GetOperation(GetOperationRequest) returns (google.longrunning.Operation);
}

ストリーミングパターン

サーバーストリーミングパターン

1. ページネーションストリーミング

大規模な結果セットを効率的にストリーミングします。

service ProductService {
  rpc SearchProducts(SearchRequest) returns (stream Product);
}

message SearchRequest {
  string query = 1;
  int32 limit = 2;
}

実装（Go）：

func (s *server) SearchProducts(req *pb.SearchRequest, stream pb.ProductService_SearchProductsServer) error {
    products := s.db.Search(req.Query, req.Limit)

    for _, product := range products {
        if err := stream.Send(&product); err != nil {
            return err
        }
    }

    return nil
}

2. リアルタイム更新

発生した更新をクライアントにプッシュします。

service EventService {
  rpc SubscribeToEvents(SubscribeRequest) returns (stream Event);
}

message SubscribeRequest {
  repeated string event_types = 1;
  google.protobuf.Timestamp since = 2;
}

3. ログの追跡

ログや監査証跡をストリーミングします。

service LogService {
  rpc TailLogs(TailRequest) returns (stream LogEntry);
}

message TailRequest {
  string service_name = 1;
  string level = 2;
  int32 lines = 3;
}

クライアントストリーミングパターン

1. 一括アップロード

クライアントがデータをストリーミングし、サーバーが処理して概要を返します。

service UploadService {
  rpc UploadImages(stream ImageChunk) returns (UploadSummary);
}

message ImageChunk {
  string filename = 1;
  bytes data = 2;
  int32 chunk_number = 3;
}

message UploadSummary {
  int32 total_images = 1;
  int64 total_bytes = 2;
  repeated string uploaded_filenames = 3;
}

実装（Go）：

func (s *server) UploadImages(stream pb.UploadService_UploadImagesServer) error {
    var count int32
    var totalBytes int64
    var filenames []string

    for {
        chunk, err := stream.Recv()
        if err == io.EOF {
            return stream.SendAndClose(&pb.UploadSummary{
                TotalImages: count,
                TotalBytes:  totalBytes,
                UploadedFilenames: filenames,
            })
        }
        if err != nil {
            return err
        }

        // Process chunk
        totalBytes += int64(len(chunk.Data))
        if chunk.ChunkNumber == 0 {
            count++
            filenames = append(filenames, chunk.Filename)
        }
    }
}

2. 集約

クライアントが複数のデータポイントを送信し、サーバーが集約します。

service AnalyticsService {
  rpc RecordMetrics(stream Metric) returns (AggregateResult);
}

message Metric {
  string name = 1;
  double value = 2;
  google.protobuf.Timestamp timestamp = 3;
}

双方向ストリーミングパターン

1. チャットアプリケーション

リアルタイムの双方向通信です。

service ChatService {
  rpc Chat(stream ChatMessage) returns (stream ChatMessage);
}

message ChatMessage {
  string user_id = 1;
  string room_id = 2;
  string content = 3;
  google.protobuf.Timestamp timestamp = 4;
}

実装（Go）：

func (s *server) Chat(stream pb.ChatService_ChatServer) error {
    // Create channel for this client
    clientID := uuid.New().String()
    msgChan := make(chan *pb.ChatMessage, 10)

    // Register client
    s.mu.Lock()
    s.clients[clientID] = msgChan
    s.mu.Unlock()

    defer func() {
        s.mu.Lock()
        delete(s.clients, clientID)
        close(msgChan)
        s.mu.Unlock()
    }()

    // Goroutine to send messages to client
    go func() {
        for msg := range msgChan {
            if err := stream.Send(msg); err != nil {
                return
            }
        }
    }()

    // Receive messages from client
    for {
        msg, err := stream.Recv()
        if err == io.EOF {
            return nil
        }
        if err != nil {
            return err
        }

        // Broadcast to all clients in room
        s.broadcast(msg)
    }
}

2. ライブコラボレーション

リアルタイムのドキュメント編集です。

service CollaborationService {
  rpc Collaborate(stream DocumentEdit) returns (stream DocumentEdit);
}

message DocumentEdit {
  string document_id = 1;
  string user_id = 2;
  int32 position = 3;
  string content = 4;
  enum Operation {
    OPERATION_UNSPECIFIED = 0;
    OPERATION_INSERT = 1;
    OPERATION_DELETE = 2;
    OPERATION_UPDATE = 3;
  }
  Operation operation = 5;
}

3. ゲーム状態の同期

マルチプレイヤーゲームの更新をリアルタイムで行います。

service GameService {
  rpc PlayGame(stream GameAction) returns (stream GameState);
}

message GameAction {
  string player_id = 1;
  string game_id = 2;
  string action_type = 3;
  bytes action_data = 4;
}

message GameState {
  string game_id = 1;
  repeated PlayerState players = 2;
  bytes world_state = 3;
  google.protobuf.Timestamp timestamp = 4;
}

インターセプター（ミドルウェア）

インターセプターを使用すると、gRPC サービスに横断的関心事を追加できます。

単項インターセプター

サーバー側の単項インターセプター：

func UnaryServerInterceptor() grpc.UnaryServerInterceptor {
    return func(
        ctx context.Context,
        req interface{},
        info *grpc.UnaryServerInfo,
        handler grpc.UnaryHandler,
    ) (interface{}, error) {
        // Pre-processing
        start := time.Now()
        log.Printf("Method: %s, Start: %v", info.FullMethod, start)

        // Call the handler
        resp, err := handler(ctx, req)

        // Post-processing
        duration := time.Since(start)
        log.Printf("Method: %s, Duration: %v, Error: %v",
            info.FullMethod, duration, err)

        return resp, err
    }
}

// Usage
server := grpc.NewServer(
    grpc.UnaryInterceptor(UnaryServerInterceptor()),
)

クライアント側の単項インターセプター：

func UnaryClientInterceptor() grpc.UnaryClientInterceptor {
    return func(
        ctx context.Context,
        method string,
        req, reply interface{},
        cc *grpc.ClientConn,
        invoker grpc.UnaryInvoker,
        opts ...grpc.CallOption,
    ) error {
        start := time.Now()

        // Call the remote method
        err := invoker(ctx, method, req, reply, cc, opts...)

        log.Printf("Method: %s, Duration: %v, Error: %v",
            method, time.Since(start), err)

        return err
    }
}

// Usage
conn, err := grpc.Dial(
    address,
    grpc.WithUnaryInterceptor(UnaryClientInterceptor()),
)

ストリーミングインターセプター

サーバー側のストリームインターセプター：

func StreamServerInterceptor() grpc.StreamServerInterceptor {
    return func(
        srv interface{},
        ss grpc.ServerStream,
        info *grpc.StreamServerInfo,
        handler grpc.StreamHandler,
    ) error {
        log.Printf("Stream started: %s", info.FullMethod)

        err := handler(srv, ss)

        log.Printf("Stream ended: %s, Error: %v", info.FullMethod, err)

        return err
    }
}

一般的なインターセプターパターン

1. 認証インターセプター

func AuthInterceptor(secret string) grpc.UnaryServerInterceptor {
    return func(
        ctx context.Context,
        req interface{},
        info *grpc.UnaryServerInfo,
        handler grpc.UnaryHandler,
    ) (interface{}, error) {
        // Extract metadata
        md, ok := metadata.FromIncomingContext(ctx)
        if !ok {
            return nil, status.Error(codes.Unauthenticated, "no metadata")
        }

        // Get authorization token
        tokens := md["authorization"]
        if len(tokens) == 0 {
            return nil, status.Error(codes.Unauthenticated, "no token")
        }

        // Validate token
        token := tokens[0]
        claims, err := validateJWT(token, secret)
        if err != nil {
            return nil, status.Error(codes.Unauthenticated, "invalid token")
        }

        // Add claims to context
        ctx = context.WithValue(ctx, "claims", claims)

        return handler(ctx, req)
    }
}

2. ロギングインターセプター

func LoggingInterceptor(logger *log.Logger) grpc.UnaryServerInterceptor {
    return func(
        ctx context.Context,
        req interface{},
        info *grpc.UnaryServerInfo,
        handler grpc.UnaryHandler,
    ) (interface{}, error) {
        start := time.Now()

        // Get request ID from metadata
        requestID := getRequestID(ctx)

        logger.Printf("[%s] Request: %s", requestID, info.FullMethod)

        resp, err := handler(ctx, req)

        duration := time.Since(start)
        statusCode := status.Code(err)

        logger.Printf("[%s] Response: %s, Duration: %v, Status: %v",
            requestID, info.FullMethod, duration, statusCode)

        return resp, err
    }
}

3. レート制限インターセプター

func RateLimitInterceptor(limiter *rate.Limiter) grpc.UnaryServerInterceptor {
    return func(
        ctx context.Context,
        req interface{},
        info *grpc.UnaryServerInfo,
        handler grpc.UnaryHandler,
    ) (interface{}, error) {
        if !limiter.Allow() {
            return nil, status.Error(
                codes.ResourceExhausted,
                "rate limit exceeded",
            )
        }

        return handler(ctx, req)
    }
}

4. トレーシングインターセプター（OpenTelemetry）

func TracingInterceptor(tracer trace.Tracer) grpc.UnaryServerInterceptor {
    return func(
        ctx context.Context,
        req interface{},
        info *grpc.UnaryServerInfo,
        handler grpc.UnaryHandler,
    ) (interface{}, error) {
        ctx, span := tracer.Start(ctx, info.FullMethod)
        defer span.End()

        // Add attributes
        span.SetAttributes(
            attribute.String("rpc.method", info.FullMethod),
            attribute.String("rpc.service", "MyService"),
        )

        resp, err := handler(ctx, req)

        if err != nil {
            span.RecordError(err)
            span.SetStatus(codes2.Error, err.Error())
        } else {
            span.SetStatus(codes2.Ok, "")
        }

        return resp, err
    }
}

5. エラーリカバリーインターセプター

func RecoveryInterceptor() grpc.UnaryServerInterceptor {
    return func(
        ctx context.Context,
        req interface{},
        info *grpc.UnaryServerInfo,
        handler grpc.UnaryHandler,
    ) (resp interface{}, err error) {
        defer func() {
            if r := recover(); r != nil {
                log.Printf("Panic recovered: %v\n%s", r, debug.Stack())
                err = status.Error(codes.Internal, "internal server error")
            }
        }()

        return handler(ctx, req)
    }
}

複数のインターセプターの連結

server := grpc.NewServer(
    grpc.ChainUnaryInterceptor(
        RecoveryInterceptor(),
        LoggingInterceptor(logger),
        TracingInterceptor(tracer),
        AuthInterceptor(jwtSecret),
        RateLimitInterceptor(limiter),
    ),
    grpc.ChainStreamInterceptor(
        StreamRecoveryInterceptor(),
        StreamLoggingInterceptor(logger),
    ),
)

負荷分散

クライアント側の負荷分散

gRPC には、複数のポリシーによるクライアント側の負荷分散機能が組み込まれています。

1. ラウンドロビン

import "google.golang.org/grpc/balancer/roundrobin"

conn, err := grpc.Dial(
    "dns:///my-service.example.com:50051",
    grpc.WithDefaultServiceConfig(`{"loadBalancingPolicy":"round_robin"}`),
    grpc.WithInsecure(),
)

2. 最初の接続先を選択（デフォルト）

conn, err := grpc.Dial(
    "dns:///my-service.example.com:50051",
    grpc.WithDefaultServiceConfig(`{"loadBalancingPolicy":"pick_first"}`),
    grpc.WithInsecure(),
)

3. カスタムリゾルバー

カスタムのサービスディスカバリーを実装します。

type exampleResolver struct {
    target     resolver.Target
    cc         resolver.ClientConn
    addrsStore map[string][]string
}

func (r *exampleResolver) ResolveNow(resolver.ResolveNowOptions) {
    // Discover service addresses
    addresses := r.discoverServices()

    var addrs []resolver.Address
    for _, addr := range addresses {
        addrs = append(addrs, resolver.Address{Addr: addr})
    }

    r.cc.UpdateState(resolver.State{Addresses: addrs})
}

func init() {
    resolver.Register(&exampleResolverBuilder{})
}

サービスメッシュによる負荷分散

サービスメッシュを使用した Kubernetes（Istio/Linkerd）：

apiVersion: v1
kind: Service
metadata:
  name: grpc-service
spec:
  selector:
    app: grpc-app
  ports:
  - name: grpc
    port: 50051
    targetPort: 50051
    protocol: TCP
  type: ClusterIP
---
apiVersion: networking.istio.io/v1alpha3
kind: DestinationRule
metadata:
  name: grpc-service
spec:
  host: grpc-service
  trafficPolicy:
    loadBalancer:
      simple: ROUND_ROBIN
    connectionPool:
      http:
        http2MaxRequests: 1000
        maxRequestsPerConnection: 10

ヘルスチェック

ヘルスチェックサービスを実装します。

service Health {
  rpc Check(HealthCheckRequest) returns (HealthCheckResponse);
  rpc Watch(HealthCheckRequest) returns (stream HealthCheckResponse);
}

message HealthCheckRequest {
  string service = 1;
}

message HealthCheckResponse {
  enum ServingStatus {
    UNKNOWN = 0;
    SERVING = 1;
    NOT_SERVING = 2;
    SERVICE_UNKNOWN = 3;
  }
  ServingStatus status = 1;
}

実装：

import "google.golang.org/grpc/health"
import healthpb "google.golang.org/grpc/health/grpc_health_v1"

healthServer := health.NewServer()
healthpb.RegisterHealthServer(grpcServer, healthServer)

// Set service status
healthServer.SetServingStatus("UserService", healthpb.HealthCheckResponse_SERVING)

エラー処理

gRPC ステータスコード

gRPC はエラー処理に標準化されたステータスコードを使用します。

import "google.golang.org/grpc/codes"
import "google.golang.org/grpc/status"

// Return errors with appropriate codes
func (s *server) GetUser(ctx context.Context, req *pb.GetUserRequest) (*pb.User, error) {
    if req.Id <= 0 {
        return nil, status.Error(codes.InvalidArgument, "id must be positive")
    }

    user, err := s.db.GetUser(req.Id)
    if err == sql.ErrNoRows {
        return nil, status.Error(codes.NotFound, "user not found")
    }
    if err != nil {
        return nil, status.Error(codes.Internal, "database error")
    }

    return user, nil
}

一般的なステータスコード：

• OK：成功
• Canceled：操作がキャンセルされました
• Unknown：不明なエラー
• InvalidArgument：クライアントが無効な引数を指定しました
• DeadlineExceeded：操作の完了前に期限が切れました
• NotFound：エンティティが見つかりません
• AlreadyExists：エンティティはすでに存在します
• PermissionDenied：アクセスが拒否されました
• ResourceExhausted：リソースを使い果たしました（レート制限）
• FailedPrecondition：操作が拒否されました（システムが有効な状態ではありません）
• Aborted：操作が中断されました
• OutOfRange：有効な範囲外です
• Unimplemented：操作は実装されていません
• Internal：サーバー内部エラー
• Unavailable：サービスを利用できません
• DataLoss：回復不能なデータ損失
• Unauthenticated：リクエストに有効な認証情報がありません

詳細なエラー情報

構造化されたエラー情報を追加します。

import "google.golang.org/genproto/googleapis/rpc/errdetails"

func (s *server) CreateUser(ctx context.Context, req *pb.CreateUserRequest) (*pb.User, error) {
    // Validate request
    violations := validateCreateUserRequest(req)
    if len(violations) > 0 {
        badRequest := &errdetails.BadRequest{}
        for field, msg := range violations {
            badRequest.FieldViolations = append(
                badRequest.FieldViolations,
                &errdetails.BadRequest_FieldViolation{
                    Field:       field,
                    Description: msg,
                },
            )
        }

        st := status.New(codes.InvalidArgument, "invalid request")
        st, _ = st.WithDetails(badRequest)
        return nil, st.Err()
    }

    // Create user...
}

クライアント側のエラー処理：

resp, err := client.CreateUser(ctx, req)
if err != nil {
    st := status.Convert(err)

    for _, detail := range st.Details() {
        switch t := detail.(type) {
        case *errdetails.BadRequest:
            for _, violation := range t.FieldViolations {
                fmt.Printf("Invalid field %s: %s\n",
                    violation.Field, violation.Description)
            }
        }
    }
}

エラーの伝播

func (s *server) ProcessOrder(ctx context.Context, req *pb.OrderRequest) (*pb.OrderResponse, error) {
    // Call inventory service
    inventory, err := s.inventoryClient.CheckInventory(ctx, &pb.InventoryRequest{
        ProductId: req.ProductId,
    })
    if err != nil {
        // Propagate error with additional context
        st := status.Convert(err)
        return nil, status.Errorf(st.Code(),
            "inventory check failed: %v", st.Message())
    }

    // Continue processing...
}

リトライロジック

import "google.golang.org/grpc/codes"
import "google.golang.org/grpc/status"

func CallWithRetry(ctx context.Context, maxRetries int, fn func() error) error {
    var err error

    for i := 0; i < maxRetries; i++ {
        err = fn()
        if err == nil {
            return nil
        }

        // Check if error is retryable
        st := status.Convert(err)
        if !isRetryable(st.Code()) {
            return err
        }

        // Exponential backoff
        backoff := time.Duration(math.Pow(2, float64(i))) * time.Second
        time.Sleep(backoff)
    }

    return err
}

func isRetryable(code codes.Code) bool {
    return code == codes.Unavailable ||
           code == codes.DeadlineExceeded ||
           code == codes.ResourceExhausted
}

ベストプラクティス

1. スキーマの進化

推奨事項：

• 常に syntax = "proto3" を使用します
• フィールド番号を再利用しないでください
• 非推奨のフィールドには reserved を使用します
• 新しいフィールドには新しい番号を割り当てます
• NULL 許容フィールドには optional ラッパーを使用します

message User {
  int32 id = 1;
  string name = 2;
  // string age = 3;  // Deprecated
  reserved 3;
  reserved "age";

  string email = 4;
  google.protobuf.Int32Value phone = 5;  // Optional
}

禁止事項：

• フィールドの型を変更しないでください
• フィールド番号を再利用しないでください
• 番号を予約せずにフィールドを削除しないでください
• エイリアスなしでメッセージ名を変更しないでください

2. パフォーマンスの最適化

接続管理：

// Reuse connections
var conn *grpc.ClientConn
var once sync.Once

func getConnection() *grpc.ClientConn {
    once.Do(func() {
        var err error
        conn, err = grpc.Dial(
            address,
            grpc.WithKeepaliveParams(keepalive.ClientParameters{
                Time:                10 * time.Second,
                Timeout:             3 * time.Second,
                PermitWithoutStream: true,
            }),
        )
        if err != nil {
            log.Fatal(err)
        }
    })
    return conn
}

コネクションプーリング：

type ConnectionPool struct {
    connections []*grpc.ClientConn
    next        uint32
}

func (p *ConnectionPool) GetConnection() *grpc.ClientConn {
    n := atomic.AddUint32(&p.next, 1)
    return p.connections[n%uint32(len(p.connections))]
}

大容量データのストリーミング：

// Instead of this:
rpc GetAllUsers(Empty) returns (UserList);  // Large response

// Use this:
rpc ListUsers(ListUsersRequest) returns (stream User);  // Streamed

3. セキュリティのベストプラクティス

TLS の設定：

// Server-side
creds, err := credentials.NewServerTLSFromFile(certFile, keyFile)
server := grpc.NewServer(grpc.Creds(creds))

// Client-side
creds, err := credentials.NewClientTLSFromFile(certFile, "")
conn, err := grpc.Dial(address, grpc.WithTransportCredentials(creds))

相互 TLS（mTLS）：

cert, err := tls.LoadX509KeyPair(certFile, keyFile)
certPool := x509.NewCertPool()
ca, err := ioutil.ReadFile(caFile)
certPool.AppendCertsFromPEM(ca)

creds := credentials.NewTLS(&tls.Config{
    Certificates: []tls.Certificate{cert},
    ClientAuth:   tls.RequireAndVerifyClientCert,
    ClientCAs:    certPool,
})

server := grpc.NewServer(grpc.Creds(creds))

トークン認証：

type tokenAuth struct {
    token string
}

func (t tokenAuth) GetRequestMetadata(ctx context.Context, uri ...string) (map[string]string, error) {
    return map[string]string{
        "authorization": "Bearer " + t.token,
    }, nil
}

func (t tokenAuth) RequireTransportSecurity() bool {
    return true
}

// Usage
conn, err := grpc.Dial(
    address,
    grpc.WithPerRPCCredentials(tokenAuth{token: "my-token"}),
)

4. タイムアウトと期限の管理

// Set deadline for request
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()

resp, err := client.GetUser(ctx, &pb.GetUserRequest{Id: 123})
if err != nil {
    if status.Code(err) == codes.DeadlineExceeded {
        log.Println("Request timed out")
    }
}

サーバー側の期限伝播：

func (s *server) ComplexOperation(ctx context.Context, req *pb.Request) (*pb.Response, error) {
    // Check if deadline is already exceeded
    deadline, ok := ctx.Deadline()
    if ok && time.Now().After(deadline) {
        return nil, status.Error(codes.DeadlineExceeded, "deadline exceeded")
    }

    // Propagate context to downstream calls
    user, err := s.userClient.GetUser(ctx, &pb.GetUserRequest{Id: req.UserId})
    if err != nil {
        return nil, err
    }

    // Continue with remaining time...
}

5. 監視と可観測性

Prometheus メトリクス：

import "github.com/grpc-ecosystem/go-grpc-prometheus"

// Server metrics
grpcMetrics := grpc_prometheus.NewServerMetrics()
server := grpc.NewServer(
    grpc.UnaryInterceptor(grpcMetrics.UnaryServerInterceptor()),
    grpc.StreamInterceptor(grpcMetrics.StreamServerInterceptor()),
)
grpcMetrics.InitializeMetrics(server)

// Expose metrics
http.Handle("/metrics", promhttp.Handler())

6. グレースフルシャットダウン

server := grpc.NewServer()
// Register services...

go func() {
    if err := server.Serve(listener); err != nil {
        log.Fatalf("failed to serve: %v", err)
    }
}()

// Wait for interrupt signal
quit := make(chan os.Signal, 1)
signal.Notify(quit, os.Interrupt, syscall.SIGTERM)
<-quit

log.Println("Shutting down server...")

// Graceful shutdown
server.GracefulStop()
log.Println("Server stopped")

7. サービスのバージョニング

URL ベースのバージョニング：

package api.v1;

service UserServiceV1 {
  rpc GetUser(GetUserRequest) returns (User);
}

package api.v2;

service UserServiceV2 {
  rpc GetUser(GetUserRequest) returns (User);
}

フィールドベースのバージョニング：

message User {
  int32 id = 1;
  string name = 2;
  string email = 3;

  // v2 additions
  string phone = 4;
  Address address = 5;
}

8. テストのベストプラクティス

モックを使った単体テスト：

type mockUserClient struct {
    pb.UserServiceClient
    getUserFunc func(ctx context.Context, req *pb.GetUserRequest) (*pb.User, error)
}

func (m *mockUserClient) GetUser(ctx context.Context, req *pb.GetUserRequest, opts ...grpc.CallOption) (*pb.User, error) {
    return m.getUserFunc(ctx, req)
}

func TestOrderService(t *testing.T) {
    mockClient := &mockUserClient{
        getUserFunc: func(ctx context.Context, req *pb.GetUserRequest) (*pb.User, error) {
            return &pb.User{Id: 1, Name: "Test User"}, nil
        },
    }

    // Test with mock...
}

統合テスト：

func TestIntegration(t *testing.T) {
    // Start test server
    lis, err := net.Listen("tcp", ":0")
    require.NoError(t, err)

    server := grpc.NewServer()
    pb.RegisterUserServiceServer(server, &userServer{})

    go server.Serve(lis)
    defer server.Stop()

    // Connect client
    conn, err := grpc.Dial(lis.Addr().String(), grpc.WithInsecure())
    require.NoError(t, err)
    defer conn.Close()

    client := pb.NewUserServiceClient(conn)

    // Test requests...
}

本番環境へのデプロイパターン

Docker のデプロイ

Dockerfile：

FROM golang:1.21-alpine AS builder

WORKDIR /app
COPY go.mod go.sum ./
RUN go mod download

COPY . .
RUN CGO_ENABLED=0 GOOS=linux go build -o server ./cmd/server

FROM alpine:latest
RUN apk --no-cache add ca-certificates
WORKDIR /root/

COPY --from=builder /app/server .
COPY --from=builder /app/proto ./proto

EXPOSE 50051

CMD ["./server"]

Kubernetes のデプロイ

deployment.yaml：

apiVersion: apps/v1
kind: Deployment
metadata:
  name: grpc-service
spec:
  replicas: 3
  selector:
    matchLabels:
      app: grpc-service
  template:
    metadata:
      labels:
        app: grpc-service
    spec:
      containers:
      - name: grpc-service
        image: grpc-service:latest
        ports:
        - containerPort: 50051
          name: grpc
          protocol: TCP
        env:
        - name: PORT
          value: "50051"
        livenessProbe:
          exec:
            command: ["/bin/grpc_health_probe", "-addr=:50051"]
          initialDelaySeconds: 10
        readinessProbe:
          exec:
            command: ["/bin/grpc_health_probe", "-addr=:50051"]
          initialDelaySeconds: 5
        resources:
          requests:
            memory: "128Mi"
            cpu: "100m"
          limits:
            memory: "512Mi"
            cpu: "500m"
---
apiVersion: v1
kind: Service
metadata:
  name: grpc-service
spec:
  selector:
    app: grpc-service
  ports:
  - port: 50051
    targetPort: 50051
    protocol: TCP
    name: grpc
  type: ClusterIP

サービスメッシュとの統合（Istio）

トラフィックルーティング用の VirtualService：

apiVersion: networking.istio.io/v1alpha3
kind: VirtualService
metadata:
  name: grpc-service
spec:
  hosts:
  - grpc-service
  http:
  - match:
    - headers:
        version:
          exact: v2
    route:
    - destination:
        host: grpc-service
        subset: v2
  - route:
    - destination:
        host: grpc-service
        subset: v1

よく使われるパターンとアンチパターン

✅ 実施すること：

1. 大規模なデータセットにはストリーミングを使用する
2. ステータスコードを使って適切なエラー処理を実装する
3. 横断的関心事にはインターセプターを追加する
4. スループットの高いクライアントにはコネクションプーリングを使用する
5. ヘルスチェックを実装する
6. 適切なタイムアウトと期限を設定する
7. 本番環境では TLS を使用する
8. API をバージョニングする
9. メトリクスとトレーシングで監視する
10. 統合テストでテストする

❌ 避けること：

1. 大規模なデータセットに単項 RPC を使用しない - 代わりにストリーミングを使用する
2. コンテキストのキャンセルを無視しない - 必ず context.Done() を確認する
3. リクエストごとに新しい接続を作成しない - 接続を再利用する
4. 認証／認可を省略しない - 必ず検証する
5. グレースフルシャットダウンを忘れない - SIGTERM を適切に処理する
6. エンドポイントをハードコードしない - サービスディスカバリーを使用する
7. エラーを無視しない - すべてのエラーケースを処理する
8. タイムアウトなしでブロッキング処理を使用しない - 必ず期限を設定する
9. ヘルスチェックを省略しない - ライブネス／レディネスプローブを実装する
10. 監視なしでデプロイしない - メトリクスとログを追加する

---

スキルのバージョン：1.0.0 最終更新日：October 2025 スキルのカテゴリ：マイクロサービス、gRPC、分散システム、API 設計 対応言語：Go、Python、Node.js、Java、C++、C#、Ruby など