grpc

What I do

• Define Protocol Buffer schemas
• Implement gRPC services
• Handle streaming
• Implement authentication
• Handle errors properly
• Manage versioning
• Test gRPC services
• Scale gRPC deployments

When to use me

When implementing gRPC services or working with Protocol Buffers.

Protocol Buffer Definition

syntax = "proto3";

package user.v1;

option go_package = "github.com/company/api/user/v1";

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


// Service definition
service UserService {
  // Unary RPC
  rpc GetUser(GetUserRequest) returns (User);
  
  // Server streaming
  rpc ListUsers(ListUsersRequest) returns (stream User);
  
  // Client streaming
  rpc CreateUsers(stream CreateUserRequest) returns (CreateUsersResponse);
  
  // Bidirectional streaming
  rpc StreamUserUpdates(stream UserUpdateRequest) returns (stream User);
  
  // Health check
  rpc HealthCheck(google.protobuf.Empty) returns (HealthCheckResponse);
}


// Message types
message GetUserRequest {
  string user_id = 1;
}

message User {
  string id = 1;
  string email = 2;
  string name = 3;
  UserStatus status = 4;
  google.protobuf.Timestamp created_at = 5;
  google.protobuf.Timestamp updated_at = 6;
}

enum UserStatus {
  USER_STATUS_UNSPECIFIED = 0;
  USER_STATUS_ACTIVE = 1;
  USER_STATUS_INACTIVE = 2;
  USER_STATUS_SUSPENDED = 3;
}

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

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

message CreateUsersResponse {
  repeated User users = 1;
  int32 failed_count = 2;
}

message UserUpdateRequest {
  string user_id = 1;
  string name = 2;
}

message HealthCheckResponse {
  bool healthy = 1;
  string version = 2;
}

Go gRPC Server

package main

import (
    "context"
    "log"
    "net"

    "google.golang.org/grpc"
    "google.golang.org/grpc/codes"
    "google.golang.org/grpc/credentials"
    "google.golang.org/grpc/metadata"
    "google.golang.org/grpc/status"
    "google.golang.org/protobuf/types/known/timestamppb"

    pb "github.com/company/api/user/v1"
)

type UserServer struct {
    pb.UnimplementedUserServiceServer
    userStore UserStore
}

type UserStore interface {
    GetUser(ctx context.Context, id string) (*pb.User, error)
    ListUsers(ctx context.Context, filter *pb.UserStatus, limit int) ([]*pb.User, error)
    CreateUser(ctx context.Context, user *pb.User) error
}

func NewUserServer(store UserStore) *UserServer {
    return &UserServer{userStore: store}
}

func (s *UserServer) GetUser(
    ctx context.Context,
    req *pb.GetUserRequest,
) (*pb.User, error) {
    // Extract metadata for logging
    md, ok := metadata.FromIncomingContext(ctx)
    if ok {
        log.Printf("GetUser request: user_id=%s, from=%v", req.UserId, md.Get("x-forwarded-for"))
    }

    // Validate request
    if req.UserId == "" {
        return nil, status.Error(codes.InvalidArgument, "user_id is required")
    }

    // Fetch user
    user, err := s.userStore.GetUser(ctx, req.UserId)
    if err != nil {
        return nil, status.Errorf(codes.NotFound, "user not found: %s", err)
    }

    return user, nil
}

func (s *UserServer) ListUsers(
    req *pb.ListUsersRequest,
    stream pb.UserService_ListUsersServer,
) error {
    ctx := stream.Context()

    // Parse pagination
    pageSize := int(req.PageSize)
    if pageSize <= 0 {
        pageSize = 10
    }
    if pageSize > 100 {
        pageSize = 100
    }

    users, err := s.userStore.ListUsers(ctx, req.StatusFilter, pageSize)
    if err != nil {
        return status.Errorf(codes.Internal, "failed to list users: %v", err)
    }

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

    return nil
}

func (s *UserService) CreateUsers(
    stream pb.UserService_CreateUsersServer,
) error {
    ctx := stream.Context()
    var users []*pb.User
    var failedCount int32

    for {
        req, err := stream.Recv()
        if err != nil {
            if err.Error() == "EOF" {
                break
            }
            return status.Errorf(codes.Internal, "failed to receive request: %v", err)
        }

        user := &pb.User{
            Email:     req.Email,
            Name:      req.Name,
            Status:    pb.UserStatus_USER_STATUS_ACTIVE,
            CreatedAt: timestamppb.Now(),
        }

        if err := s.userStore.CreateUser(ctx, user); err != nil {
            failedCount++
            log.Printf("Failed to create user: %v", err)
            continue
        }

        users = append(users, user)
    }

    return stream.SendAndClose(&pb.CreateUsersResponse{
        Users:       users,
        FailedCount: failedCount,
    })
}

func (s *UserServer) StreamUserUpdates(
    stream pb.UserService_StreamUserUpdatesServer,
) error {
    ctx := stream.Context()

    for {
        select {
        case <-ctx.Done():
            return ctx.Err()
        default:
            req, err := stream.Recv()
            if err != nil {
                return err
            }

            user, err := s.userStore.GetUser(ctx, req.UserId)
            if err != nil {
                continue
            }

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

// Server reflection for debugging
func init() {
    // Enable reflection
    // reflection.Register(grpcServer)
}

gRPC Client

package main

import (
    "context"
    "log"
    "time"

    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials/insecure"
    "google.golang.org/grpc/metadata"

    pb "github.com/company/api/user/v1"
)

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

func NewUserClient(addr string) (*UserClient, error) {
    conn, err := grpc.Dial(
        addr,
        grpc.WithTransportCredentials(insecure.NewCredentials()),
        grpc.WithUnaryInterceptor(loggingInterceptor),
        grpc.WithStreamInterceptor(streamLoggingInterceptor),
    )
    if err != nil {
        return nil, err
    }

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

func (c *UserClient) GetUser(ctx context.Context, userId string) (*pb.User, error) {
    // Add metadata for tracing
    ctx = metadata.AppendToOutgoingContext(ctx, "x-request-id", "req-123")

    ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
    defer cancel()

    return c.client.GetUser(ctx, &pb.GetUserRequest{
        UserId: userId,
    })
}

func (c *UserClient) ListActiveUsers(ctx context.Context) ([]*pb.User, error) {
    ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
    defer cancel()

    stream, err := c.client.ListUsers(ctx, &pb.ListUsersRequest{
        PageSize:  100,
        StatusFilter: pb.UserStatus_USER_STATUS_ACTIVE,
    })
    if err != nil {
        return nil, err
    }

    var users []*pb.User
    for {
        user, err := stream.Recv()
        if err != nil {
            if err.Error() == "EOF" {
                break
            }
            return nil, err
        }
        users = append(users, user)
    }

    return users, nil
}

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

// Interceptors
func loggingInterceptor(
    ctx context.Context,
    method string,
    req, reply interface{},
    cc *grpc.ClientConn,
    invoker grpc.UnaryInvoker,
    opts ...grpc.CallOption,
) error {
    start := time.Now()
    
    err := invoker(ctx, method, req, reply, cc, opts...)
    
    log.Printf("gRPC call: %s, duration: %v, error: %v",
        method, time.Since(start), err)
    
    return err
}

func streamLoggingInterceptor(
    ctx context.Context,
    desc *grpc.StreamDesc,
    cc *grpc.ClientConn,
    method string,
    stream grpc.ClientStream,
    opts ...grpc.CallOption,
) error {
    start := time.Now()
    
    err := stream.RecvMsg(nil)
    
    log.Printf("gRPC stream: %s, duration: %v, error: %v",
        method, time.Since(start), err)
    
    return err
}

Error Handling

package status

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

// Custom error codes
const (
    ErrCodeUserNotFound = "USER_NOT_FOUND"
    ErrCodeInvalidInput = "INVALID_INPUT"
    ErrCodeUnauthorized = "UNAUTHORIZED"
)

func UserNotFound(id string) error {
    return status.Errorf(codes.NotFound, "user not found: %s", id)
}

func InvalidInput(field, reason string) error {
    return status.Errorf(codes.InvalidArgument, "invalid %s: %s", field, reason)
}

func Unauthorized(reason string) error {
    return status.Errorf(codes.Unauthenticated, "unauthorized: %s", reason)
}

gRPC Best Practices

1. Use Protocol Buffers for schema
   - Define contracts explicitly
   - Enable backwards compatibility

2. Handle errors properly
   - Use gRPC status codes
   - Include error details

3. Implement interceptors
   - Logging, auth, metrics

4. Use streaming appropriately
   - Server streaming for lists
   - Bidirectional for real-time

5. Set timeouts
   - Always use context with timeout

6. Enable reflection
   - For debugging and tools

7. Version your APIs
   - Include version in package name
   - Support multiple versions

8. Use proper authentication
   - Token-based or mTLS

9. Monitor and trace
   - Add interceptors for metrics
   - Use OpenTelemetry

10. Test thoroughly
    - Unit tests with mock stores
    - Integration tests