10장: gRPC와 마이크로서비스 통신

syntax = "proto3";
 
package user.v1;
 
option go_package = "github.com/myapp/gen/user/v1;userv1";
 
import "google/protobuf/timestamp.proto";
 
// 사용자 메시지 정의
message User {
  int64 id = 1;
  string name = 2;
  string email = 3;
  Role role = 4;
  google.protobuf.Timestamp created_at = 5;
}
 
enum Role {
  ROLE_UNSPECIFIED = 0;
  ROLE_USER = 1;
  ROLE_ADMIN = 2;
}
 
// 서비스 정의
service UserService {
  rpc GetUser(GetUserRequest) returns (GetUserResponse);
  rpc ListUsers(ListUsersRequest) returns (ListUsersResponse);
  rpc CreateUser(CreateUserRequest) returns (CreateUserResponse);
  rpc UpdateUser(UpdateUserRequest) returns (UpdateUserResponse);
  rpc DeleteUser(DeleteUserRequest) returns (DeleteUserResponse);
}
 
message GetUserRequest {
  int64 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;
  string password = 3;
}
 
message CreateUserResponse {
  User user = 1;
}

# 도구 설치
go install google.golang.org/protobuf/cmd/protoc-gen-go@latest
go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@latest
 
# Go 코드 생성
protoc --go_out=. --go_opt=paths=source_relative \
       --go-grpc_out=. --go-grpc_opt=paths=source_relative \
       proto/user/v1/user.proto

import (
    "google.golang.org/grpc"
    "google.golang.org/grpc/codes"
    "google.golang.org/grpc/status"
    userv1 "github.com/myapp/gen/user/v1"
)
 
type userServer struct {
    userv1.UnimplementedUserServiceServer
    repo UserRepository
}
 
func (s *userServer) GetUser(ctx context.Context, req *userv1.GetUserRequest) (*userv1.GetUserResponse, error) {
    if req.Id <= 0 {
        return nil, status.Errorf(codes.InvalidArgument, "유효하지 않은 사용자 ID: %d", req.Id)
    }
 
    user, err := s.repo.FindByID(ctx, req.Id)
    if err != nil {
        if errors.Is(err, ErrNotFound) {
            return nil, status.Errorf(codes.NotFound, "사용자를 찾을 수 없습니다: ID %d", req.Id)
        }
        return nil, status.Errorf(codes.Internal, "사용자 조회 실패: %v", err)
    }
 
    return &userv1.GetUserResponse{
        User: toProtoUser(user),
    }, nil
}
 
func (s *userServer) CreateUser(ctx context.Context, req *userv1.CreateUserRequest) (*userv1.CreateUserResponse, error) {
    if req.Name == "" || req.Email == "" {
        return nil, status.Error(codes.InvalidArgument, "이름과 이메일은 필수입니다")
    }
 
    user := &User{
        Name:  req.Name,
        Email: req.Email,
    }
 
    if err := s.repo.Create(ctx, user); err != nil {
        return nil, status.Errorf(codes.Internal, "사용자 생성 실패: %v", err)
    }
 
    return &userv1.CreateUserResponse{
        User: toProtoUser(user),
    }, nil
}
 
// 서버 시작
func main() {
    lis, err := net.Listen("tcp", ":50051")
    if err != nil {
        log.Fatal("리슨 실패:", err)
    }
 
    grpcServer := grpc.NewServer()
    userv1.RegisterUserServiceServer(grpcServer, &userServer{repo: repo})
 
    slog.Info("gRPC 서버 시작", "addr", ":50051")
    if err := grpcServer.Serve(lis); err != nil {
        log.Fatal("서버 실행 실패:", err)
    }
}

codes.OK              -> 200 OK
codes.InvalidArgument -> 400 Bad Request
codes.Unauthenticated -> 401 Unauthorized
codes.PermissionDenied-> 403 Forbidden
codes.NotFound        -> 404 Not Found
codes.AlreadyExists   -> 409 Conflict
codes.Internal        -> 500 Internal Server Error
codes.Unavailable     -> 503 Service Unavailable
codes.DeadlineExceeded-> 504 Gateway Timeout

service EventService {
  rpc StreamEvents(StreamEventsRequest) returns (stream Event);
}

func (s *eventServer) StreamEvents(
    req *eventv1.StreamEventsRequest,
    stream eventv1.EventService_StreamEventsServer,
) error {
    ticker := time.NewTicker(time.Second)
    defer ticker.Stop()
 
    for {
        select {
        case <-stream.Context().Done():
            return nil
        case t := <-ticker.C:
            event := &eventv1.Event{
                Id:        uuid.New().String(),
                Type:      "heartbeat",
                Timestamp: timestamppb.New(t),
            }
            if err := stream.Send(event); err != nil {
                return status.Errorf(codes.Internal, "이벤트 전송 실패: %v", err)
            }
        }
    }
}

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

func (s *chatServer) Chat(stream chatv1.ChatService_ChatServer) error {
    for {
        msg, err := stream.Recv()
        if err == io.EOF {
            return nil
        }
        if err != nil {
            return status.Errorf(codes.Internal, "메시지 수신 실패: %v", err)
        }
 
        // 메시지 처리 후 응답
        response := &chatv1.ChatMessage{
            UserId:  "system",
            Content: fmt.Sprintf("수신 확인: %s", msg.Content),
        }
 
        if err := stream.Send(response); err != nil {
            return status.Errorf(codes.Internal, "응답 전송 실패: %v", err)
        }
    }
}

// 로깅 인터셉터
func LoggingInterceptor(
    ctx context.Context,
    req interface{},
    info *grpc.UnaryServerInfo,
    handler grpc.UnaryHandler,
) (interface{}, error) {
    start := time.Now()
 
    // 핸들러 실행
    resp, err := handler(ctx, req)
 
    duration := time.Since(start)
    statusCode := status.Code(err)
 
    slog.Info("gRPC 요청 처리",
        "method", info.FullMethod,
        "status", statusCode.String(),
        "duration_ms", duration.Milliseconds(),
    )
 
    return resp, err
}
 
// 인증 인터셉터
func AuthInterceptor(jwtSvc *JWTService) grpc.UnaryServerInterceptor {
    return func(
        ctx context.Context,
        req interface{},
        info *grpc.UnaryServerInfo,
        handler grpc.UnaryHandler,
    ) (interface{}, error) {
        // 인증이 필요 없는 메서드 건너뛰기
        if isPublicMethod(info.FullMethod) {
            return handler(ctx, req)
        }
 
        md, ok := metadata.FromIncomingContext(ctx)
        if !ok {
            return nil, status.Error(codes.Unauthenticated, "메타데이터 없음")
        }
 
        tokens := md.Get("authorization")
        if len(tokens) == 0 {
            return nil, status.Error(codes.Unauthenticated, "인증 토큰 없음")
        }
 
        claims, err := jwtSvc.ValidateToken(strings.TrimPrefix(tokens[0], "Bearer "))
        if err != nil {
            return nil, status.Error(codes.Unauthenticated, "유효하지 않은 토큰")
        }
 
        ctx = context.WithValue(ctx, userClaimsKey, claims)
        return handler(ctx, req)
    }
}
 
// 서버에 인터셉터 적용
grpcServer := grpc.NewServer(
    grpc.ChainUnaryInterceptor(
        LoggingInterceptor,
        AuthInterceptor(jwtSvc),
        RecoveryInterceptor,
    ),
)

import "google/api/annotations.proto";
 
service UserService {
  rpc GetUser(GetUserRequest) returns (GetUserResponse) {
    option (google.api.http) = {
      get: "/api/v1/users/{id}"
    };
  }
 
  rpc CreateUser(CreateUserRequest) returns (CreateUserResponse) {
    option (google.api.http) = {
      post: "/api/v1/users"
      body: "*"
    };
  }
 
  rpc ListUsers(ListUsersRequest) returns (ListUsersResponse) {
    option (google.api.http) = {
      get: "/api/v1/users"
    };
  }
}

func runGateway(ctx context.Context, grpcAddr, httpAddr string) error {
    mux := runtime.NewServeMux(
        runtime.WithMarshalerOption(
            runtime.MIMEWildcard,
            &runtime.JSONPb{
                MarshalOptions: protojson.MarshalOptions{
                    UseProtoNames: true,
                },
            },
        ),
    )
 
    opts := []grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())}
 
    err := userv1.RegisterUserServiceHandlerFromEndpoint(ctx, mux, grpcAddr, opts)
    if err != nil {
        return fmt.Errorf("게이트웨이 등록 실패: %w", err)
    }
 
    slog.Info("HTTP 게이트웨이 시작", "addr", httpAddr)
    return http.ListenAndServe(httpAddr, mux)
}

import "google.golang.org/grpc/health"
import healthpb "google.golang.org/grpc/health/grpc_health_v1"
 
func main() {
    grpcServer := grpc.NewServer()
 
    // 헬스 체크 서비스 등록
    healthServer := health.NewServer()
    healthpb.RegisterHealthServer(grpcServer, healthServer)
 
    // 서비스 상태 설정
    healthServer.SetServingStatus("user.v1.UserService",
        healthpb.HealthCheckResponse_SERVING)
 
    // 서비스 중지 시
    // healthServer.SetServingStatus("user.v1.UserService",
    //     healthpb.HealthCheckResponse_NOT_SERVING)
}

// gRPC 리졸버를 통한 서비스 디스커버리
conn, err := grpc.Dial(
    "dns:///user-service.default.svc.cluster.local:50051",
    grpc.WithDefaultServiceConfig(`{
        "loadBalancingConfig": [{"round_robin": {}}]
    }`),
    grpc.WithTransportCredentials(insecure.NewCredentials()),
)