gRPC-Secret: 期限、タイムアウト、カスタムコンテキストをマスターする

オープンソースのリモート プロシージャ コール (RPC) フレームワークである gRPC は、サービス間の効率的でスケーラブルな通信を可能にします。gRPC の重要な側面の 1 つは、期限、要求タイムアウト、およびカスタム構造を含むコンテキストの伝播の管理です。

これらのメカニズムを理解することで、サービスが迅速に応答し、合理的な時間枠を超える操作でリソースが無駄に消費されず、カスタム メタデータが効果的に送信されるようになります。

期限とリクエストタイムアウトを理解する

締め切り

gRPC の期限は、操作を完了しなければならない最大時間を指定します。この時間内に操作が完了しない場合、操作は自動的に終了します。期限は、応答しないサービスや遅いサービスが原因でシステム リソースが無期限に拘束されないようにするために不可欠です。

リクエストタイムアウト

要求タイムアウトとは、クライアントがサーバーからの応答を待機する期間です。サーバーがこの期間内に応答しない場合、要求は中止されます。このメカニズムにより、クライアントが応答を無期限に待機し続けることが防止されます。

gRPC での期限とリクエストのタイムアウトの設定

gRPC は、クライアント側とサーバー側の両方で期限を設定したりタイムアウトを要求したりするための柔軟なオプションを提供します。Go でこれを行う方法は次のとおりです。

クライアント側の期限設定

import ( "context" "log" "time" "google.golang.org/grpc" pb "path/to/your/protobuf/package" ) func main() { conn, err := grpc.Dial("server_address", grpc.WithInsecure()) if err != nil { log.Fatalf("did not connect: %v", err) } defer conn.Close() client := pb.NewYourServiceClient(conn) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() resp, err := client.YourMethod(ctx, &pb.YourRequest{}) if err != nil { log.Fatalf("could not call method: %v", err) } log.Printf("Response: %v", resp) }

サーバー側の処理

サーバー側では、gRPC を使用すると期限を強制し、クライアントが指定した期限を超えたシナリオを処理できます。

 import ( "context" "log" "net" "time" "google.golang.org/grpc" pb "path/to/your/protobuf/package" ) type server struct { pb.UnimplementedYourServiceServer } func (s *server) YourMethod(ctx context.Context, req *pb.YourRequest) (*pb.YourResponse, error) { select { case <-time.After(10 * time.Second): return &pb.YourResponse{}, nil case <-ctx.Done(): return nil, ctx.Err() } } func main() { lis, err := net.Listen("tcp", ":50051") if err != nil { log.Fatalf("failed to listen: %v", err) } s := grpc.NewServer() pb.RegisterYourServiceServer(s, &server{}) if err := s.Serve(lis); err != nil { log.Fatalf("failed to serve: %v", err) } }

コンテキスト内でのカスタム構造の伝播

gRPC でコンテキストを介してカスタム構造を送信するには、データをコンテキストにアタッチする前にシリアル化し、受信側でデシリアル化する必要があります。これには、カスタム構造を JSON やプロトコル バッファーなどのネットワーク経由で送信できる形式に変換し、このシリアル化されたデータをコンテキスト メタデータに追加することが含まれます。

ステップバイステップのプロセス

1. カスタム構造を定義する: 送信するカスタム構造を定義します。
2. 構造をシリアル化する: カスタム構造を文字列またはバイト配列に変換します。
3. コンテキストにアタッチ: シリアル化されたデータをコンテキスト メタデータに追加します。
4. 送信: コンテキストとともに gRPC 呼び出しを送信します。
5. サーバー上で抽出および逆シリアル化: サーバー側のコンテキストからメタデータを抽出し、カスタム構造に逆シリアル化します。

ステップ1: カスタム構造を定義する

type CustomStruct struct { Field1 string Field2 int }

ステップ2: 構造をシリアル化する

import ( "context" "encoding/json" "fmt" "google.golang.org/grpc/metadata" ) func serializeCustomStruct(customStruct CustomStruct) (string, error) { data, err := json.Marshal(customStruct) if err != nil { return "", err } return string(data), nil }

ステップ3: コンテキストに関連付ける

func attachCustomStructToContext(ctx context.Context, customStruct CustomStruct) (context.Context, error) { serializedData, err := serializeCustomStruct(customStruct) if err != nil { return nil, err } md := metadata.Pairs("custom-struct", serializedData) ctx = metadata.NewOutgoingContext(ctx, md) return ctx, nil }

ステップ4: 送信

func main() { conn, err := grpc.Dial("localhost:50051", grpc.WithInsecure()) if err != nil { log.Fatalf("did not connect: %v", err) } defer conn.Close() client := pb.NewYourServiceClient(conn) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() customStruct := CustomStruct{Field1: "value1", Field2: 42} ctx, err = attachCustomStructToContext(ctx, customStruct) if err != nil { log.Fatalf("could not attach custom struct to context: %v", err) } resp, err := client.YourMethod(ctx, &pb.YourRequest{}) if err != nil { log.Fatalf("could not call method: %v", err) } log.Printf("Response: %v", resp) }

ステップ5: サーバー上で抽出してデシリアライズする

func deserializeCustomStruct(data string) (CustomStruct, error) { var customStruct CustomStruct err := json.Unmarshal([]byte(data), &customStruct) if err != nil { return CustomStruct{}, err } return customStruct, nil } func extractCustomStructFromContext(ctx context.Context) (CustomStruct, error) { md, ok := metadata.FromIncomingContext(ctx) if !ok { return CustomStruct{}, fmt.Errorf("no metadata found in context") } serializedData := md["custom-struct"] if len(serializedData) == 0 { return CustomStruct{}, fmt.Errorf("no custom struct found in metadata") } return deserializeCustomStruct(serializedData[0]) } func (s *server) YourMethod(ctx context.Context, req *pb.YourRequest) (*pb.YourResponse, error) { customStruct, err := extractCustomStructFromContext(ctx) if err != nil { return nil, err } log.Printf("Received custom struct: %+v", customStruct) select { case <-time.After(10 * time.Second): return &pb.YourResponse{}, nil case <-ctx.Done(): return nil, ctx.Err() } }

すべての gRPC 呼び出し用のミドルウェアの実装

カスタム構造を含むコンテキストの伝播をすべての gRPC 呼び出しにわたって一貫して処理するには、インターセプターを使用できます。インターセプターは、リクエストとレスポンスを処理し、ログ記録、監視、コンテキスト メタデータの処理などの機能を追加するミドルウェアです。

単項インターセプターとストリーミングインターセプター

さまざまな種類の RPC 呼び出しを管理するには、単項インターセプターとストリーミング インターセプターの両方が必要です。

• 単項インターセプター: 単一の要求応答サイクルを処理します。

• ストリーミング インターセプター: クライアント側ストリーミング、サーバー側ストリーミング、双方向ストリーミングなどの要求と応答のストリームを処理します。

単項インターセプターの実装

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

 func unaryClientInterceptor( ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption, ) error { customStruct, ok := ctx.Value("customStruct").(CustomStruct) if ok { ctx, err := attachCustomStructToContext(ctx, customStruct) if err != nil { return err } } return invoker(ctx, method, req, reply, cc, opts...) }

サーバー側単項インターセプター:

 func unaryServerInterceptor( ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler, ) (interface{}, error) { customStruct, err := extractCustomStructFromContext(ctx) if err != nil { return nil, err } ctx = context.WithValue(ctx, "customStruct", customStruct) return handler(ctx, req) }

ストリーミングインターセプターの実装

クライアント側ストリーミングインターセプター:

 func streamClientInterceptor( ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption, ) (grpc.ClientStream, error) { customStruct, ok := ctx.Value("customStruct").(CustomStruct) if ok { ctx, err := attachCustomStructToContext(ctx, customStruct) if err != nil { return nil, err } } return

サーバー側ストリーミングインターセプター:

 import ( "context" "google.golang.org/grpc" "google.golang.org/grpc/metadata" ) // StreamServerInterceptor handles server-side streaming func streamServerInterceptor( srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler, ) error { ctx := ss.Context() customStruct, err := extractCustomStructFromContext(ctx) if err != nil { return err } // Add custom struct to context for server handling newCtx := context.WithValue(ctx, "customStruct", customStruct) wrapped := grpc_middleware.WrapServerStream(ss) wrapped.WrappedContext = newCtx // Handle the request return handler(srv, wrapped) } // Example of using the interceptor in a gRPC server setup func main() { lis, err := net.Listen("tcp", ":50051") if err != nil { log.Fatalf("failed to listen: %v", err) } // Register the interceptors server := grpc.NewServer( grpc.UnaryInterceptor(unaryServerInterceptor), grpc.StreamInterceptor(streamServerInterceptor), ) // Register your gRPC service implementations here pb.RegisterYourServiceServer(server, &yourServiceServer{}) if err := server.Serve(lis); err != nil { log.Fatalf("failed to serve: %v", err) } }

単項インターセプターとストリーミング インターセプターを作成して登録することで、カスタム構造を含むコンテキストの伝播がすべての gRPC 呼び出しで一貫して処理されることを保証できます。このアプローチにより、カスタム メタデータが適切に管理および伝播され、堅牢で柔軟な gRPC サービスを構築できるようになります。