Golang gRPC Microservices chuẩn Production

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

package driver.v1;

option go_package = "github.com/yourorg/platform/gen/driver/v1;driverv1";

import "google/protobuf/timestamp.proto";

// Dịch vụ DriverService quản trị vị trí và sự hiện diện của lái xe
service DriverService {
  // Lệnh Unary: Trích xuất một tài xế đơn lẻ bằng mã ID
  rpc GetDriver(GetDriverRequest) returns (GetDriverResponse);

  // Streaming từ chiều Server (Server streaming): Dò bắt địa điểm xe chạy ở khung thời gian thực
  rpc StreamLocation(StreamLocationRequest) returns (stream LocationUpdate);

  // Streaming từ chiều Client (Client streaming): App tài xế truyền gộp cục (bulk) liên tiếp GPS updates
  rpc UploadLocations(stream LocationUpdate) returns (UploadSummary);

  // Streaming hai chiều (Bidirectional): Liên lạc truyền dẫn tín hiệu song công (Full-duplex) giữa tài xế-máy chủ
  rpc DriverSession(stream DriverEvent) returns (stream ServerCommand);
}

message GetDriverRequest {
  string driver_id = 1;
}

message GetDriverResponse {
  string driver_id = 1;
  string status = 2;          // AVAILABLE, BUSY, OFFLINE
  double latitude = 3;
  double longitude = 4;
  google.protobuf.Timestamp last_seen_at = 5;
}

message StreamLocationRequest {
  string driver_id = 1;
}

message LocationUpdate {
  string driver_id = 1;
  double latitude = 2;
  double longitude = 3;
  float speed_mps = 4;
  float heading_degrees = 5;
  google.protobuf.Timestamp timestamp = 6;
}

message UploadSummary {
  int32 received_count = 1;
  int32 persisted_count = 2;
  string session_id = 3;
}

message DriverEvent {
  oneof event {
    LocationUpdate location = 1;
    DriverStatusChange status_change = 2;
    HeartbeatPing heartbeat = 3;
  }
}

message ServerCommand {
  oneof command {
    RideOffer ride_offer = 1;
    NavigationUpdate navigation = 2;
    PingResponse pong = 3;
  }
}

message DriverStatusChange {
  string driver_id = 1;
  string new_status = 2;
}

message HeartbeatPing { int64 client_ts_ms = 1; }
message PingResponse { int64 server_ts_ms = 1; }
message RideOffer { string offer_id = 1; string pickup_address = 2; }
message NavigationUpdate { string polyline = 1; }

# Cài đặt các công cụ cần thiết
go install google.golang.org/protobuf/cmd/protoc-gen-go@latest
go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@latest

# Sinh mã nguồn Go từ file proto — thực hiện tại thư mục gốc của project
protoc \
  --go_out=gen \
  --go_opt=paths=source_relative \
  --go-grpc_out=gen \
  --go-grpc_opt=paths=source_relative \
  proto/driver/v1/driver.proto

// internal/driver/server.go
package driver

import (
    "context"
    "fmt"
    "io"
    "log/slog"
    "time"

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

    driverv1 "github.com/yourorg/platform/gen/driver/v1"
)

// Định nghĩa Server struct triển khai interface driverv1.DriverServiceServer
type Server struct {
    driverv1.UnimplementedDriverServiceServer
    repo      DriverRepository
    publisher LocationPublisher
    logger    *slog.Logger
}

func NewServer(repo DriverRepository, pub LocationPublisher, log *slog.Logger) *Server {
    return &Server{repo: repo, publisher: pub, logger: log}
}

// Hàm GetDriver — Mẫu Unary RPC tiêu chuẩn
func (s *Server) GetDriver(ctx context.Context, req *driverv1.GetDriverRequest) (*driverv1.GetDriverResponse, error) {
    if req.DriverId == "" {
        return nil, status.Error(codes.InvalidArgument, "bắt buộc phải truyền mã driver_id")
    }

    driver, err := s.repo.FindByID(ctx, req.DriverId)
    if err != nil {
        s.logger.ErrorContext(ctx, "GetDriver: lỗi truy vấn repository", "driver_id", req.DriverId, "err", err)
        return nil, status.Errorf(codes.Internal, "việc trích xuất tài xế gặp lỗi: %v", err)
    }
    if driver == nil {
        return nil, status.Errorf(codes.NotFound, "không tìm thấy tài xế %s", req.DriverId)
    }

    return &driverv1.GetDriverResponse{
        DriverId:   driver.ID,
        Status:     driver.Status,
        Latitude:   driver.Lat,
        Longitude:  driver.Lng,
        LastSeenAt: timestamppb.New(driver.LastSeenAt),
    }, nil
}

// StreamLocation là Server-streaming RPC
// Định kỳ gửi vị trí thực (live location) của tài xế tới client mỗi 2 giây
func (s *Server) StreamLocation(req *driverv1.StreamLocationRequest, stream driverv1.DriverService_StreamLocationServer) error {
    ctx := stream.Context()

    for {
        select {
        case <-ctx.Done():
            return nil // Client đã bấm ngắt kết nối
        case <-time.After(2 * time.Second):
            loc, err := s.repo.GetCurrentLocation(ctx, req.DriverId)
            if err != nil {
                return status.Errorf(codes.Internal, "chức năng kéo location gặp sự cố thất bại: %v", err)
            }
            if err := stream.Send(&driverv1.LocationUpdate{
                DriverId:  req.DriverId,
                Latitude:  loc.Lat,
                Longitude: loc.Lng,
                Timestamp: timestamppb.Now(),
            }); err != nil {
                return err // Client đã ngắt kết nối trong khi streaming
            }
        }
    }
}

// UploadLocations là Client-streaming RPC
// App tài xế gửi tọa độ GPS theo batch; server tiếp nhận và lưu vào cơ sở dữ liệu
func (s *Server) UploadLocations(stream driverv1.DriverService_UploadLocationsServer) error {
    var received, persisted int32
    var sessionID string

    for {
        update, err := stream.Recv()
        if err == io.EOF {
            // Client đã hoàn tất gửi dữ liệu; đóng luồng và trả về summary
            return stream.SendAndClose(&driverv1.UploadSummary{
                ReceivedCount: received,
                PersistedCount: persisted,
                SessionId:     sessionID,
            })
        }
        if err != nil {
            return status.Errorf(codes.Internal, "xảy ra lỗi phần tiếp nhận recv error: %v", err)
        }

        received++
        sessionID = fmt.Sprintf("sess-%s-%d", update.DriverId, time.Now().UnixMilli())

        if err := s.publisher.Publish(stream.Context(), update); err != nil {
            s.logger.Warn("lỗi khi publish dữ liệu", "driver_id", update.DriverId, "err", err)
            continue // Bỏ qua bản ghi lỗi, tiếp tục xử lý các bản ghi còn lại trong batch
        }
        persisted++
    }
}

// Hàm DriverSession — Dạng Bidirectional streaming RPC
func (s *Server) DriverSession(stream driverv1.DriverService_DriverSessionServer) error {
    ctx := stream.Context()

    for {
        event, err := stream.Recv()
        if err == io.EOF {
            return nil
        }
        if err != nil {
            return err
        }

        switch e := event.Event.(type) {
        case *driverv1.DriverEvent_Location:
            _ = s.publisher.Publish(ctx, e.Location)

        case *driverv1.DriverEvent_Heartbeat:
            if err := stream.Send(&driverv1.ServerCommand{
                Command: &driverv1.ServerCommand_Pong{
                    Pong: &driverv1.PingResponse{ServerTsMs: time.Now().UnixMilli()},
                },
            }); err != nil {
                return err
            }

        case *driverv1.DriverEvent_StatusChange:
            s.logger.InfoContext(ctx, "trạng thái tài xế thay đổi",
                "driver_id", e.StatusChange.DriverId,
                "new_status", e.StatusChange.NewStatus,
            )
        }
    }
}

// internal/interceptor/chain.go
package interceptor

import (
    "context"
    "log/slog"
    "runtime/debug"
    "time"

    "google.golang.org/grpc"
    "google.golang.org/grpc/codes"
    "google.golang.org/grpc/metadata"
    "google.golang.org/grpc/status"
)

// LoggingUnaryInterceptor ghi nhận phương thức, thời gian xử lý và status code của từng RPC.
func LoggingUnaryInterceptor(logger *slog.Logger) grpc.UnaryServerInterceptor {
    return func(
        ctx context.Context,
        req any,
        info *grpc.UnaryServerInfo,
        handler grpc.UnaryHandler,
    ) (any, error) {
        start := time.Now()
        resp, err := handler(ctx, req)

        code := codes.OK
        if err != nil {
            code = status.Code(err)
        }

        logger.InfoContext(ctx, "dòng grpc unary",
            "method", info.FullMethod,
            "duration_ms", time.Since(start).Milliseconds(),
            "code", code.String(),
        )
        return resp, err
    }
}

// AuthUnaryInterceptor xác thực header Authorization.
func AuthUnaryInterceptor(tokenValidator TokenValidator) 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, "thiếu metadata")
        }

        tokens := md.Get("authorization")
        if len(tokens) == 0 {
            return nil, status.Error(codes.Unauthenticated, "thiếu authorization token")
        }

        claims, err := tokenValidator.Validate(tokens[0])
        if err != nil {
            return nil, status.Errorf(codes.Unauthenticated, "token không hợp lệ: %v", err)
        }

        // Lưu claims vào context để chuyển cho các handler downstream
        ctx = context.WithValue(ctx, claimsKey{}, claims)
        return handler(ctx, req)
    }
}

// RecoveryUnaryInterceptor bắt các sự cố panic và chuyển thành lỗi gRPC Internal.
func RecoveryUnaryInterceptor(logger *slog.Logger) 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 {
                logger.ErrorContext(ctx, "phục hồi từ panic",
                    "method", info.FullMethod,
                    "panic", r,
                    "stack", string(debug.Stack()),
                )
                err = status.Errorf(codes.Internal, "lỗi nội bộ server")
            }
        }()
        return handler(ctx, req)
    }
}

type claimsKey struct{}
type TokenValidator interface {
    Validate(token string) (Claims, error)
}
type Claims struct{ SubjectID string }

// cmd/server/main.go
package main

import (
    "crypto/tls"
    "crypto/x509"
    "fmt"
    "log"
    "net"
    "os"
    "os/signal"
    "syscall"
    "time"

    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials"
    "google.golang.org/grpc/health"
    "google.golang.org/grpc/health/grpc_health_v1"
    "google.golang.org/grpc/keepalive"
    "google.golang.org/grpc/reflection"

    driverv1 "github.com/yourorg/platform/gen/driver/v1"
    "github.com/yourorg/platform/internal/driver"
    "github.com/yourorg/platform/internal/interceptor"
)

func main() {
    // --- Lắp mTLS credentials ---
    cert, err := tls.LoadX509KeyPair("certs/server.crt", "certs/server.key")
    if err != nil {
        log.Fatalf("lỗi tải server certificate: %v", err)
    }

    caCert, err := os.ReadFile("certs/ca.crt")
    if err != nil {
        log.Fatalf("lỗi đọc CA certificate: %v", err)
    }
    caPool := x509.NewCertPool()
    caPool.AppendCertsFromPEM(caCert)

    tlsCreds := credentials.NewTLS(&tls.Config{
        Certificates: []tls.Certificate{cert},
        ClientAuth:   tls.RequireAndVerifyClientCert, // Chuẩn mTLS: Yêu cầu và xác thực certificate từ client
        ClientCAs:    caPool,
        MinVersion:   tls.VersionTLS13,
    })

    // --- Khởi tạo gRPC server với interceptor chain ---
    logger := slog.New(slog.NewJSONHandler(os.Stdout, nil))

    srv := grpc.NewServer(
        grpc.Creds(tlsCreds),
        grpc.ChainUnaryInterceptor(
            interceptor.RecoveryUnaryInterceptor(logger),   // Recovery interceptor phải đặt đầu chain để bắt panic từ các handler sau
            interceptor.LoggingUnaryInterceptor(logger),
            interceptor.AuthUnaryInterceptor(tokenValidator),
        ),
        // Cấu hình Keepalive: Duy trì kết nối qua các load balancer / NAT gateway
        grpc.KeepaliveParams(keepalive.ServerParameters{
            MaxConnectionIdle:     15 * time.Minute,
            MaxConnectionAge:      30 * time.Minute,
            MaxConnectionAgeGrace: 5 * time.Second,
            Time:                  5 * time.Minute,
            Timeout:               1 * time.Second,
        }),
        grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{
            MinTime:             5 * time.Second,
            PermitWithoutStream: true,
        }),
    )

    // --- Thông quan các dịch vụ (Register services) ---
    driverServer := driver.NewServer(repo, publisher, logger)
    driverv1.RegisterDriverServiceServer(srv, driverServer)

    // Register gRPC Health Check service phục vụ Kubernetes liveness/readiness probes
    healthSrv := health.NewServer()
    grpc_health_v1.RegisterHealthServer(srv, healthSrv)
    healthSrv.SetServingStatus("driver.v1.DriverService", grpc_health_v1.HealthCheckResponse_SERVING)

    // Đăng ký Reflection service cho phép các công cụ như grpcurl hay Postman truy vấn schema
    reflection.Register(srv)

    // --- Khởi tạo listener ---
    lis, err := net.Listen("tcp", ":50051")
    if err != nil {
        log.Fatalf("mục listen: %v", err)
    }

    log.Printf("gRPC server đang lắng nghe tại cổng :50051")

    // --- Chạy gRPC server trong goroutine ---
    go func() {
        if err := srv.Serve(lis); err != nil {
            log.Printf("va phải lỗi serve error: %v", err)
        }
    }()

    quit := make(chan os.Signal, 1)
    signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
    <-quit

    log.Println("đang ngắt gRPC server...")
    healthSrv.SetServingStatus("driver.v1.DriverService", grpc_health_v1.HealthCheckResponse_NOT_SERVING)
    srv.GracefulStop() // Chờ các RPC đang xử lý (in-flight) hoàn tất
    log.Println("hệ thống server đã ngừng tay")
}

// internal/client/driver_client.go
package client

import (
    "context"
    "crypto/tls"
    "crypto/x509"
    "log"
    "os"
    "time"

    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials"
    "google.golang.org/grpc/keepalive"

    driverv1 "github.com/yourorg/platform/gen/driver/v1"
)

func NewDriverClient(target string) (driverv1.DriverServiceClient, func(), error) {
    // Chứng chỉ bảo chứng của khách mTLS client credentials
    cert, err := tls.LoadX509KeyPair("certs/client.crt", "certs/client.key")
    if err != nil {
        return nil, nil, fmt.Errorf("lỗi tải client cert: %w", err)
    }

    caCert, _ := os.ReadFile("certs/ca.crt")
    caPool := x509.NewCertPool()
    caPool.AppendCertsFromPEM(caCert)

    creds := credentials.NewTLS(&tls.Config{
        Certificates: []tls.Certificate{cert},
        RootCAs:      caPool,
        MinVersion:   tls.VersionTLS13,
    })

    conn, err := grpc.NewClient(
        target,
        grpc.WithTransportCredentials(creds),
        // Điều chuyển mặc định dùng round-robin gánh vác san sẻ (load balancing) đánh dạt qua dàn multiple server instances 
        grpc.WithDefaultServiceConfig(`{"loadBalancingPolicy":"round_robin"}`),
        grpc.WithKeepaliveParams(keepalive.ClientParameters{
            Time:                10 * time.Minute,
            Timeout:             5 * time.Second,
            PermitWithoutStream: true,
        }),
    )
    if err != nil {
        return nil, nil, fmt.Errorf("kết nối dial đến %s thất bại: %w", target, err)
    }

    cleanup := func() { conn.Close() }
    return driverv1.NewDriverServiceClient(conn), cleanup, nil
}

// Ví dụ sử dụng gRPC client
func exampleGetDriver(ctx context.Context) {
    client, cleanup, err := NewDriverClient("dns:///driver-service:50051")
    if err != nil {
        log.Fatal(err)
    }
    defer cleanup()

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

    resp, err := client.GetDriver(ctx, &driverv1.GetDriverRequest{DriverId: "drv-abc123"})
    if err != nil {
        log.Printf("gặp lỗi ở khúc GetDriver error: %v", err)
        return
    }
    log.Printf("Thông tin driver %s: status=%s, location=(%f, %f)", resp.DriverId, resp.Status, resp.Latitude, resp.Longitude)
}

# Multi-stage Dockerfile tối ưu kích thước image
FROM golang:1.23-alpine AS builder
WORKDIR /app

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

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

FROM gcr.io/distroless/static-debian12
COPY --from=builder /bin/driver-service /driver-service
COPY certs/ /certs/

EXPOSE 50051
ENTRYPOINT ["/driver-service"]

# Lệnh file k8s/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: driver-service
spec:
  replicas: 3
  selector:
    matchLabels:
      app: driver-service
  template:
    metadata:
      labels:
        app: driver-service
    spec:
      containers:
        - name: driver-service
          image: yourorg/driver-service:latest
          ports:
            - containerPort: 50051
              name: grpc
          livenessProbe:
            grpc:
              port: 50051
              service: driver.v1.DriverService
            initialDelaySeconds: 10
            periodSeconds: 10
          readinessProbe:
            grpc:
              port: 50051
              service: driver.v1.DriverService
            initialDelaySeconds: 5
            periodSeconds: 5
          resources:
            requests:
              cpu: "100m"
              memory: "128Mi"
            limits:
              cpu: "500m"
              memory: "512Mi"

// ❌ Bad: Không cấu hình Deadline — nếu server bị treo, goroutine sẽ bị leak bộ nhớ
resp, err := client.GetDriver(context.Background(), req)

// ✅ Good: Luôn thiết lập timeout/deadline cho mọi RPC call
ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel()
resp, err := client.GetDriver(ctx, req)

// ❌ Bad: Đánh đồng tất cả các lỗi gRPC thành một loại duy nhất
if err != nil {
    return fmt.Errorf("gRPC call failed: %v", err)
}

// ✅ Good: Kiểm tra gRPC Status Code để quyết định chiến lược xử lý lỗi (ví dụ: retry)
if err != nil {
    st, _ := status.FromError(err)
    switch st.Code() {
    case codes.NotFound:
        return nil, ErrDriverNotFound
    case codes.Unavailable, codes.ResourceExhausted:
        // Lỗi tạm thời (retryable) — áp dụng exponential backoff
        return nil, ErrRetryable
    default:
        return nil, err
    }
}

// ❌ Bad: Nếu không cấu hình keepalive, các router/NAT firewall hoặc Load Balancer có thể chủ động ngắt kết nối TCP nhàn rỗi (idle connection) sau khoảng 4-10 phút.
// Phía client vẫn lầm tưởng kết nối đang hoạt động (connected) nhưng thực tế không nhận được dữ liệu.
// ✅ Good: Cấu hình tham số keepalive ở cả phía Client và Server (như hướng dẫn tại Mục 4 & 5 ở trên) để duy trì ping định kỳ.

// ❌ Bad: Cấu hình mặc định của gRPC client dùng policy pick_first — toàn bộ lưu lượng request sẽ chỉ dồn vào 1 Pod duy nhất
conn, _ := grpc.NewClient("dns:///driver-service:50051", grpc.WithTransportCredentials(creds))

// ✅ Good: Cấu hình policy round_robin để phân phối tải đều sang tất cả các Pod healthy
conn, _ := grpc.NewClient(
    "dns:///driver-service:50051",
    grpc.WithTransportCredentials(creds),
    grpc.WithDefaultServiceConfig(`{"loadBalancingPolicy":"round_robin"}`),
)