docs(proxy,loadbalance): 翻译英文注释为中文
将 internal/proxy 和 internal/loadbalance 模块的英文注释翻译为中文， 保持项目注释语言一致性。 Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
parent
6ae7e32ef1
commit
0128898136
@ -1,10 +1,10 @@
|
||||
// Package loadbalance provides load balancing algorithms for the Lolly HTTP server.
|
||||
// Package loadbalance 负载均衡包为 Lolly HTTP 服务器提供负载均衡算法。
|
||||
//
|
||||
// This package implements various load balancing strategies including round-robin,
|
||||
// weighted round-robin, least connections, and IP hash. All implementations are
|
||||
// concurrency-safe using atomic operations.
|
||||
// 本包实现了多种负载均衡策略，包括轮询（round-robin）、
|
||||
// 权重轮询（weighted round-robin）、最少连接（least connections）和 IP 哈希（IP hash）。
|
||||
// 所有实现都使用原子操作来保证并发安全。
|
||||
//
|
||||
// Example usage:
|
||||
// 使用示例：
|
||||
//
|
||||
// targets := []*Target{
|
||||
// {URL: "http://backend1:8080", Weight: 1, Healthy: true},
|
||||
@ -22,84 +22,83 @@ import (
|
||||
"sync/atomic"
|
||||
)
|
||||
|
||||
// Target represents a backend server target for load balancing.
|
||||
// All fields are designed for concurrent access using atomic operations
|
||||
// where applicable.
|
||||
// Target 表示负载均衡的后端服务器目标。
|
||||
// 所有字段都设计为使用原子操作进行并发访问（如适用）。
|
||||
type Target struct {
|
||||
// URL is the target address, e.g., "http://backend1:8080"
|
||||
// URL 是目标地址，例如 "http://backend1:8080"
|
||||
URL string
|
||||
|
||||
// Weight is the weight of this target for weighted algorithms.
|
||||
// Higher weight means more requests will be routed to this target.
|
||||
// Weight 是此目标在权重算法中的权重值。
|
||||
// 权重越高，表示有更多请求会被路由到此目标。
|
||||
Weight int
|
||||
|
||||
// Healthy indicates whether this target is healthy and available.
|
||||
// Use atomic operations to read/write this field concurrently.
|
||||
// Healthy 表示此目标是否健康可用。
|
||||
// 并发读写此字段时应使用原子操作。
|
||||
Healthy bool
|
||||
|
||||
// Connections tracks the current number of active connections.
|
||||
// Use atomic operations to modify this field concurrently.
|
||||
// Connections 跟踪当前活跃连接数。
|
||||
// 并发修改此字段时应使用原子操作。
|
||||
Connections int64
|
||||
}
|
||||
|
||||
// Balancer is the interface for load balancing algorithms.
|
||||
// Implementations must be safe for concurrent use.
|
||||
// Balancer 是负载均衡算法的接口。
|
||||
// 实现必须是并发安全的。
|
||||
type Balancer interface {
|
||||
// Select chooses a target from the provided list based on the
|
||||
// algorithm's strategy. Returns nil if no healthy targets are available.
|
||||
// Select 根据算法策略从提供的列表中选择一个目标。
|
||||
// 如果没有健康目标可用，返回 nil。
|
||||
Select(targets []*Target) *Target
|
||||
}
|
||||
|
||||
// RoundRobin implements simple round-robin load balancing.
|
||||
// It distributes requests evenly across all healthy targets in sequence.
|
||||
// RoundRobin 实现简单的轮询负载均衡。
|
||||
// 它按顺序将请求均匀分配到所有健康目标上。
|
||||
type RoundRobin struct {
|
||||
// counter is incremented atomically for each request
|
||||
// counter 原子地为每个请求递增
|
||||
counter uint64
|
||||
}
|
||||
|
||||
// NewRoundRobin creates a new round-robin load balancer.
|
||||
// NewRoundRobin 创建一个新的轮询负载均衡器。
|
||||
func NewRoundRobin() *RoundRobin {
|
||||
return &RoundRobin{}
|
||||
}
|
||||
|
||||
// Select chooses the next target in round-robin order.
|
||||
// Only healthy targets are considered. Returns nil if no healthy targets exist.
|
||||
// Select 选择轮询顺序中的下一个目标。
|
||||
// 只考虑健康目标。如果没有健康目标则返回 nil。
|
||||
func (r *RoundRobin) Select(targets []*Target) *Target {
|
||||
healthy := filterHealthy(targets)
|
||||
if len(healthy) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Atomically increment and get the counter value
|
||||
// 原子地递增并获取计数器值
|
||||
idx := atomic.AddUint64(&r.counter, 1) - 1
|
||||
return healthy[idx%uint64(len(healthy))]
|
||||
}
|
||||
|
||||
// WeightedRoundRobin implements weighted round-robin load balancing.
|
||||
// Targets with higher weights receive proportionally more requests.
|
||||
// WeightedRoundRobin 实现权重轮询负载均衡。
|
||||
// 权重越高的目标接收成比例更多的请求。
|
||||
type WeightedRoundRobin struct {
|
||||
// counter is incremented atomically for each request
|
||||
// counter 原子地为每个请求递增
|
||||
counter uint64
|
||||
}
|
||||
|
||||
// NewWeightedRoundRobin creates a new weighted round-robin load balancer.
|
||||
// NewWeightedRoundRobin 创建一个新的权重轮询负载均衡器。
|
||||
func NewWeightedRoundRobin() *WeightedRoundRobin {
|
||||
return &WeightedRoundRobin{}
|
||||
}
|
||||
|
||||
// Select chooses a target based on weight distribution.
|
||||
// Only healthy targets are considered. Returns nil if no healthy targets exist.
|
||||
// Select 基于权重分布选择目标。
|
||||
// 只考虑健康目标。如果没有健康目标则返回 nil。
|
||||
func (w *WeightedRoundRobin) Select(targets []*Target) *Target {
|
||||
healthy := filterHealthy(targets)
|
||||
if len(healthy) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Calculate total weight
|
||||
// 计算总权重
|
||||
totalWeight := 0
|
||||
for _, t := range healthy {
|
||||
if t.Weight <= 0 {
|
||||
totalWeight += 1 // Minimum weight of 1
|
||||
totalWeight += 1 // 最小权重为 1
|
||||
} else {
|
||||
totalWeight += t.Weight
|
||||
}
|
||||
@ -109,11 +108,11 @@ func (w *WeightedRoundRobin) Select(targets []*Target) *Target {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Use atomic counter to determine position in weight distribution
|
||||
// 使用原子计数器确定权重分布中的位置
|
||||
idx := atomic.AddUint64(&w.counter, 1) - 1
|
||||
pos := int(idx % uint64(totalWeight))
|
||||
|
||||
// Find target at the calculated position
|
||||
// 找到计算位置处的目标
|
||||
currentWeight := 0
|
||||
for _, t := range healthy {
|
||||
weight := t.Weight
|
||||
@ -126,21 +125,21 @@ func (w *WeightedRoundRobin) Select(targets []*Target) *Target {
|
||||
}
|
||||
}
|
||||
|
||||
// Fallback to last target (should not reach here)
|
||||
// 回退到最后一个目标（不应到达这里）
|
||||
return healthy[len(healthy)-1]
|
||||
}
|
||||
|
||||
// LeastConnections implements least connections load balancing.
|
||||
// It selects the target with the fewest active connections.
|
||||
// LeastConnections 实现最少连接负载均衡。
|
||||
// 它选择活跃连接数最少的目标。
|
||||
type LeastConnections struct{}
|
||||
|
||||
// NewLeastConnections creates a new least-connections load balancer.
|
||||
// NewLeastConnections 创建一个新的最少连接负载均衡器。
|
||||
func NewLeastConnections() *LeastConnections {
|
||||
return &LeastConnections{}
|
||||
}
|
||||
|
||||
// Select chooses the target with the minimum connection count.
|
||||
// Only healthy targets are considered. Returns nil if no healthy targets exist.
|
||||
// Select 选择连接数最少的目标。
|
||||
// 只考虑健康目标。如果没有健康目标则返回 nil。
|
||||
func (l *LeastConnections) Select(targets []*Target) *Target {
|
||||
var selected *Target
|
||||
var minConns int64 = -1
|
||||
@ -150,7 +149,7 @@ func (l *LeastConnections) Select(targets []*Target) *Target {
|
||||
continue
|
||||
}
|
||||
|
||||
// Atomically read the connection count
|
||||
// 原子地读取连接计数
|
||||
conns := atomic.LoadInt64(&t.Connections)
|
||||
|
||||
if selected == nil || conns < minConns {
|
||||
@ -162,31 +161,31 @@ func (l *LeastConnections) Select(targets []*Target) *Target {
|
||||
return selected
|
||||
}
|
||||
|
||||
// IPHash implements IP hash-based load balancing.
|
||||
// It consistently routes requests from the same client IP to the same target.
|
||||
// IPHash 实现基于 IP 哈希的负载均衡。
|
||||
// 它将来自同一客户端 IP 的请求始终路由到同一目标。
|
||||
type IPHash struct{}
|
||||
|
||||
// NewIPHash creates a new IP hash load balancer.
|
||||
// NewIPHash 创建一个新的 IP 哈希负载均衡器。
|
||||
func NewIPHash() *IPHash {
|
||||
return &IPHash{}
|
||||
}
|
||||
|
||||
// Select chooses a target based on the hash of the client IP.
|
||||
// Only healthy targets are considered. Returns nil if no healthy targets exist.
|
||||
// The clientIP parameter should be the client's IP address as a string.
|
||||
// Select 基于客户端 IP 的哈希值选择目标。
|
||||
// 只考虑健康目标。如果没有健康目标则返回 nil。
|
||||
// clientIP 参数应该是客户端的 IP 地址字符串。
|
||||
func (i *IPHash) Select(targets []*Target) *Target {
|
||||
return i.SelectByIP(targets, "")
|
||||
}
|
||||
|
||||
// SelectByIP chooses a target based on the hash of the provided IP address.
|
||||
// Only healthy targets are considered. Returns nil if no healthy targets exist.
|
||||
// SelectByIP 基于提供的 IP 地址的哈希值选择目标。
|
||||
// 只考虑健康目标。如果没有健康目标则返回 nil。
|
||||
func (i *IPHash) SelectByIP(targets []*Target, clientIP string) *Target {
|
||||
healthy := filterHealthy(targets)
|
||||
if len(healthy) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Hash the client IP
|
||||
// 对客户端 IP 进行哈希
|
||||
h := fnv.New64a()
|
||||
h.Write([]byte(clientIP))
|
||||
hash := h.Sum64()
|
||||
@ -195,8 +194,8 @@ func (i *IPHash) SelectByIP(targets []*Target, clientIP string) *Target {
|
||||
return healthy[idx]
|
||||
}
|
||||
|
||||
// filterHealthy returns a new slice containing only healthy targets.
|
||||
// This is a helper function used by load balancing implementations.
|
||||
// filterHealthy 返回仅包含健康目标的新切片。
|
||||
// 这是负载均衡实现使用的辅助函数。
|
||||
func filterHealthy(targets []*Target) []*Target {
|
||||
healthy := make([]*Target, 0, len(targets))
|
||||
for _, t := range targets {
|
||||
@ -207,33 +206,32 @@ func filterHealthy(targets []*Target) []*Target {
|
||||
return healthy
|
||||
}
|
||||
|
||||
// IncrementConnections atomically increments the connection count for a target.
|
||||
// This should be called when a new connection is established.
|
||||
// IncrementConnections 原子地增加目标的连接计数。
|
||||
// 当新连接建立时应调用此函数。
|
||||
func IncrementConnections(t *Target) {
|
||||
atomic.AddInt64(&t.Connections, 1)
|
||||
}
|
||||
|
||||
// DecrementConnections atomically decrements the connection count for a target.
|
||||
// This should be called when a connection is closed.
|
||||
// DecrementConnections 原子地减少目标的连接计数。
|
||||
// 当连接关闭时应调用此函数。
|
||||
func DecrementConnections(t *Target) {
|
||||
atomic.AddInt64(&t.Connections, -1)
|
||||
}
|
||||
|
||||
// IsHealthy atomically reads the health status of a target.
|
||||
// IsHealthy 原子地读取目标的健康状态。
|
||||
func IsHealthy(t *Target) bool {
|
||||
// Healthy is a bool, which is safe to read without atomic operations
|
||||
// but for consistency with the setter, we could use atomic
|
||||
// For bool, simple read is safe in Go's memory model
|
||||
// Healthy 是 bool 类型，在 Go 的内存模型中无需原子操作即可安全读取
|
||||
// 但为了与 setter 保持一致，我们可以使用原子操作
|
||||
// 对于 bool，简单的读取是安全的
|
||||
return t.Healthy
|
||||
}
|
||||
|
||||
// SetHealthy atomically sets the health status of a target.
|
||||
// Note: In Go, bool operations are not directly atomic.
|
||||
// This function provides a synchronized way to update health status.
|
||||
// For true atomic operations on bool, consider using atomic.Bool (Go 1.19+)
|
||||
// or sync.RWMutex. For this implementation, we use direct assignment
|
||||
// which is typically sufficient when combined with proper synchronization
|
||||
// at the caller level.
|
||||
// SetHealthy 原子地设置目标的健康状态。
|
||||
// 注意：在 Go 中，bool 操作不能直接是原子的。
|
||||
// 此函数提供了同步更新健康状态的方式。
|
||||
// 对于 bool 的真正原子操作，请考虑使用 atomic.Bool（Go 1.19+）
|
||||
// 或 sync.RWMutex。对于本实现，我们使用直接赋值
|
||||
// 当与调用层的适当同步结合时，这通常是足够的。
|
||||
func SetHealthy(t *Target, healthy bool) {
|
||||
t.Healthy = healthy
|
||||
}
|
||||
|
||||
@ -1,8 +1,8 @@
|
||||
// Package proxy provides reverse proxy functionality for the Lolly HTTP server.
|
||||
//
|
||||
// This file implements health checking for backend targets, supporting both
|
||||
// active health checks (periodic HTTP probes) and passive health checks
|
||||
// (marking targets unhealthy based on observed failures).
|
||||
// 此文件实现了针对后端目标的健康检查功能，支持
|
||||
// 主动健康检查（定期 HTTP 探测）和被动健康检查
|
||||
//（基于观察到的失败标记目标为不健康）。
|
||||
//
|
||||
//go:generate go test -v ./...
|
||||
package proxy
|
||||
@ -18,14 +18,14 @@ import (
|
||||
"rua.plus/lolly/internal/loadbalance"
|
||||
)
|
||||
|
||||
// HealthChecker performs health checks on backend targets.
|
||||
// It supports both active (periodic HTTP probes) and passive (failure-based)
|
||||
// health checking modes.
|
||||
// HealthChecker 对后端目标执行健康检查。
|
||||
// 它支持主动（定期 HTTP 探测）和被动（基于失败的）
|
||||
// 两种健康检查模式。
|
||||
//
|
||||
// The checker runs in a background goroutine when started, periodically
|
||||
// sending HTTP GET requests to each target's health check endpoint.
|
||||
// Targets responding with 2xx status codes are marked as healthy;
|
||||
// timeouts, connection failures, or non-2xx responses mark them as unhealthy.
|
||||
// 当启动后，检查器在后台 goroutine 中运行，定期
|
||||
// 向每个目标的健康检查端点发送 HTTP GET 请求。
|
||||
// 返回 2xx 状态码的目标被标记为健康；
|
||||
// 超时、连接失败或非 2xx 响应将其标记为不健康。
|
||||
//
|
||||
// Example usage:
|
||||
//
|
||||
@ -54,15 +54,15 @@ type HealthChecker struct {
|
||||
mu sync.RWMutex
|
||||
}
|
||||
|
||||
// NewHealthChecker creates a new HealthChecker with the specified targets and configuration.
|
||||
// The configuration defines the check interval, timeout, and health check path.
|
||||
// NewHealthChecker 使用指定的目标和配置创建一个新的 HealthChecker。
|
||||
// 配置定义了检查间隔、超时和健康检查路径。
|
||||
//
|
||||
// Default values are applied if not specified in the config:
|
||||
// - Interval: 10 seconds
|
||||
// - Timeout: 5 seconds
|
||||
// 如果配置中未指定，将应用默认值：
|
||||
// - Interval: 10 秒
|
||||
// - Timeout: 5 秒
|
||||
// - Path: "/health"
|
||||
//
|
||||
// The returned HealthChecker is not started; call Start() to begin health checks.
|
||||
// 返回的 HealthChecker 尚未启动；调用 Start() 开始健康检查。
|
||||
func NewHealthChecker(targets []*loadbalance.Target, cfg *config.HealthCheckConfig) *HealthChecker {
|
||||
interval := cfg.Interval
|
||||
if interval <= 0 {
|
||||
@ -92,11 +92,11 @@ func NewHealthChecker(targets []*loadbalance.Target, cfg *config.HealthCheckConf
|
||||
}
|
||||
}
|
||||
|
||||
// Start begins the background health check process.
|
||||
// It launches a goroutine that periodically checks all targets at the configured interval.
|
||||
// Start is idempotent; calling it on an already running checker has no effect.
|
||||
// Start 启动后台健康检查进程。
|
||||
// 它启动一个 goroutine，按照配置的间隔定期检查所有目标。
|
||||
// Start 是幂等的；在已运行的检查器上调用它不会产生任何效果。
|
||||
//
|
||||
// The health check process continues until Stop() is called.
|
||||
// 健康检查进程将持续运行，直到调用 Stop()。
|
||||
func (h *HealthChecker) Start() {
|
||||
if h.running.Load() {
|
||||
return
|
||||
@ -106,9 +106,9 @@ func (h *HealthChecker) Start() {
|
||||
go h.run()
|
||||
}
|
||||
|
||||
// Stop halts the background health check process.
|
||||
// It signals the background goroutine to stop and waits for it to complete.
|
||||
// Stop is idempotent; calling it on a stopped checker has no effect.
|
||||
// Stop 停止后台健康检查进程。
|
||||
// 它向后台 goroutine 发送停止信号并等待其完成。
|
||||
// Stop 是幂等的；在已停止的检查器上调用它不会产生任何效果。
|
||||
func (h *HealthChecker) Stop() {
|
||||
if !h.running.Load() {
|
||||
return
|
||||
@ -118,11 +118,11 @@ func (h *HealthChecker) Stop() {
|
||||
close(h.stopCh)
|
||||
}
|
||||
|
||||
// run is the main health check loop running in a background goroutine.
|
||||
// It performs an initial check on all targets, then enters a loop that
|
||||
// checks targets at regular intervals until stopped.
|
||||
// run 是在后台 goroutine 中运行的主要健康检查循环。
|
||||
// 它对所有目标执行初始检查，然后进入循环，
|
||||
// 以固定间隔检查目标，直到被停止。
|
||||
func (h *HealthChecker) run() {
|
||||
// Perform initial health check
|
||||
// 执行初始健康检查
|
||||
h.checkAll()
|
||||
|
||||
ticker := time.NewTicker(h.interval)
|
||||
@ -138,8 +138,8 @@ func (h *HealthChecker) run() {
|
||||
}
|
||||
}
|
||||
|
||||
// checkAll performs health checks on all configured targets.
|
||||
// It checks each target concurrently using goroutines to minimize latency.
|
||||
// checkAll 对所有配置的目标执行健康检查。
|
||||
// 它使用 goroutines 并发检查每个目标以最小化延迟。
|
||||
func (h *HealthChecker) checkAll() {
|
||||
var wg sync.WaitGroup
|
||||
|
||||
@ -154,23 +154,23 @@ func (h *HealthChecker) checkAll() {
|
||||
wg.Wait()
|
||||
}
|
||||
|
||||
// checkTarget performs a health check on a single target.
|
||||
// It sends an HTTP GET request to the target's health check endpoint
|
||||
// and updates the target's Healthy status based on the response.
|
||||
// checkTarget 对单个目标执行健康检查。
|
||||
// 它向目标的健康检查端点发送 HTTP GET 请求
|
||||
// 并根据响应更新目标的 Healthy 状态。
|
||||
//
|
||||
// A target is considered healthy if:
|
||||
// - The HTTP request succeeds
|
||||
// - The response status code is between 200 and 299
|
||||
// 目标被认为健康，如果满足以下条件：
|
||||
// - HTTP 请求成功
|
||||
// - 响应状态码在 200 到 299 之间
|
||||
//
|
||||
// A target is marked unhealthy if:
|
||||
// - The connection fails
|
||||
// - The request times out
|
||||
// - The response status code is not 2xx
|
||||
// 目标被标记为不健康，如果满足以下条件：
|
||||
// - 连接失败
|
||||
// - 请求超时
|
||||
// - 响应状态码不是 2xx
|
||||
func (h *HealthChecker) checkTarget(target *loadbalance.Target) {
|
||||
// Build health check URL
|
||||
// 构建健康检查 URL
|
||||
url := target.URL + h.path
|
||||
|
||||
// Prepare request and response
|
||||
// 准备请求和响应
|
||||
req := fasthttp.AcquireRequest()
|
||||
resp := fasthttp.AcquireResponse()
|
||||
defer fasthttp.ReleaseRequest(req)
|
||||
@ -180,16 +180,16 @@ func (h *HealthChecker) checkTarget(target *loadbalance.Target) {
|
||||
req.Header.SetMethod(fasthttp.MethodGet)
|
||||
req.Header.Set("User-Agent", "Lolly-HealthChecker/1.0")
|
||||
|
||||
// Perform health check with timeout
|
||||
// 执行带超时的健康检查
|
||||
err := h.client.DoTimeout(req, resp, h.timeout)
|
||||
|
||||
if err != nil {
|
||||
// Connection failed or timeout - mark as unhealthy
|
||||
// 连接失败或超时 - 标记为不健康
|
||||
loadbalance.SetHealthy(target, false)
|
||||
return
|
||||
}
|
||||
|
||||
// Check status code - 2xx is healthy
|
||||
// 检查状态码 - 2xx 为健康
|
||||
statusCode := resp.StatusCode()
|
||||
if statusCode >= 200 && statusCode < 300 {
|
||||
loadbalance.SetHealthy(target, true)
|
||||
@ -198,44 +198,44 @@ func (h *HealthChecker) checkTarget(target *loadbalance.Target) {
|
||||
}
|
||||
}
|
||||
|
||||
// MarkUnhealthy marks a target as unhealthy.
|
||||
// This method is intended for passive health checking, where the proxy
|
||||
// marks targets as unhealthy based on observed failures during request handling.
|
||||
// MarkUnhealthy 将目标标记为不健康。
|
||||
// 此方法用于被动健康检查，代理根据请求处理过程中
|
||||
// 观察到的失败将目标标记为不健康。
|
||||
//
|
||||
// Example usage in proxy error handling:
|
||||
// 在代理错误处理中的使用示例：
|
||||
//
|
||||
// if err := forwardRequest(target, req, resp); err != nil {
|
||||
// healthChecker.MarkUnhealthy(target)
|
||||
// // Try another target or return error
|
||||
// // 尝试其他目标或返回错误
|
||||
// }
|
||||
//
|
||||
// Note: To mark a target as healthy again, the active health check
|
||||
// must succeed. There is no MarkHealthy method - health status can only
|
||||
// be positively restored through successful health checks.
|
||||
// 注意：要再次将目标标记为健康，主动健康检查
|
||||
// 必须成功。没有 MarkHealthy 方法 - 健康状态只能通过
|
||||
// 成功的健康检查积极恢复。
|
||||
func (h *HealthChecker) MarkUnhealthy(target *loadbalance.Target) {
|
||||
loadbalance.SetHealthy(target, false)
|
||||
}
|
||||
|
||||
// IsRunning returns true if the health checker is currently running.
|
||||
// IsRunning 如果健康检查器当前正在运行，则返回 true。
|
||||
func (h *HealthChecker) IsRunning() bool {
|
||||
return h.running.Load()
|
||||
}
|
||||
|
||||
// GetInterval returns the configured check interval.
|
||||
// GetInterval 返回配置的检查间隔。
|
||||
func (h *HealthChecker) GetInterval() time.Duration {
|
||||
h.mu.RLock()
|
||||
defer h.mu.RUnlock()
|
||||
return h.interval
|
||||
}
|
||||
|
||||
// GetTimeout returns the configured check timeout.
|
||||
// GetTimeout 返回配置的检查超时时间。
|
||||
func (h *HealthChecker) GetTimeout() time.Duration {
|
||||
h.mu.RLock()
|
||||
defer h.mu.RUnlock()
|
||||
return h.timeout
|
||||
}
|
||||
|
||||
// GetPath returns the configured health check path.
|
||||
// GetPath 返回配置的健康检查路径。
|
||||
func (h *HealthChecker) GetPath() string {
|
||||
h.mu.RLock()
|
||||
defer h.mu.RUnlock()
|
||||
|
||||
@ -1,10 +1,9 @@
|
||||
// Package proxy provides reverse proxy functionality for the Lolly HTTP server.
|
||||
// Package proxy 反向代理包，为 Lolly HTTP 服务器提供反向代理功能。
|
||||
//
|
||||
// This package implements a high-performance reverse proxy using fasthttp.HostClient
|
||||
// for connection pooling and automatic keep-alive management. It supports load balancing,
|
||||
// WebSocket forwarding, custom headers, and comprehensive timeout configurations.
|
||||
// 该包使用 fasthttp.HostClient 实现高性能反向代理，支持连接池和自动 keep-alive 管理。
|
||||
// 支持负载均衡、WebSocket 转发、自定义请求头/响应头和全面的超时配置。
|
||||
//
|
||||
// Example usage:
|
||||
// 使用示例：
|
||||
//
|
||||
// targets := []*loadbalance.Target{
|
||||
// {URL: "http://backend1:8080", Weight: 1, Healthy: true},
|
||||
@ -26,7 +25,7 @@
|
||||
// log.Fatal(err)
|
||||
// }
|
||||
//
|
||||
// // Use p.ServeHTTP as fasthttp request handler
|
||||
// // 使用 p.ServeHTTP 作为 fasthttp 请求处理器
|
||||
//
|
||||
//go:generate go test -v ./...
|
||||
package proxy
|
||||
@ -43,8 +42,8 @@ import (
|
||||
"rua.plus/lolly/internal/loadbalance"
|
||||
)
|
||||
|
||||
// Proxy represents a reverse proxy instance that forwards HTTP requests to backend targets.
|
||||
// It manages connection pools for each target and provides load balancing capabilities.
|
||||
// Proxy 表示反向代理实例，负责将 HTTP 请求转发到后端目标。
|
||||
// 它为每个后端目标管理连接池，并提供负载均衡功能。
|
||||
type Proxy struct {
|
||||
targets []*loadbalance.Target
|
||||
clients map[string]*fasthttp.HostClient // key: target URL
|
||||
@ -53,16 +52,16 @@ type Proxy struct {
|
||||
mu sync.RWMutex
|
||||
}
|
||||
|
||||
// NewProxy creates a new reverse proxy instance with the given configuration and targets.
|
||||
// It initializes the load balancer based on the config and creates HostClients for each target.
|
||||
// NewProxy 使用给定的配置和后台目标创建一个新的反向代理实例。
|
||||
// 它根据配置初始化负载均衡器，并为每个后端目标创建 HostClient。
|
||||
//
|
||||
// Parameters:
|
||||
// - cfg: Proxy configuration including timeouts, headers, and load balancing strategy
|
||||
// - targets: List of backend targets to proxy requests to
|
||||
// 参数：
|
||||
// - cfg: 代理配置，包括超时时间、请求头和负载均衡策略
|
||||
// - targets: 要代理请求的后端目标列表
|
||||
//
|
||||
// Returns:
|
||||
// - *Proxy: Configured proxy instance ready to serve requests
|
||||
// - error: Non-nil if initialization fails (invalid config, no healthy targets, etc.)
|
||||
// 返回值：
|
||||
// - *Proxy: 配置完成并可处理请求的代理实例
|
||||
// - error: 初始化失败时非空（无效配置、没有健康目标等）
|
||||
func NewProxy(cfg *config.ProxyConfig, targets []*loadbalance.Target) (*Proxy, error) {
|
||||
if cfg == nil {
|
||||
return nil, errors.New("proxy config is nil")
|
||||
@ -72,7 +71,7 @@ func NewProxy(cfg *config.ProxyConfig, targets []*loadbalance.Target) (*Proxy, e
|
||||
return nil, errors.New("no proxy targets provided")
|
||||
}
|
||||
|
||||
// Create balancer based on configuration
|
||||
// 根据配置创建负载均衡器
|
||||
balancer, err := createBalancer(cfg.LoadBalance)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@ -85,7 +84,7 @@ func NewProxy(cfg *config.ProxyConfig, targets []*loadbalance.Target) (*Proxy, e
|
||||
config: cfg,
|
||||
}
|
||||
|
||||
// Initialize HostClient for each target
|
||||
// 为每个后端目标初始化 HostClient
|
||||
for _, target := range targets {
|
||||
if target.URL == "" {
|
||||
continue
|
||||
@ -98,7 +97,7 @@ func NewProxy(cfg *config.ProxyConfig, targets []*loadbalance.Target) (*Proxy, e
|
||||
return p, nil
|
||||
}
|
||||
|
||||
// createBalancer creates a load balancer based on the configured algorithm.
|
||||
// createBalancer 根据配置的算法创建负载均衡器。
|
||||
func createBalancer(algorithm string) (loadbalance.Balancer, error) {
|
||||
switch algorithm {
|
||||
case "round_robin", "":
|
||||
@ -114,9 +113,9 @@ func createBalancer(algorithm string) (loadbalance.Balancer, error) {
|
||||
}
|
||||
}
|
||||
|
||||
// createHostClient creates a fasthttp.HostClient for a target URL.
|
||||
// createHostClient 为后台目标 URL 创建 fasthttp.HostClient。
|
||||
func createHostClient(targetURL string, timeout config.ProxyTimeout) *fasthttp.HostClient {
|
||||
// Parse host and scheme from target URL
|
||||
// 从目标 URL 解析主机和协议
|
||||
addr := targetURL
|
||||
isTLS := false
|
||||
|
||||
@ -127,7 +126,7 @@ func createHostClient(targetURL string, timeout config.ProxyTimeout) *fasthttp.H
|
||||
isTLS = true
|
||||
}
|
||||
|
||||
// Remove path if present, keep only host:port
|
||||
// 如果存在路径则移除，只保留 host:port
|
||||
if idx := strings.Index(addr, "/"); idx != -1 {
|
||||
addr = addr[:idx]
|
||||
}
|
||||
@ -148,52 +147,52 @@ func createHostClient(targetURL string, timeout config.ProxyTimeout) *fasthttp.H
|
||||
return client
|
||||
}
|
||||
|
||||
// ServeHTTP handles the incoming HTTP request by forwarding it to a selected backend target.
|
||||
// It implements the fasthttp request handler interface.
|
||||
// ServeHTTP 通过将传入的 HTTP 请求转发到选定的后端目标来处理请求。
|
||||
// 实现了 fasthttp 请求处理器接口。
|
||||
//
|
||||
// The method:
|
||||
// 1. Selects a target using load balancing
|
||||
// 2. Prepares the request (modifies headers)
|
||||
// 3. Forwards the request to the backend
|
||||
// 4. Copies the response back to the client
|
||||
// 处理流程：
|
||||
// 1. 使用负载均衡选择目标
|
||||
// 2. 准备请求（修改请求头）
|
||||
// 3. 将请求转发到后端
|
||||
// 4. 将响应复制回客户端
|
||||
//
|
||||
// If no healthy targets are available, returns 502 Bad Gateway.
|
||||
// If the backend request fails, returns appropriate error response.
|
||||
// 如果没有可用的健康目标，返回 502 Bad Gateway。
|
||||
// 如果后端请求失败，返回相应的错误响应。
|
||||
func (p *Proxy) ServeHTTP(ctx *fasthttp.RequestCtx) {
|
||||
// Select target using load balancer
|
||||
// 使用负载均衡器选择目标
|
||||
target := p.selectTarget(ctx)
|
||||
if target == nil {
|
||||
ctx.Error("Bad Gateway: no healthy upstream", fasthttp.StatusBadGateway)
|
||||
return
|
||||
}
|
||||
|
||||
// Get the client for selected target
|
||||
// 获取所选目标的客户端
|
||||
client := p.getClient(target.URL)
|
||||
if client == nil {
|
||||
ctx.Error("Bad Gateway: upstream client unavailable", fasthttp.StatusBadGateway)
|
||||
return
|
||||
}
|
||||
|
||||
// Increment connection count for least_connections tracking
|
||||
// 增加连接计数（用于最少连接数负载均衡）
|
||||
loadbalance.IncrementConnections(target)
|
||||
defer loadbalance.DecrementConnections(target)
|
||||
|
||||
// Check if this is a WebSocket upgrade request
|
||||
// 检查是否为 WebSocket 升级请求
|
||||
if isWebSocketRequest(ctx) {
|
||||
p.handleWebSocket(ctx, target, client)
|
||||
return
|
||||
}
|
||||
|
||||
// Prepare request
|
||||
// 准备请求
|
||||
req := &ctx.Request
|
||||
|
||||
// Modify request headers
|
||||
// 修改请求头
|
||||
p.modifyRequestHeaders(ctx, target)
|
||||
|
||||
// Perform the proxy request
|
||||
// 执行代理请求
|
||||
err := client.Do(req, &ctx.Response)
|
||||
if err != nil {
|
||||
// Handle different error types
|
||||
// 处理不同类型的错误
|
||||
if errors.Is(err, fasthttp.ErrTimeout) {
|
||||
ctx.Error("Gateway Timeout", fasthttp.StatusGatewayTimeout)
|
||||
} else if errors.Is(err, fasthttp.ErrConnectionClosed) {
|
||||
@ -204,13 +203,13 @@ func (p *Proxy) ServeHTTP(ctx *fasthttp.RequestCtx) {
|
||||
return
|
||||
}
|
||||
|
||||
// Modify response headers
|
||||
// 修改响应头
|
||||
p.modifyResponseHeaders(ctx)
|
||||
}
|
||||
|
||||
// selectTarget selects a backend target using the configured load balancer.
|
||||
// It extracts the client IP from the request for IP hash balancing.
|
||||
// Returns nil if no healthy targets are available.
|
||||
// selectTarget 使用配置的负载均衡器选择后端目标。
|
||||
// 对于 IP 哈希负载均衡，从请求中提取客户端 IP。
|
||||
// 如果没有可用的健康目标则返回 nil。
|
||||
func (p *Proxy) selectTarget(ctx *fasthttp.RequestCtx) *loadbalance.Target {
|
||||
p.mu.RLock()
|
||||
balancer := p.balancer
|
||||
@ -221,7 +220,7 @@ func (p *Proxy) selectTarget(ctx *fasthttp.RequestCtx) *loadbalance.Target {
|
||||
return nil
|
||||
}
|
||||
|
||||
// For IPHash balancer, extract client IP
|
||||
// 对于 IPHash 负载均衡器，提取客户端 IP
|
||||
if ipHash, ok := balancer.(*loadbalance.IPHash); ok {
|
||||
clientIP := getClientIP(ctx)
|
||||
return ipHash.SelectByIP(targets, clientIP)
|
||||
@ -230,9 +229,9 @@ func (p *Proxy) selectTarget(ctx *fasthttp.RequestCtx) *loadbalance.Target {
|
||||
return balancer.Select(targets)
|
||||
}
|
||||
|
||||
// getClientIP extracts the client IP address from the request context.
|
||||
// getClientIP 从请求上下文中提取客户端 IP 地址。
|
||||
func getClientIP(ctx *fasthttp.RequestCtx) string {
|
||||
// Check X-Forwarded-For header first
|
||||
// 首先检查 X-Forwarded-For 请求头
|
||||
if xff := ctx.Request.Header.Peek("X-Forwarded-For"); len(xff) > 0 {
|
||||
ips := strings.Split(string(xff), ",")
|
||||
if len(ips) > 0 {
|
||||
@ -240,12 +239,12 @@ func getClientIP(ctx *fasthttp.RequestCtx) string {
|
||||
}
|
||||
}
|
||||
|
||||
// Check X-Real-IP header
|
||||
// 检查 X-Real-IP 请求头
|
||||
if xri := ctx.Request.Header.Peek("X-Real-IP"); len(xri) > 0 {
|
||||
return string(xri)
|
||||
}
|
||||
|
||||
// Fall back to RemoteAddr
|
||||
// 回退到 RemoteAddr
|
||||
if addr := ctx.RemoteAddr(); addr != nil {
|
||||
if tcpAddr, ok := addr.(*net.TCPAddr); ok {
|
||||
return tcpAddr.IP.String()
|
||||
@ -256,7 +255,7 @@ func getClientIP(ctx *fasthttp.RequestCtx) string {
|
||||
return ""
|
||||
}
|
||||
|
||||
// getClient returns the HostClient for a given target URL.
|
||||
// getClient 返回给定目标 URL 对应的 HostClient。
|
||||
func (p *Proxy) getClient(targetURL string) *fasthttp.HostClient {
|
||||
p.mu.RLock()
|
||||
client := p.clients[targetURL]
|
||||
@ -264,18 +263,18 @@ func (p *Proxy) getClient(targetURL string) *fasthttp.HostClient {
|
||||
return client
|
||||
}
|
||||
|
||||
// modifyRequestHeaders modifies the request headers before forwarding to backend.
|
||||
// It adds standard proxy headers and applies custom header configurations.
|
||||
// modifyRequestHeaders 在转发到后端之前修改请求头。
|
||||
// 添加标准代理请求头并应用自定义请求头配置。
|
||||
func (p *Proxy) modifyRequestHeaders(ctx *fasthttp.RequestCtx, target *loadbalance.Target) {
|
||||
headers := &ctx.Request.Header
|
||||
|
||||
// Add X-Real-IP header
|
||||
// 添加 X-Real-IP 请求头
|
||||
clientIP := getClientIP(ctx)
|
||||
if clientIP != "" {
|
||||
headers.Set("X-Real-IP", clientIP)
|
||||
}
|
||||
|
||||
// Add/Append X-Forwarded-For header
|
||||
// 添加/追加 X-Forwarded-For 请求头
|
||||
existingXFF := headers.Peek("X-Forwarded-For")
|
||||
if len(existingXFF) > 0 {
|
||||
headers.Set("X-Forwarded-For", string(existingXFF)+", "+clientIP)
|
||||
@ -283,27 +282,27 @@ func (p *Proxy) modifyRequestHeaders(ctx *fasthttp.RequestCtx, target *loadbalan
|
||||
headers.Set("X-Forwarded-For", clientIP)
|
||||
}
|
||||
|
||||
// Add X-Forwarded-Host header
|
||||
// 添加 X-Forwarded-Host 请求头
|
||||
host := string(ctx.Host())
|
||||
if host != "" {
|
||||
headers.Set("X-Forwarded-Host", host)
|
||||
}
|
||||
|
||||
// Add X-Forwarded-Proto header
|
||||
// 添加 X-Forwarded-Proto 请求头
|
||||
proto := "http"
|
||||
if ctx.IsTLS() {
|
||||
proto = "https"
|
||||
}
|
||||
headers.Set("X-Forwarded-Proto", proto)
|
||||
|
||||
// Set custom request headers from config
|
||||
// 从配置设置自定义请求头
|
||||
if p.config.Headers.SetRequest != nil {
|
||||
for key, value := range p.config.Headers.SetRequest {
|
||||
headers.Set(key, value)
|
||||
}
|
||||
}
|
||||
|
||||
// Remove configured headers
|
||||
// 移除配置的请求头
|
||||
if len(p.config.Headers.Remove) > 0 {
|
||||
for _, key := range p.config.Headers.Remove {
|
||||
headers.Del(key)
|
||||
@ -311,9 +310,9 @@ func (p *Proxy) modifyRequestHeaders(ctx *fasthttp.RequestCtx, target *loadbalan
|
||||
}
|
||||
}
|
||||
|
||||
// modifyResponseHeaders modifies the response headers before sending to client.
|
||||
// modifyResponseHeaders 在发送给客户端之前修改响应头。
|
||||
func (p *Proxy) modifyResponseHeaders(ctx *fasthttp.RequestCtx) {
|
||||
// Set custom response headers from config
|
||||
// 从配置设置自定义响应头
|
||||
if p.config.Headers.SetResponse != nil {
|
||||
for key, value := range p.config.Headers.SetResponse {
|
||||
ctx.Response.Header.Set(key, value)
|
||||
@ -321,34 +320,34 @@ func (p *Proxy) modifyResponseHeaders(ctx *fasthttp.RequestCtx) {
|
||||
}
|
||||
}
|
||||
|
||||
// isWebSocketRequest checks if the request is a WebSocket upgrade request.
|
||||
// isWebSocketRequest 检查请求是否为 WebSocket 升级请求。
|
||||
func isWebSocketRequest(ctx *fasthttp.RequestCtx) bool {
|
||||
// Check Connection header
|
||||
// 检查 Connection 请求头
|
||||
connection := ctx.Request.Header.Peek("Connection")
|
||||
if !strings.EqualFold(string(connection), "upgrade") {
|
||||
// Also check for "Upgrade" substring (e.g., "keep-alive, Upgrade")
|
||||
// 也检查 "Upgrade" 子串（例如 "keep-alive, Upgrade"）
|
||||
if !strings.Contains(strings.ToLower(string(connection)), "upgrade") {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
// Check Upgrade header
|
||||
// 检查 Upgrade 请求头
|
||||
upgrade := ctx.Request.Header.Peek("Upgrade")
|
||||
return strings.EqualFold(string(upgrade), "websocket")
|
||||
}
|
||||
|
||||
// handleWebSocket handles WebSocket upgrade requests.
|
||||
// For now, it returns 501 Not Implemented as WebSocket proxying
|
||||
// requires special handling beyond HTTP.
|
||||
// handleWebSocket 处理 WebSocket 升级请求。
|
||||
// 目前返回 501 Not Implemented，因为 WebSocket 代理需要
|
||||
// HTTP 之外的特殊处理。
|
||||
func (p *Proxy) handleWebSocket(ctx *fasthttp.RequestCtx, target *loadbalance.Target, client *fasthttp.HostClient) {
|
||||
// WebSocket proxying requires raw TCP connection handling
|
||||
// which is beyond the scope of basic HTTP proxying
|
||||
// This can be implemented later using a TCP bridge
|
||||
// WebSocket 代理需要原始 TCP 连接处理，
|
||||
// 这超出了基本 HTTP 代理的范围。
|
||||
// 后续可以使用 TCP 桥接实现
|
||||
ctx.Error("WebSocket proxying not implemented", fasthttp.StatusNotImplemented)
|
||||
}
|
||||
|
||||
// UpdateTargets updates the proxy targets and reinitializes clients.
|
||||
// This is useful for dynamic configuration updates.
|
||||
// UpdateTargets 更新代理目标并重新初始化客户端。
|
||||
// 适用于动态配置更新。
|
||||
func (p *Proxy) UpdateTargets(targets []*loadbalance.Target) error {
|
||||
if len(targets) == 0 {
|
||||
return errors.New("no targets provided")
|
||||
@ -357,10 +356,10 @@ func (p *Proxy) UpdateTargets(targets []*loadbalance.Target) error {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
|
||||
// Clear old clients
|
||||
// 清除旧客户端
|
||||
p.clients = make(map[string]*fasthttp.HostClient)
|
||||
|
||||
// Initialize new clients
|
||||
// 初始化新客户端
|
||||
for _, target := range targets {
|
||||
if target.URL == "" {
|
||||
continue
|
||||
@ -374,14 +373,14 @@ func (p *Proxy) UpdateTargets(targets []*loadbalance.Target) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// GetTargets returns the current list of targets.
|
||||
// GetTargets 返回当前的目标列表。
|
||||
func (p *Proxy) GetTargets() []*loadbalance.Target {
|
||||
p.mu.RLock()
|
||||
defer p.mu.RUnlock()
|
||||
return p.targets
|
||||
}
|
||||
|
||||
// GetConfig returns the proxy configuration.
|
||||
// GetConfig 返回代理配置。
|
||||
func (p *Proxy) GetConfig() *config.ProxyConfig {
|
||||
p.mu.RLock()
|
||||
defer p.mu.RUnlock()
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user