Cours 14 : gRPC avec Go

1. Introduction à gRPC

gRPC est un framework RPC haute performance développé par Google. Il utilise Protocol Buffers pour la sérialisation et HTTP/2 pour le transport.

1.1 Pourquoi gRPC ?

1.2 Structure d'un service gRPC

// proto/user/v1/user.proto
syntax = "proto3";

package user.v1;

option go_package = "github.com/example/gen/go/user/v1;userv1";

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

// Message
message User {
  string id = 1;
  string name = 2;
  string email = 3;
  repeated string roles = 4;
  google.protobuf.Timestamp created_at = 5;
  google.protobuf.StringValue nickname = 6; // Optional
}

message GetUserRequest {
  string id = 1;
}

message GetUserResponse {
  User user = 1;
}

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

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

message CreateUserRequest {
  string name = 1;
  string email = 2;
  repeated string roles = 3;
}

message CreateUserResponse {
  User user = 1;
}

message UpdateUserRequest {
  User user = 1;
  repeated string update_mask = 2; // Field mask
}

message DeleteUserRequest {
  string id = 1;
}

// Service
service UserService {
  // Unary
  rpc GetUser(GetUserRequest) returns (GetUserResponse);
  rpc ListUsers(ListUsersRequest) returns (ListUsersResponse);
  rpc CreateUser(CreateUserRequest) returns (CreateUserResponse);
  rpc UpdateUser(UpdateUserRequest) returns (UpdateUserResponse);
  rpc DeleteUser(DeleteUserRequest) returns (google.protobuf.Empty);

  // Server streaming
  rpc StreamUsers(ListUsersRequest) returns (stream User);
  
  // Client streaming
  rpc BatchCreateUsers(stream CreateUserRequest) returns (CreateUserResponse);
  
  // Bidirectional streaming
  rpc Chat(stream ChatMessage) returns (stream ChatMessage);
}

2. Protocol Buffers (proto3)

2.1 Types de données

syntax = "proto3";

message AllTypes {
  // Scalaires
  double   double_field = 1;
  float    float_field = 2;
  int32    int32_field = 3;
  int64    int64_field = 4;
  uint32   uint32_field = 5;
  uint64   uint64_field = 6;
  sint32   sint32_field = 7;
  sint64   sint64_field = 8;
  fixed32  fixed32_field = 9;
  fixed64  fixed64_field = 10;
  sfixed32 sfixed32_field = 11;
  sfixed64 sfixed64_field = 12;
  bool     bool_field = 13;
  string   string_field = 14;
  bytes    bytes_field = 15;

  // Complexes
  repeated string repeated_field = 16;  // Array
  map<string, int32> map_field = 17;    // Map
  
  // Optionnel (proto3)
  optional string optional_field = 18;
  
  // Enum
  enum Status {
    STATUS_UNSPECIFIED = 0;
    STATUS_ACTIVE = 1;
    STATUS_INACTIVE = 2;
  }
  Status status = 19;

  // Oneof (union)
  oneof contact {
    string email = 20;
    string phone = 21;
  }

  // Any (type dynamique)
  google.protobuf.Any metadata = 22;

  // Duration, Timestamp
  google.protobuf.Duration ttl = 23;
  google.protobuf.Timestamp last_updated = 24;
}

2.2 Oneof et Any

message Event {
  string id = 1;
  string type = 2;
  google.protobuf.Timestamp timestamp = 3;
  
  oneof payload {
    OrderCreated order_created = 4;
    OrderPaid order_paid = 5;
    OrderShipped order_shipped = 6;
  }
}

// Utilisation de Any pour des types dynamiques
message DynamicEvent {
  string id = 1;
  string type = 2;
  google.protobuf.Any data = 3;
}

// Well-known types
import "google/protobuf/wrappers.proto";
import "google/protobuf/empty.proto";
import "google/protobuf/struct.proto";
import "google/protobuf/field_mask.proto";
import "google/protobuf/api.proto";

2.3 Options proto

syntax = "proto3";

package order.v1;

// Options de package
option go_package = "github.com/example/gen/go/order/v1;orderv1";
option java_package = "com.example.order.v1";
option csharp_namespace = "Example.Order.V1";

// Options de message
message Order {
  option deprecated = true;
  string id = 1 [(validate.rules).string.uuid = true];
  string user_id = 2 [(google.api.field_behavior) = REQUIRED];
}

// Options de champ
message CreateOrderRequest {
  string user_id = 1 [(validate.rules).string.min_len = 1];
  double total = 2 [(validate.rules).double.gte = 0];
  repeated OrderItem items = 3 [(validate.rules).repeated.min_items = 1];
}

message OrderItem {
  string product_id = 1;
  int32 quantity = 2 [(validate.rules).int32.gte = 1];
  double price = 3 [(validate.rules).double.gt = 0];
}

// Custom options
import "google/protobuf/descriptor.proto";

extend google.protobuf.FieldOptions {
  bool redact = 50000;
}

message User {
  string id = 1;
  string email = 2 [(redact) = true]; // Redacted in logs
  string password = 3 [(redact) = true];
}

3. Code Generation

3.1 Installation

# Installer protoc
# https://github.com/protocolbuffers/protobuf/releases

# Plugins Go
go install google.golang.org/protobuf/cmd/protoc-gen-go@latest
go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@latest

# Plugins supplémentaires
go install github.com/grpc-ecosystem/grpc-gateway/v2/protoc-gen-grpc-gateway@latest
go install github.com/grpc-ecosystem/grpc-gateway/v2/protoc-gen-openapiv2@latest
go install github.com/envoyproxy/protoc-gen-validate@latest

3.2 Génération

# Générer les stubs
protoc --go_out=. \
  --go_opt=paths=source_relative \
  --go-grpc_out=. \
  --go-grpc_opt=paths=source_relative \
  proto/user/v1/user.proto

# Avec gateway
protoc --go_out=. \
  --go-grpc_out=. \
  --grpc-gateway_out=. \
  --grpc-gateway_opt=logtostderr=true \
  --openapiv2_out=./swagger \
  proto/user/v1/user.proto

# Avec validation
protoc --go_out=. \
  --go-grpc_out=. \
  --validate_out="lang=go:." \
  proto/user/v1/user.proto

3.3 Générer avec buf

# buf.yaml
version: v2
modules:
  - path: proto
lint:
  use:
    - DEFAULT
    - COMMENTS
breaking:
  use:
    - FILE

# buf.gen.yaml
version: v2
plugins:
  - local: protoc-gen-go
    out: gen
    opt: paths=source_relative
  - local: protoc-gen-go-grpc
    out: gen
    opt: paths=source_relative
  - local: protoc-gen-grpc-gateway
    out: gen
    opt:
      - paths=source_relative
      - generate_unbound_methods=true
  - local: protoc-gen-openapiv2
    out: gen/openapiv2
    opt: allow_merge=true,merge_file_name=api

# Générer avec buf
buf generate
buf lint
buf breaking --against '.git#branch=main'

4. Unary RPC

4.1 Server

package server

import (
    "context"
    "fmt"

    userv1 "github.com/example/gen/go/user/v1"
    "google.golang.org/grpc/codes"
    "google.golang.org/grpc/status"
)

type UserServer struct {
    userv1.UnimplementedUserServiceServer
    repo UserRepository
}

func NewUserServer(repo UserRepository) *UserServer {
    return &UserServer{repo: repo}
}

func (s *UserServer) GetUser(ctx context.Context, req *userv1.GetUserRequest) (*userv1.GetUserResponse, error) {
    if err := validateGetUser(req); err != nil {
        return nil, status.Error(codes.InvalidArgument, err.Error())
    }

    user, err := s.repo.FindByID(ctx, req.Id)
    if err != nil {
        if errors.Is(err, ErrNotFound) {
            return nil, status.Error(codes.NotFound, "user not found")
        }
        return nil, status.Errorf(codes.Internal, "find user: %v", err)
    }

    return &userv1.GetUserResponse{User: user.ToProto()}, nil
}

func (s *UserServer) CreateUser(ctx context.Context, req *userv1.CreateUserRequest) (*userv1.CreateUserResponse, error) {
    user := &User{
        Name:  req.Name,
        Email: req.Email,
        Roles: req.Roles,
    }

    if err := s.repo.Save(ctx, user); err != nil {
        return nil, status.Errorf(codes.Internal, "save user: %v", err)
    }

    return &userv1.CreateUserResponse{User: user.ToProto()}, nil
}

func (s *UserServer) DeleteUser(ctx context.Context, req *userv1.DeleteUserRequest) (*emptypb.Empty, error) {
    if err := s.repo.Delete(ctx, req.Id); err != nil {
        return nil, status.Errorf(codes.Internal, "delete user: %v", err)
    }
    return &emptypb.Empty{}, nil
}

4.2 Client

package client

import (
    "context"
    "time"

    userv1 "github.com/example/gen/go/user/v1"
    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials/insecure"
)

type UserClient struct {
    client userv1.UserServiceClient
    conn   *grpc.ClientConn
}

func NewUserClient(address string) (*UserClient, error) {
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()

    conn, err := grpc.DialContext(ctx, address,
        grpc.WithTransportCredentials(insecure.NewCredentials()),
        grpc.WithBlock(),
    )
    if err != nil {
        return nil, fmt.Errorf("dial %s: %w", address, err)
    }

    return &UserClient{
        client: userv1.NewUserServiceClient(conn),
        conn:   conn,
    }, nil
}

func (c *UserClient) GetUser(ctx context.Context, id string) (*userv1.User, error) {
    ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
    defer cancel()

    resp, err := c.client.GetUser(ctx, &userv1.GetUserRequest{Id: id})
    if err != nil {
        return nil, fmt.Errorf("get user: %w", err)
    }
    return resp.User, nil
}

func (c *UserClient) Close() error {
    return c.conn.Close()
}

5. Streaming RPC

5.1 Server Streaming

// Server side
func (s *UserServer) StreamUsers(req *userv1.ListUsersRequest, stream userv1.UserService_StreamUsersServer) error {
    users, err := s.repo.FindAll(stream.Context())
    if err != nil {
        return status.Errorf(codes.Internal, "find all: %v", err)
    }

    for _, user := range users {
        if err := stream.Send(user.ToProto()); err != nil {
            return status.Errorf(codes.Internal, "send: %v", err)
        }
    }
    return nil
}

// Client side
func (c *UserClient) StreamUsers(ctx context.Context) ([]*userv1.User, error) {
    stream, err := c.client.StreamUsers(ctx, &userv1.ListUsersRequest{})
    if err != nil {
        return nil, err
    }

    var users []*userv1.User
    for {
        user, err := stream.Recv()
        if errors.Is(err, io.EOF) {
            break
        }
        if err != nil {
            return nil, err
        }
        users = append(users, user)
    }
    return users, nil
}

5.2 Client Streaming

// Server side
func (s *UserServer) BatchCreateUsers(stream userv1.UserService_BatchCreateUsersServer) error {
    var users []*User
    for {
        req, err := stream.Recv()
        if errors.Is(err, io.EOF) {
            break
        }
        if err != nil {
            return err
        }

        user := &User{
            Name:  req.Name,
            Email: req.Email,
            Roles: req.Roles,
        }
        users = append(users, user)
    }

    // Save all in transaction
    if err := s.repo.BatchSave(stream.Context(), users); err != nil {
        return status.Errorf(codes.Internal, "batch save: %v", err)
    }

    return stream.SendAndClose(&userv1.CreateUserResponse{
        User: users[len(users)-1].ToProto(),
    })
}

// Client side
func (c *UserClient) BatchCreateUsers(ctx context.Context, users []*userv1.CreateUserRequest) (*userv1.User, error) {
    stream, err := c.client.BatchCreateUsers(ctx)
    if err != nil {
        return nil, err
    }

    for _, user := range users {
        if err := stream.Send(user); err != nil {
            return nil, err
        }
    }

    return stream.CloseAndRecv()
}

5.3 Bidirectional Streaming

// Server side
func (s *UserServer) Chat(stream userv1.UserService_ChatServer) error {
    for {
        msg, err := stream.Recv()
        if errors.Is(err, io.EOF) {
            return nil
        }
        if err != nil {
            return err
        }

        // Process message
        reply := &userv1.ChatMessage{
            UserId:    msg.UserId,
            Content:   fmt.Sprintf("Echo: %s", msg.Content),
            Timestamp: timestamppb.Now(),
        }

        if err := stream.Send(reply); err != nil {
            return err
        }
    }
}

// Client side
func (c *UserClient) Chat(ctx context.Context) (userv1.UserService_ChatClient, error) {
    return c.client.Chat(ctx)
}

// Usage
stream, _ := client.Chat(ctx)
go func() {
    for _, msg := range messages {
        stream.Send(&userv1.ChatMessage{Content: msg})
    }
    stream.CloseSend()
}()
for {
    reply, err := stream.Recv()
    if err == io.EOF {
        break
    }
    fmt.Println(reply.Content)
}

6. Intercepteurs

6.1 Unary Interceptor

// Server interceptor (logging)
func LoggingInterceptor(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
    start := time.Now()
    log.Printf("--> %s", info.FullMethod)

    resp, err := handler(ctx, req)

    duration := time.Since(start)
    if err != nil {
        log.Printf("<-- %s error: %v (%s)", info.FullMethod, err, duration)
    } else {
        log.Printf("<-- %s success (%s)", info.FullMethod, duration)
    }
    return resp, err
}

// Server interceptor (auth)
func AuthInterceptor(validToken string) grpc.UnaryServerInterceptor {
    return func(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
        md, ok := metadata.FromIncomingContext(ctx)
        if !ok {
            return nil, status.Error(codes.Unauthenticated, "missing metadata")
        }

        tokens := md.Get("authorization")
        if len(tokens) == 0 || tokens[0] != "Bearer "+validToken {
            return nil, status.Error(codes.Unauthenticated, "invalid token")
        }

        return handler(ctx, req)
    }
}

// Client interceptor (timing)
func TimingInterceptor(ctx context.Context, method string, req, reply any, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
    start := time.Now()
    err := invoker(ctx, method, req, reply, cc, opts...)
    log.Printf("%s took %s", method, time.Since(start))
    return err
}

6.2 Stream Interceptor

// Server stream interceptor
func StreamLoggingInterceptor(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
    log.Printf("--> stream %s", info.FullMethod)
    err := handler(srv, ss)
    if err != nil {
        log.Printf("<-- stream %s error: %v", info.FullMethod, err)
    } else {
        log.Printf("<-- stream %s done", info.FullMethod)
    }
    return err
}

// Recovery interceptor (panic recovery)
func RecoveryInterceptor() grpc.UnaryServerInterceptor {
    return func(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp any, err error) {
        defer func() {
            if r := recover(); r != nil {
                log.Printf("panic recovered: %v", r)
                err = status.Errorf(codes.Internal, "internal error")
            }
        }()
        return handler(ctx, req)
    }
}

// Rate limiter interceptor
func RateLimitInterceptor(maxRequests int) grpc.UnaryServerInterceptor {
    var mu sync.Mutex
    var requests int
    var lastReset time.Time

    return func(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
        mu.Lock()
        if time.Since(lastReset) > time.Second {
            requests = 0
            lastReset = time.Now()
        }
        requests++
        if requests > maxRequests {
            mu.Unlock()
            return nil, status.Error(codes.ResourceExhausted, "rate limit exceeded")
        }
        mu.Unlock()
        return handler(ctx, req)
    }
}

6.3 Registration

func main() {
    lis, _ := net.Listen("tcp", ":50051")

    s := grpc.NewServer(
        grpc.ChainUnaryInterceptor(
            LoggingInterceptor,
            AuthInterceptor(os.Getenv("JWT_SECRET")),
            RecoveryInterceptor(),
            RateLimitInterceptor(1000),
        ),
        grpc.ChainStreamInterceptor(
            StreamLoggingInterceptor,
        ),
    )

    userv1.RegisterUserServiceServer(s, server.NewUserServer(repo))
    s.Serve(lis)
}

7. Deadlines et Timeouts

// Server side
func (s *UserServer) GetUser(ctx context.Context, req *userv1.GetUserRequest) (*userv1.GetUserResponse, error) {
    // Vérifier si le deadline est dépassé
    if deadline, ok := ctx.Deadline(); ok {
        if time.Until(deadline) < 100*time.Millisecond {
            return nil, status.Error(codes.DeadlineExceeded, "too late to process")
        }
    }

    // Propagation du contexte
    dbCtx, cancel := context.WithTimeout(ctx, 1*time.Second)
    defer cancel()

    user, err := s.repo.FindByID(dbCtx, req.Id)
    if err != nil {
        return nil, err
    }

    return &userv1.GetUserResponse{User: user.ToProto()}, nil
}

// Client side
func (c *UserClient) GetUserWithTimeout(ctx context.Context, id string) (*userv1.User, error) {
    ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
    defer cancel()

    resp, err := c.client.GetUser(ctx, &userv1.GetUserRequest{Id: id})
    if err != nil {
        if status.Code(err) == codes.DeadlineExceeded {
            log.Println("request timed out")
        }
        return nil, err
    }
    return resp.User, nil
}

8. Error Handling

package errors

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

var (
    ErrNotFound     = status.Error(codes.NotFound, "resource not found")
    ErrInvalidArg   = status.Error(codes.InvalidArgument, "invalid argument")
    ErrUnauthenticated = status.Error(codes.Unauthenticated, "unauthenticated")
    ErrPermissionDenied = status.Error(codes.PermissionDenied, "permission denied")
    ErrAlreadyExists = status.Error(codes.AlreadyExists, "already exists")
    ErrUnimplemented = status.Error(codes.Unimplemented, "unimplemented")
    ErrUnavailable   = status.Error(codes.Unavailable, "service unavailable")
)

func NewInvalidFieldError(field, reason string) error {
    st := status.New(codes.InvalidArgument, "validation failed")
    st, _ = st.WithDetails(&errdetails.BadRequest{
        FieldViolations: []*errdetails.BadRequest_FieldViolation{
            {Field: field, Description: reason},
        },
    })
    return st.Err()
}

func NewErrorf(code codes.Code, format string, args ...any) error {
    return status.Errorf(code, format, args...)
}

func IsNotFound(err error) bool {
    return status.Code(err) == codes.NotFound
}

func ErrorCode(err error) codes.Code {
    return status.Code(err)
}

9. Load Balancing

package lb

import (
    "google.golang.org/grpc"
    "google.golang.org/grpc/balancer"
    _ "google.golang.org/grpc/balancer/roundrobin"
    _ "google.golang.org/grpc/balancer/weightedroundrobin"
    "google.golang.org/grpc/resolver"
    _ "google.golang.org/grpc/resolver/dns"
    _ "google.golang.org/grpc/resolver/manual"
)

func NewClientWithLB(address string) (*grpc.ClientConn, error) {
    // DNS-based load balancing
    return grpc.Dial(
        fmt.Sprintf("dns:///%s", address),
        grpc.WithDefaultServiceConfig(`{
            "loadBalancingConfig": [{"round_robin":{}}]
        }`),
        grpc.WithInsecure(),
    )
}

func NewClientWithWeightedLB(address string) (*grpc.ClientConn, error) {
    return grpc.Dial(
        address,
        grpc.WithDefaultServiceConfig(`{
            "loadBalancingConfig": [{"weighted_round_robin":{}}]
        }`),
        grpc.WithInsecure(),
    )
}

// Custom resolver
type CustomResolver struct {
    target resolver.Target
    cc     resolver.ClientConn
}

func (r *CustomResolver) ResolveNow(o resolver.ResolveNowOptions) {
    // Get endpoints from service registry
    endpoints := []string{"10.0.0.1:50051", "10.0.0.2:50051", "10.0.0.3:50051"}
    
    addrs := make([]resolver.Address, len(endpoints))
    for i, ep := range endpoints {
        addrs[i] = resolver.Address{Addr: ep}
    }
    r.cc.UpdateState(resolver.State{Addresses: addrs})
}

10. Gateway REST

// grpc-gateway registration
package main

import (
    "net/http"
    "github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials/insecure"

    userv1 "github.com/example/gen/go/user/v1"
)

func main() {
    ctx := context.Background()
    mux := runtime.NewServeMux(
        runtime.WithIncomingHeaderMatcher(func(key string) (string, bool) {
            if strings.HasPrefix(key, "X-") {
                return key, true
            }
            return runtime.DefaultHeaderMatcher(key)
        }),
    )

    opts := []grpc.DialOption{
        grpc.WithTransportCredentials(insecure.NewCredentials()),
    }

    userv1.RegisterUserServiceHandlerFromEndpoint(ctx, mux, "localhost:50051", opts)

    http.ListenAndServe(":8080", mux)
}

11. Diagrammes

Diagramme en cours de génération...

Diagramme en cours de génération...

Points Clés

1. Proto3 est le langage de définition d'interface
2. 4 types de RPC : unary, server stream, client stream, bidirectional
3. Intercepteurs : cross-cutting concerns (auth, logging, rate limit)
4. Deadlines : toujours set un timeout côté client
5. Status codes : utiliser les codes gRPC standards
6. grpc-gateway : REST ↔ gRPC bridge
7. buf : meilleur outil pour gérer les protos