Phần 4: Tích Hợp API Golang & Microservices (Kratos & Dapr)

// CẤU TRÚC KÉM HIỆU NĂNG
message MatrixRow {
  repeated double distances = 1;
}

message DistanceMatrixResponse {
  repeated MatrixRow rows = 1; // Mảng 2 chiều lồng nhau
}

// CẤU TRÚC CHUẨN SẢN XUẤT TỐI ƯU BỘ NHỚ
message OptimizedMatrixResponse {
  int32 rows = 1;
  int32 cols = 2;
  repeated double data = 3 [json_name = "data"]; // Mảng 1 chiều phẳng
}

// Package main cung cấp API Gateway điều phối định tuyến chuẩn Go 1.25
package main

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"iter"
	"log/slog"
	"net/http"
	"os"
	"runtime"
	"sync/atomic"
	"time"

	"github.com/sony/gobreaker"
	"golang.org/x/sync/singleflight"
)

// Coordinates đại diện cho toạ độ địa lý WGS-84
type Coordinates struct {
	Lat float64 `json:"lat"`
	Lon float64 `json:"lon"`
}

// RouteResult chứa dữ liệu lộ trình trả về cho client
type RouteResult struct {
	DistanceMeters float64       `json:"distance_meters"`
	TravelDuration time.Duration `json:"duration"`
	EngineSource   string        `json:"engine_source"`
	IsCoalesced    bool          `json:"is_coalesced"`
}

// GatewayConfig chứa các thông số điều phối và ngưỡng bảo vệ
type GatewayConfig struct {
	GraphHopperURL   string
	MaxOpenRequests  uint32
	FailureRatio     float64
	CircuitTimeout   time.Duration
	HTTPRequestTimeout time.Duration
}

// EnterpriseRoutingGateway quản lý luồng điều phối với các tầng bảo vệ
type EnterpriseRoutingGateway struct {
	cfg        GatewayConfig
	httpClient *http.Client
	breaker    *gobreaker.CircuitBreaker
	flight     singleflight.Group
	logger     *slog.Logger
	metrics    struct {
		totalRequests     atomic.Uint64
		coalescedRequests atomic.Uint64
		circuitTrips      atomic.Uint64
	}
}

// NewEnterpriseRoutingGateway khởi tạo Gateway kèm Circuit Breaker và Cleanup Hook
func NewEnterpriseRoutingGateway(cfg GatewayConfig, logger *slog.Logger) (*EnterpriseRoutingGateway, error) {
	if cfg.HTTPRequestTimeout <= 0 {
		cfg.HTTPRequestTimeout = 1 * time.Second
	}
	if cfg.CircuitTimeout <= 0 {
		cfg.CircuitTimeout = 10 * time.Second
	}
	if cfg.FailureRatio <= 0 {
		cfg.FailureRatio = 0.5
	}

	transport := &http.Transport{
		MaxIdleConns:        1000,
		MaxIdleConnsPerHost: 200,
		IdleConnTimeout:     90 * time.Second,
	}

	gw := &EnterpriseRoutingGateway{
		cfg: cfg,
		httpClient: &http.Client{
			Transport: transport,
			Timeout:   cfg.HTTPRequestTimeout,
		},
		logger: logger,
	}

	// Cấu hình Cầu dao ngắt mạch Sony Gobreaker
	st := gobreaker.Settings{
		Name:        "GraphHopper-CircuitBreaker",
		MaxRequests: cfg.MaxOpenRequests,
		Interval:    30 * time.Second,
		Timeout:     cfg.CircuitTimeout,
		ReadyToTrip: func(counts gobreaker.Counts) bool {
			failureRatio := float64(counts.TotalFailures) / float64(counts.Requests)
			return counts.Requests >= 10 && failureRatio >= cfg.FailureRatio
		},
		OnStateChange: func(name string, from gobreaker.State, to gobreaker.State) {
			logger.Warn("Trạng thái Circuit Breaker thay đổi",
				slog.String("name", name),
				slog.String("from", from.String()),
				slog.String("to", to.String()),
			)
			if to == gobreaker.StateOpen {
				gw.metrics.circuitTrips.Add(1)
			}
		},
	}
	gw.breaker = gobreaker.NewCircuitBreaker(st)

	// Đăng ký giải phóng connection pool với Go 1.25 runtime.AddCleanup
	runtime.AddCleanup(gw, func(t *http.Transport) {
		t.CloseIdleConnections()
	}, transport)

	return gw, nil
}

// MatrixBatchIterator tạo Go 1.25 Iterator duyệt qua danh sách các cặp toạ độ
func MatrixBatchIterator(origins, dests []Coordinates) iter.Seq2[int, [2]Coordinates] {
	return func(yield func(int, [2]Coordinates) bool) {
		idx := 0
		for _, o := range origins {
			for _, d := range dests {
				if !yield(idx, [2]Coordinates{o, d}) {
					return
				}
				idx++
			}
		}
	}
}

// GetRoute tính toán đường đi qua cơ chế Singleflight và Circuit Breaker
func (g *EnterpriseRoutingGateway) GetRoute(ctx context.Context, from, to Coordinates) (*RouteResult, error) {
	g.metrics.totalRequests.Add(1)
	cacheKey := fmt.Sprintf("%.5f,%.5f->%.5f,%.5f", from.Lat, from.Lon, to.Lat, to.Lon)

	// Áp dụng Singleflight: Gộp các yêu cầu trùng lặp đang bay đồng thời
	rawResult, err, shared := g.flight.Do(cacheKey, func() (any, error) {
		// Bọc cuộc gọi trong Circuit Breaker
		result, cbErr := g.breaker.Execute(func() (any, error) {
			return g.executeRoutingRequest(ctx, from, to)
		})
		if cbErr != nil {
			return nil, cbErr
		}
		return result, nil
	})

	if shared {
		g.metrics.coalescedRequests.Add(1)
		g.logger.Debug("Yêu cầu đã được gộp thành công qua Singleflight", slog.String("key", cacheKey))
	}

	if err != nil {
		return nil, err
	}

	routeRes := rawResult.(*RouteResult)
	routeRes.IsCoalesced = shared
	return routeRes, nil
}

// executeRoutingRequest thực thi cuộc gọi HTTP thực tế tới GraphHopper
func (g *EnterpriseRoutingGateway) executeRoutingRequest(ctx context.Context, from, to Coordinates) (*RouteResult, error) {
	url := fmt.Sprintf("%s/route?point=%.6f,%.6f&point=%.6f,%.6f&profile=car&calc_points=false",
		g.cfg.GraphHopperURL, from.Lat, from.Lon, to.Lat, to.Lon)

	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
	if err != nil {
		return nil, err
	}

	resp, err := g.httpClient.Do(req)
	if err != nil {
		return nil, err
	}
	defer func() {
		// Bắt buộc phải đọc cạn và đóng Body để tái sử dụng TCP Connection
		io.Copy(io.Discard, resp.Body)
		resp.Body.Close()
	}()

	if resp.StatusCode != http.StatusOK {
		return nil, fmt.Errorf("backend trả về HTTP lỗi: %d", resp.StatusCode)
	}

	var payload struct {
		Paths []struct {
			Distance float64 `json:"distance"`
			Time     int64   `json:"time"`
		} `json:"paths"`
	}

	if err := json.NewDecoder(resp.Body).Decode(&payload); err != nil {
		return nil, err
	}

	if len(payload.Paths) == 0 {
		return nil, errors.New("không tìm thấy lộ trình hợp lệ")
	}

	return &RouteResult{
		DistanceMeters: payload.Paths[0].Distance,
		TravelDuration: time.Duration(payload.Paths[0].Time) * time.Millisecond,
		EngineSource:   "GraphHopper-Engine",
	}, nil
}

func main() {
	handler := slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo})
	logger := slog.New(handler)

	cfg := GatewayConfig{
		GraphHopperURL:     "http://localhost:8989",
		MaxOpenRequests:    5,
		FailureRatio:       0.5,
		CircuitTimeout:     5 * time.Second,
		HTTPRequestTimeout: 500 * time.Millisecond,
	}

	gateway, err := NewEnterpriseRoutingGateway(cfg, logger)
	if err != nil {
		logger.Error("Khởi tạo Gateway thất bại", slog.String("error", err.Error()))
		os.Exit(1)
	}

	p1 := Coordinates{Lat: 10.7769, Lon: 106.7009}
	p2 := Coordinates{Lat: 10.8231, Lon: 106.6297}

	// Mô phỏng 10 Goroutines đồng thời gọi cùng một lộ trình để kiểm chứng Singleflight
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()

	for i := 0; i < 5; i++ {
		go func(workerID int) {
			res, err := gateway.GetRoute(ctx, p1, p2)
			if err != nil {
				logger.Warn("Lỗi định tuyến", slog.Int("worker", workerID), slog.String("err", err.Error()))
				return
			}
			logger.Info("Nhận kết quả",
				slog.Int("worker", workerID),
				slog.Float64("khoang_cach_m", res.DistanceMeters),
				slog.Bool("coalesced", res.IsCoalesced),
			)
		}(i + 1)
	}

	time.Sleep(1 * time.Second)
	logger.Info("Thống kê Gateway",
		slog.Uint64("tong_so_yeu_cau", gateway.metrics.totalRequests.Load()),
		slog.Uint64("so_yeu_cau_duoc_gop", gateway.metrics.coalescedRequests.Load()),
		slog.Uint64("so_lan_ngat_mach", gateway.metrics.circuitTrips.Load()),
	)
}

> 🔥 **[Production Failure]: Rò Rỉ 120,000 Goroutines Và Cạn Kiệt Cổng Socket Dưới Tải Đột Biến**
> **Thời gian xảy ra:** 18:05 - 18:50 UTC+7, Ngày 03/10/2025.
> **Phạm vi ảnh hưởng:** Toàn bộ API Gateway định tuyến; 100% dịch vụ tìm xe và giao thức ăn nhanh bị trả về mã lỗi 504 Gateway Timeout; sập toàn bộ hệ thống điều phối cuốc xe.
> **Triệu chứng (Symptom):** Số lượng Goroutines trên máy chủ Gateway Go tăng vọt từ 2,500 lên hơn 120,000; máy chủ cạn kiệt cổng tạm thời (Ephemeral Port Exhaustion); lệnh `netstat` ghi nhận hơn 55,000 sockets ở trạng thái `TIME_WAIT` và `CLOSE_WAIT`.
> 
> **Nguyên nhân gốc rễ (Root Cause):**
> 1. Khi backend OSRM bị nghẽn nhẹ và bắt đầu trả về mã lỗi HTTP 500/503 cho một số truy vấn, đoạn mã HTTP Client của Gateway xử lý lỗi theo dạng:
>    ```go
>    resp, err := client.Do(req)
>    if resp.StatusCode != 200 { return nil, errors.New("lỗi") } // SAI LẦM!
>    defer resp.Body.Close()
>    ```
> 2. Đoạn mã trên đã return ngay khi `StatusCode != 200` mà **KHÔNG gọi `resp.Body.Close()` và KHÔNG đọc cạn dữ liệu `io.Copy(io.Discard, resp.Body)`**.
> 3. Trong giao thức HTTP/1.1, nếu không đọc cạn Body và không đóng Body, Go runtime không thể tái sử dụng (reuse) kết nối TCP bên dưới vào connection pool.
> 4. Mỗi request lỗi buộc Go phải mở một kết nối TCP mới toanh. Chỉ sau 10 phút, 65,535 cổng socket tạm thời trên máy chủ bị chiếm sạch, toàn bộ các luồng Goroutine tiếp theo bị đóng băng vĩnh viễn ở hàm `net.Dial()`.
> 
> 📊 **Hậu quả (Impact):** Tê liệt toàn bộ hệ thống điều phối trong 45 phút; hàng chục ngàn tài xế bị mất cuốc; thiệt hại ước tính 78.000 USD.
> 
> 📈 **Khắc phục & Kiến trúc phòng ngừa (Resolution & Prevention Architecture):**
> 1. **Khắc phục tức thời:** Khởi động lại toàn bộ Gateway pods; cấu hình lại kernel `net.ipv4.tcp_tw_reuse = 1` để tái sử dụng ngay lập tức các socket TIME_WAIT.
> 2. **Chuẩn hoá việc đóng Body với Linter nội bộ:** Bắt buộc áp dụng mẫu hình chuẩn trong mọi HTTP Client:
>    ```go
>    resp, err := client.Do(req)
>    if err != nil { return nil, err }
>    defer func() {
>        io.Copy(io.Discard, resp.Body)
>        resp.Body.Close()
>    }()
>    ```
> 3. **Trang bị Circuit Breaker Fail-Fast:** Triển khai `sony/gobreaker` để khi backend bắt đầu nấc cục, Gateway tự động ngắt kết nối trong 10 giây, chặn đứng nguy cơ bùng nổ Goroutine trước khi nó chạm tới tầng mạng.