feat(docker-compose, config, handler): enhance deployment configuration, add metrics support, and improve proxy handling

- Updated docker-compose.yml to include resource limits and health checks for the service.
- Modified go.mod and go.sum to include the new dependency on golang.org/x/time.
- Enhanced main.go to add new metrics routes for monitoring.
- Updated config.json to include metrics configuration with password and token expiry.
- Refactored internal/config to implement a ConfigManager for dynamic configuration loading.
- Improved internal/handler to utilize a shared HTTP client and added metrics tracking for requests.
This commit is contained in:
wood chen 2024-11-30 21:11:05 +08:00
parent 7ee1b0efb5
commit 8770d79bb8
12 changed files with 758 additions and 67 deletions

View File

@ -33,5 +33,9 @@
"TargetHost": "cdn.jsdelivr.net", "TargetHost": "cdn.jsdelivr.net",
"TargetURL": "https://cdn.jsdelivr.net" "TargetURL": "https://cdn.jsdelivr.net"
} }
] ],
"Metrics": {
"Password": "admin123",
"TokenExpiry": 86400
}
} }

View File

@ -9,3 +9,16 @@ services:
environment: environment:
- TZ=Asia/Shanghai - TZ=Asia/Shanghai
restart: always restart: always
deploy:
resources:
limits:
cpus: '1'
memory: 512M
reservations:
cpus: '0.25'
memory: 128M
healthcheck:
test: ["CMD", "wget", "-q", "--spider", "http://localhost:80/"]
interval: 30s
timeout: 3s
retries: 3

5
go.mod
View File

@ -2,4 +2,7 @@ module proxy-go
go 1.23.1 go 1.23.1
require github.com/andybalholm/brotli v1.1.1 require (
github.com/andybalholm/brotli v1.1.1
golang.org/x/time v0.8.0
)

2
go.sum
View File

@ -2,3 +2,5 @@ github.com/andybalholm/brotli v1.1.1 h1:PR2pgnyFznKEugtsUo0xLdDop5SKXd5Qf5ysW+7X
github.com/andybalholm/brotli v1.1.1/go.mod h1:05ib4cKhjx3OQYUY22hTVd34Bc8upXjOLL2rKwwZBoA= github.com/andybalholm/brotli v1.1.1/go.mod h1:05ib4cKhjx3OQYUY22hTVd34Bc8upXjOLL2rKwwZBoA=
github.com/xyproto/randomstring v1.0.5 h1:YtlWPoRdgMu3NZtP45drfy1GKoojuR7hmRcnhZqKjWU= github.com/xyproto/randomstring v1.0.5 h1:YtlWPoRdgMu3NZtP45drfy1GKoojuR7hmRcnhZqKjWU=
github.com/xyproto/randomstring v1.0.5/go.mod h1:rgmS5DeNXLivK7YprL0pY+lTuhNQW3iGxZ18UQApw/E= github.com/xyproto/randomstring v1.0.5/go.mod h1:rgmS5DeNXLivK7YprL0pY+lTuhNQW3iGxZ18UQApw/E=
golang.org/x/time v0.8.0 h1:9i3RxcPv3PZnitoVGMPDKZSq1xW1gK1Xy3ArNOGZfEg=
golang.org/x/time v0.8.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=

View File

@ -3,8 +3,29 @@ package config
import ( import (
"encoding/json" "encoding/json"
"os" "os"
"sync/atomic"
"time"
) )
type ConfigManager struct {
config atomic.Value
configPath string
}
func NewConfigManager(path string) *ConfigManager {
cm := &ConfigManager{configPath: path}
cm.loadConfig()
go cm.watchConfig()
return cm
}
func (cm *ConfigManager) watchConfig() {
ticker := time.NewTicker(30 * time.Second)
for range ticker.C {
cm.loadConfig()
}
}
func Load(path string) (*Config, error) { func Load(path string) (*Config, error) {
data, err := os.ReadFile(path) data, err := os.ReadFile(path)
if err != nil { if err != nil {
@ -18,3 +39,16 @@ func Load(path string) (*Config, error) {
return &config, nil return &config, nil
} }
func (cm *ConfigManager) loadConfig() error {
config, err := Load(cm.configPath)
if err != nil {
return err
}
cm.config.Store(config)
return nil
}
func (cm *ConfigManager) GetConfig() *Config {
return cm.config.Load().(*Config)
}

View File

@ -9,6 +9,7 @@ type Config struct {
MAP map[string]PathConfig `json:"MAP"` // 改为使用PathConfig MAP map[string]PathConfig `json:"MAP"` // 改为使用PathConfig
Compression CompressionConfig `json:"Compression"` Compression CompressionConfig `json:"Compression"`
FixedPaths []FixedPathConfig `json:"FixedPaths"` FixedPaths []FixedPathConfig `json:"FixedPaths"`
Metrics MetricsConfig `json:"Metrics"`
} }
type PathConfig struct { type PathConfig struct {
@ -33,6 +34,11 @@ type FixedPathConfig struct {
TargetURL string `json:"TargetURL"` TargetURL string `json:"TargetURL"`
} }
type MetricsConfig struct {
Password string `json:"Password"`
TokenExpiry int `json:"TokenExpiry"` // token有效期(秒)
}
// 添加一个辅助方法来处理字符串到 PathConfig 的转换 // 添加一个辅助方法来处理字符串到 PathConfig 的转换
func (c *Config) UnmarshalJSON(data []byte) error { func (c *Config) UnmarshalJSON(data []byte) error {
// 创建一个临时结构来解析原始JSON // 创建一个临时结构来解析原始JSON
@ -40,6 +46,7 @@ func (c *Config) UnmarshalJSON(data []byte) error {
MAP map[string]json.RawMessage `json:"MAP"` MAP map[string]json.RawMessage `json:"MAP"`
Compression CompressionConfig `json:"Compression"` Compression CompressionConfig `json:"Compression"`
FixedPaths []FixedPathConfig `json:"FixedPaths"` FixedPaths []FixedPathConfig `json:"FixedPaths"`
Metrics MetricsConfig `json:"Metrics"`
} }
var temp TempConfig var temp TempConfig
@ -75,6 +82,7 @@ func (c *Config) UnmarshalJSON(data []byte) error {
// 复制其他字段 // 复制其他字段
c.Compression = temp.Compression c.Compression = temp.Compression
c.FixedPaths = temp.FixedPaths c.FixedPaths = temp.FixedPaths
c.Metrics = temp.Metrics
return nil return nil
} }

62
internal/handler/auth.go Normal file
View File

@ -0,0 +1,62 @@
package handler
import (
"crypto/rand"
"encoding/base64"
"sync"
"time"
)
type tokenInfo struct {
createdAt time.Time
expiresIn time.Duration
}
type authManager struct {
tokens sync.Map
}
func newAuthManager() *authManager {
am := &authManager{}
// 启动token清理goroutine
go am.cleanExpiredTokens()
return am
}
func (am *authManager) generateToken() string {
b := make([]byte, 32)
rand.Read(b)
return base64.URLEncoding.EncodeToString(b)
}
func (am *authManager) addToken(token string, expiry time.Duration) {
am.tokens.Store(token, tokenInfo{
createdAt: time.Now(),
expiresIn: expiry,
})
}
func (am *authManager) validateToken(token string) bool {
if info, ok := am.tokens.Load(token); ok {
tokenInfo := info.(tokenInfo)
if time.Since(tokenInfo.createdAt) < tokenInfo.expiresIn {
return true
}
am.tokens.Delete(token)
}
return false
}
func (am *authManager) cleanExpiredTokens() {
ticker := time.NewTicker(time.Hour)
for range ticker.C {
am.tokens.Range(func(key, value interface{}) bool {
token := key.(string)
info := value.(tokenInfo)
if time.Since(info.createdAt) >= info.expiresIn {
am.tokens.Delete(token)
}
return true
})
}
}

406
internal/handler/metrics.go Normal file
View File

@ -0,0 +1,406 @@
package handler
import (
"encoding/json"
"fmt"
"net/http"
"runtime"
"strings"
"sync/atomic"
"time"
)
type Metrics struct {
// 基础指标
Uptime string `json:"uptime"`
ActiveRequests int64 `json:"active_requests"`
TotalRequests int64 `json:"total_requests"`
TotalErrors int64 `json:"total_errors"`
ErrorRate float64 `json:"error_rate"`
// 系统指标
NumGoroutine int `json:"num_goroutine"`
MemoryUsage string `json:"memory_usage"`
// 性能指标
AverageResponseTime string `json:"avg_response_time"`
RequestsPerSecond float64 `json:"requests_per_second"`
// 新增字段
TotalBytes int64 `json:"total_bytes"`
BytesPerSecond float64 `json:"bytes_per_second"`
StatusCodeStats map[string]int64 `json:"status_code_stats"`
LatencyPercentiles map[string]float64 `json:"latency_percentiles"`
TopPaths []PathMetrics `json:"top_paths"`
RecentRequests []RequestLog `json:"recent_requests"`
}
type PathMetrics struct {
Path string `json:"path"`
RequestCount int64 `json:"request_count"`
ErrorCount int64 `json:"error_count"`
AvgLatency string `json:"avg_latency"`
BytesTransferred int64 `json:"bytes_transferred"`
}
// 添加格式化字节的辅助函数
func formatBytes(bytes uint64) string {
const (
MB = 1024 * 1024
KB = 1024
)
switch {
case bytes >= MB:
return fmt.Sprintf("%.2f MB", float64(bytes)/MB)
case bytes >= KB:
return fmt.Sprintf("%.2f KB", float64(bytes)/KB)
default:
return fmt.Sprintf("%d Bytes", bytes)
}
}
func (h *ProxyHandler) MetricsHandler(w http.ResponseWriter, r *http.Request) {
var m runtime.MemStats
runtime.ReadMemStats(&m)
// 获取状态码统计
statusStats := make(map[string]int64)
for i, v := range h.metrics.statusStats {
statusStats[fmt.Sprintf("%dxx", i+1)] = v.Load()
}
// 获取Top 10路径统计
var pathMetrics []PathMetrics
h.metrics.pathStats.Range(func(key, value interface{}) bool {
stats := value.(*PathStats)
pathMetrics = append(pathMetrics, PathMetrics{
Path: key.(string),
RequestCount: stats.requests.Load(),
ErrorCount: stats.errors.Load(),
AvgLatency: formatDuration(time.Duration(stats.latencySum.Load() / stats.requests.Load())),
BytesTransferred: stats.bytes.Load(),
})
return len(pathMetrics) < 10
})
// 获取最近的请求
var recentReqs []RequestLog
h.recentRequests.RLock()
cursor := h.recentRequests.cursor.Load()
for i := 0; i < 10; i++ {
idx := (cursor - int64(i) + 1000) % 1000
if h.recentRequests.items[idx] != nil {
recentReqs = append(recentReqs, *h.recentRequests.items[idx])
}
}
h.recentRequests.RUnlock()
metrics := Metrics{
Uptime: time.Since(h.startTime).String(),
ActiveRequests: atomic.LoadInt64(&h.metrics.activeRequests),
TotalRequests: atomic.LoadInt64(&h.metrics.totalRequests),
TotalErrors: atomic.LoadInt64(&h.metrics.totalErrors),
ErrorRate: float64(h.metrics.totalErrors) / float64(h.metrics.totalRequests),
NumGoroutine: runtime.NumGoroutine(),
MemoryUsage: formatBytes(m.Alloc),
TotalBytes: h.metrics.totalBytes.Load(),
BytesPerSecond: float64(h.metrics.totalBytes.Load()) / time.Since(h.startTime).Seconds(),
StatusCodeStats: statusStats,
TopPaths: pathMetrics,
RecentRequests: recentReqs,
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(metrics)
}
// 添加格式化时间的辅助函数
func formatDuration(d time.Duration) string {
if d < time.Millisecond {
return fmt.Sprintf("%.2f μs", float64(d.Microseconds()))
}
if d < time.Second {
return fmt.Sprintf("%.2f ms", float64(d.Milliseconds()))
}
return fmt.Sprintf("%.2f s", d.Seconds())
}
// 修改模板,添加登录页面
var loginTemplate = `
<!DOCTYPE html>
<html>
<head>
<title>Proxy-Go Metrics Login</title>
<meta charset="UTF-8">
<style>
body {
font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, sans-serif;
display: flex;
justify-content: center;
align-items: center;
height: 100vh;
margin: 0;
background: #f5f5f5;
}
.login-card {
background: white;
padding: 30px;
border-radius: 8px;
box-shadow: 0 2px 4px rgba(0,0,0,0.1);
width: 300px;
}
.login-title {
text-align: center;
margin-bottom: 20px;
color: #333;
}
.input-group {
margin-bottom: 15px;
}
input {
width: 100%;
padding: 8px;
border: 1px solid #ddd;
border-radius: 4px;
box-sizing: border-box;
}
button {
width: 100%;
padding: 10px;
background: #4CAF50;
color: white;
border: none;
border-radius: 4px;
cursor: pointer;
}
button:hover {
background: #45a049;
}
.error {
color: red;
text-align: center;
margin-bottom: 15px;
display: none;
}
</style>
</head>
<body>
<div class="login-card">
<h2 class="login-title">Metrics Login</h2>
<div id="error" class="error">密码错误</div>
<div class="input-group">
<input type="password" id="password" placeholder="请输入密码">
</div>
<button onclick="login()">登录</button>
</div>
<script>
function login() {
const password = document.getElementById('password').value;
fetch('/metrics/auth', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify({password: password})
})
.then(response => {
if (response.ok) {
// 登录成功,保存token并跳转
response.json().then(data => {
localStorage.setItem('metricsToken', data.token);
window.location.href = '/metrics/dashboard';
});
} else {
// 显示错误信息
document.getElementById('error').style.display = 'block';
}
})
.catch(error => {
console.error('Error:', error);
document.getElementById('error').style.display = 'block';
});
}
</script>
</body>
</html>
`
// 修改原有的 metricsTemplate,添加 token 检查
var metricsTemplate = `
<!DOCTYPE html>
<html>
<head>
<title>Proxy-Go Metrics</title>
<meta charset="UTF-8">
<style>
body {
font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, "Helvetica Neue", Arial, sans-serif;
max-width: 1200px;
margin: 0 auto;
padding: 20px;
background: #f5f5f5;
}
.card {
background: white;
border-radius: 8px;
padding: 20px;
margin-bottom: 20px;
box-shadow: 0 2px 4px rgba(0,0,0,0.1);
}
.metric {
display: flex;
justify-content: space-between;
padding: 10px 0;
border-bottom: 1px solid #eee;
}
.metric:last-child {
border-bottom: none;
}
.metric-label {
color: #666;
}
.metric-value {
font-weight: bold;
color: #333;
}
h1 {
color: #333;
margin-bottom: 30px;
}
h2 {
color: #666;
margin: 0 0 15px 0;
}
.refresh {
position: fixed;
top: 20px;
right: 20px;
padding: 10px 20px;
background: #4CAF50;
color: white;
border: none;
border-radius: 4px;
cursor: pointer;
}
.refresh:hover {
background: #45a049;
}
#lastUpdate {
position: fixed;
top: 20px;
right: 140px;
color: #666;
}
</style>
</head>
<body>
<script>
// 检查登录状态
const token = localStorage.getItem('metricsToken');
if (!token) {
window.location.href = '/metrics/ui';
}
function refreshMetrics() {
fetch('/metrics', {
headers: {
'Authorization': 'Bearer ' + token
}
})
.then(response => {
if (response.status === 401) {
// token 无效,跳转到登录页
localStorage.removeItem('metricsToken');
window.location.href = '/metrics/ui';
return;
}
return response.json();
})
.then(data => {
if (data) updateMetrics(data);
})
.catch(error => console.error('Error:', error));
}
function updateMetrics(data) {
document.getElementById('uptime').textContent = data.uptime;
document.getElementById('activeRequests').textContent = data.active_requests;
document.getElementById('totalRequests').textContent = data.total_requests;
document.getElementById('totalErrors').textContent = data.total_errors;
document.getElementById('errorRate').textContent = (data.error_rate * 100).toFixed(2) + '%';
document.getElementById('numGoroutine').textContent = data.num_goroutine;
document.getElementById('memoryUsage').textContent = data.memory_usage;
document.getElementById('avgResponseTime').textContent = data.avg_response_time;
document.getElementById('requestsPerSecond').textContent = data.requests_per_second.toFixed(2);
document.getElementById('lastUpdate').textContent = '最后更新: ' + new Date().toLocaleTimeString();
}
// 初始加载
refreshMetrics();
// 每5秒自动刷新
setInterval(refreshMetrics, 5000);
</script>
</body>
</html>
`
// 添加认证中间件
func (h *ProxyHandler) AuthMiddleware(next http.HandlerFunc) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
auth := r.Header.Get("Authorization")
if auth == "" || !strings.HasPrefix(auth, "Bearer ") {
http.Error(w, "Unauthorized", http.StatusUnauthorized)
return
}
token := strings.TrimPrefix(auth, "Bearer ")
if !h.auth.validateToken(token) {
http.Error(w, "Invalid token", http.StatusUnauthorized)
return
}
next(w, r)
}
}
// 修改处理器
func (h *ProxyHandler) MetricsPageHandler(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "text/html; charset=utf-8")
w.Write([]byte(loginTemplate))
}
func (h *ProxyHandler) MetricsDashboardHandler(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "text/html; charset=utf-8")
w.Write([]byte(metricsTemplate))
}
func (h *ProxyHandler) MetricsAuthHandler(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
return
}
var req struct {
Password string `json:"password"`
}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
http.Error(w, "Invalid request", http.StatusBadRequest)
return
}
if req.Password != h.config.Metrics.Password {
http.Error(w, "Invalid password", http.StatusUnauthorized)
return
}
token := h.auth.generateToken()
h.auth.addToken(token, time.Duration(h.config.Metrics.TokenExpiry)*time.Second)
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]string{
"token": token,
})
}

View File

@ -11,10 +11,23 @@ import (
"time" "time"
) )
type MirrorProxyHandler struct{} type MirrorProxyHandler struct {
client *http.Client
}
func NewMirrorProxyHandler() *MirrorProxyHandler { func NewMirrorProxyHandler() *MirrorProxyHandler {
return &MirrorProxyHandler{} transport := &http.Transport{
MaxIdleConns: 100,
MaxIdleConnsPerHost: 10,
IdleConnTimeout: 90 * time.Second,
}
return &MirrorProxyHandler{
client: &http.Client{
Transport: transport,
Timeout: 30 * time.Second,
},
}
} }
func (h *MirrorProxyHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { func (h *MirrorProxyHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
@ -91,14 +104,7 @@ func (h *MirrorProxyHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
proxyReq.Host = parsedURL.Host proxyReq.Host = parsedURL.Host
// 发送请求 // 发送请求
client := &http.Client{ resp, err := h.client.Do(proxyReq)
Transport: &http.Transport{
DisableCompression: true,
// 可以添加其他传输设置如TLS配置等
},
Timeout: 30 * time.Second,
}
resp, err := client.Do(proxyReq)
if err != nil { if err != nil {
http.Error(w, "Error forwarding request", http.StatusBadGateway) http.Error(w, "Error forwarding request", http.StatusBadGateway)
log.Printf("| %-6s | %3d | %12s | %15s | %10s | %-30s | Error forwarding request: %v", log.Printf("| %-6s | %3d | %12s | %15s | %10s | %-30s | Error forwarding request: %v",

View File

@ -10,33 +10,103 @@ import (
"proxy-go/internal/config" "proxy-go/internal/config"
"proxy-go/internal/utils" "proxy-go/internal/utils"
"strings" "strings"
"sync"
"sync/atomic"
"time" "time"
"golang.org/x/time/rate"
) )
const ( const (
defaultBufferSize = 32 * 1024 // 32KB defaultBufferSize = 32 * 1024 // 32KB
) )
var bufferPool = sync.Pool{
New: func() interface{} {
return make([]byte, defaultBufferSize)
},
}
type ProxyHandler struct { type ProxyHandler struct {
pathMap map[string]config.PathConfig pathMap map[string]config.PathConfig
client *http.Client
limiter *rate.Limiter
startTime time.Time
config *config.Config
auth *authManager
metrics struct {
activeRequests int64
totalRequests int64
totalErrors int64
totalBytes atomic.Int64 // 总传输字节数
pathStats sync.Map // 路径统计 map[string]*PathStats
statusStats [6]atomic.Int64 // HTTP状态码统计(1xx-5xx)
latencyBuckets [10]atomic.Int64 // 延迟分布(0-100ms, 100-200ms...)
}
recentRequests struct {
sync.RWMutex
items [1000]*RequestLog // 固定大小的环形缓冲区
cursor atomic.Int64 // 当前位置
}
}
// 单个请求的统计信息
type RequestLog struct {
Time time.Time
Path string
Status int
Latency time.Duration
BytesSent int64
ClientIP string
}
// 路径统计
type PathStats struct {
requests atomic.Int64
errors atomic.Int64
bytes atomic.Int64
latencySum atomic.Int64
} }
// 修改参数类型 // 修改参数类型
func NewProxyHandler(pathMap map[string]config.PathConfig) *ProxyHandler { func NewProxyHandler(cfg *config.Config) *ProxyHandler {
transport := &http.Transport{
MaxIdleConns: 100, // 最大空闲连接数
MaxIdleConnsPerHost: 10, // 每个 host 的最大空闲连接数
IdleConnTimeout: 90 * time.Second, // 空闲连接超时时间
}
return &ProxyHandler{ return &ProxyHandler{
pathMap: pathMap, pathMap: cfg.MAP,
client: &http.Client{
Transport: transport,
Timeout: 30 * time.Second,
},
limiter: rate.NewLimiter(rate.Limit(5000), 10000),
startTime: time.Now(),
config: cfg,
auth: newAuthManager(),
} }
} }
func (h *ProxyHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { func (h *ProxyHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
startTime := time.Now() atomic.AddInt64(&h.metrics.activeRequests, 1)
atomic.AddInt64(&h.metrics.totalRequests, 1)
defer atomic.AddInt64(&h.metrics.activeRequests, -1)
if !h.limiter.Allow() {
http.Error(w, "Too Many Requests", http.StatusTooManyRequests)
return
}
start := time.Now()
// 处理根路径请求 // 处理根路径请求
if r.URL.Path == "/" { if r.URL.Path == "/" {
w.WriteHeader(http.StatusOK) w.WriteHeader(http.StatusOK)
fmt.Fprint(w, "Welcome to CZL proxy.") fmt.Fprint(w, "Welcome to CZL proxy.")
log.Printf("[%s] %s %s -> %d (root path) [%v]", log.Printf("[%s] %s %s -> %d (root path) [%v]",
utils.GetClientIP(r), r.Method, r.URL.Path, http.StatusOK, time.Since(startTime)) utils.GetClientIP(r), r.Method, r.URL.Path, http.StatusOK, time.Since(start))
return return
} }
@ -55,7 +125,7 @@ func (h *ProxyHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
if matchedPrefix == "" { if matchedPrefix == "" {
http.NotFound(w, r) http.NotFound(w, r)
log.Printf("[%s] %s %s -> 404 (not found) [%v]", log.Printf("[%s] %s %s -> 404 (not found) [%v]",
utils.GetClientIP(r), r.Method, r.URL.Path, time.Since(startTime)) utils.GetClientIP(r), r.Method, r.URL.Path, time.Since(start))
return return
} }
@ -67,7 +137,7 @@ func (h *ProxyHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
if err != nil { if err != nil {
http.Error(w, "Error decoding path", http.StatusInternalServerError) http.Error(w, "Error decoding path", http.StatusInternalServerError)
log.Printf("[%s] %s %s -> 500 (error decoding path: %v) [%v]", log.Printf("[%s] %s %s -> 500 (error decoding path: %v) [%v]",
utils.GetClientIP(r), r.Method, r.URL.Path, err, time.Since(startTime)) utils.GetClientIP(r), r.Method, r.URL.Path, err, time.Since(start))
return return
} }
@ -96,7 +166,7 @@ func (h *ProxyHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
if err != nil { if err != nil {
http.Error(w, "Error parsing target URL", http.StatusInternalServerError) http.Error(w, "Error parsing target URL", http.StatusInternalServerError)
log.Printf("[%s] %s %s -> 500 (error parsing URL: %v) [%v]", log.Printf("[%s] %s %s -> 500 (error parsing URL: %v) [%v]",
utils.GetClientIP(r), r.Method, r.URL.Path, err, time.Since(startTime)) utils.GetClientIP(r), r.Method, r.URL.Path, err, time.Since(start))
return return
} }
@ -162,13 +232,12 @@ func (h *ProxyHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
} }
// 发送代理请求 // 发送代理请求
client := &http.Client{} resp, err := h.client.Do(proxyReq)
resp, err := client.Do(proxyReq)
if err != nil { if err != nil {
http.Error(w, "Error forwarding request", http.StatusBadGateway) http.Error(w, "Error forwarding request", http.StatusBadGateway)
log.Printf("[%s] %s %s -> 502 (proxy error: %v) [%v]", log.Printf("[%s] %s %s -> 502 (proxy error: %v) [%v]",
utils.GetClientIP(r), r.Method, r.URL.Path, err, time.Since(startTime)) utils.GetClientIP(r), r.Method, r.URL.Path, err, time.Since(start))
return return
} }
defer resp.Body.Close() defer resp.Body.Close()
@ -181,49 +250,114 @@ func (h *ProxyHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
// 设置响应状态码 // 设置响应状态码
w.WriteHeader(resp.StatusCode) w.WriteHeader(resp.StatusCode)
// 使用流式传输复制响应体 // 根据响应大小选择不同的处理策略
var bytesCopied int64 contentLength := resp.ContentLength
if f, ok := w.(http.Flusher); ok { if contentLength > 0 && contentLength < 1<<20 { // 1MB 以下的小响应
buf := make([]byte, defaultBufferSize) // 直接读取到内存并一次性写入
for { body, err := io.ReadAll(resp.Body)
n, rerr := resp.Body.Read(buf)
if n > 0 {
bytesCopied += int64(n)
_, werr := w.Write(buf[:n])
if werr != nil {
log.Printf("Error writing response: %v", werr)
return
}
f.Flush()
}
if rerr == io.EOF {
break
}
if rerr != nil {
log.Printf("Error reading response: %v", rerr)
break
}
}
} else {
// 如果不支持 Flusher使用普通的 io.Copy
bytesCopied, err = io.Copy(w, resp.Body)
if err != nil { if err != nil {
log.Printf("Error copying response: %v", err) http.Error(w, "Error reading response", http.StatusInternalServerError)
return
} }
written, _ := w.Write(body)
h.recordStats(r.URL.Path, resp.StatusCode, time.Since(start), int64(written), r)
} else {
// 大响应使用流式传输
var bytesCopied int64
if f, ok := w.(http.Flusher); ok {
buf := bufferPool.Get().([]byte)
defer bufferPool.Put(buf)
for {
n, rerr := resp.Body.Read(buf)
if n > 0 {
bytesCopied += int64(n)
_, werr := w.Write(buf[:n])
if werr != nil {
log.Printf("Error writing response: %v", werr)
return
}
f.Flush()
}
if rerr == io.EOF {
break
}
if rerr != nil {
log.Printf("Error reading response: %v", rerr)
break
}
}
} else {
// 如果不支持 Flusher使用普通的 io.Copy
bytesCopied, err = io.Copy(w, resp.Body)
if err != nil {
log.Printf("Error copying response: %v", err)
}
}
// 记录访问日志
log.Printf("| %-6s | %3d | %12s | %15s | %10s | %-30s | %-50s -> %s",
r.Method, // HTTP方法左对齐占6位
resp.StatusCode, // 状态码占3位
time.Since(start), // 处理时间占12位
utils.GetClientIP(r), // IP地址占15位
utils.FormatBytes(bytesCopied), // 传输大小占10位
utils.GetRequestSource(r), // 请求来源
r.URL.Path, // 请求路径左对齐占50位
targetURL, // 目标URL
)
h.recordStats(r.URL.Path, resp.StatusCode, time.Since(start), bytesCopied, r)
} }
// 记录访问日志 if err != nil {
log.Printf("| %-6s | %3d | %12s | %15s | %10s | %-30s | %-50s -> %s", atomic.AddInt64(&h.metrics.totalErrors, 1)
r.Method, // HTTP方法左对齐占6位 }
resp.StatusCode, // 状态码占3位 }
time.Since(startTime), // 处理时间占12位
utils.GetClientIP(r), // IP地址占15位
utils.FormatBytes(bytesCopied), // 传输大小占10位
utils.GetRequestSource(r), // 请求来源
r.URL.Path, // 请求路径左对齐占50位
targetURL, // 目标URL
)
func (h *ProxyHandler) recordStats(path string, status int, latency time.Duration, bytes int64, r *http.Request) {
// 更新总字节数
h.metrics.totalBytes.Add(bytes)
// 更新状态码统计
if status >= 100 && status < 600 {
h.metrics.statusStats[status/100-1].Add(1)
}
// 更新延迟分布
bucket := int(latency.Milliseconds() / 100)
if bucket < 10 {
h.metrics.latencyBuckets[bucket].Add(1)
}
// 更新路径统计
if stats, ok := h.metrics.pathStats.Load(path); ok {
pathStats := stats.(*PathStats)
pathStats.requests.Add(1)
pathStats.bytes.Add(bytes)
pathStats.latencySum.Add(int64(latency))
} else {
// 首次遇到该路径
newStats := &PathStats{}
newStats.requests.Add(1)
newStats.bytes.Add(bytes)
newStats.latencySum.Add(int64(latency))
h.metrics.pathStats.Store(path, newStats)
}
// 记录最近的请求
log := &RequestLog{
Time: time.Now(),
Path: path,
Status: status,
Latency: latency,
BytesSent: bytes,
ClientIP: utils.GetClientIP(r),
}
cursor := h.recentRequests.cursor.Add(1) % 1000
h.recentRequests.Lock()
h.recentRequests.items[cursor] = log
h.recentRequests.Unlock()
} }
func copyHeader(dst, src http.Header) { func copyHeader(dst, src http.Header) {

View File

@ -8,6 +8,7 @@ import (
"net/http" "net/http"
"proxy-go/internal/compression" "proxy-go/internal/compression"
"strings" "strings"
"sync"
) )
const ( const (
@ -24,6 +25,12 @@ type CompressResponseWriter struct {
compressed bool compressed bool
} }
var writerPool = sync.Pool{
New: func() interface{} {
return bufio.NewWriterSize(nil, defaultBufferSize)
},
}
func CompressionMiddleware(manager compression.Manager) func(http.Handler) http.Handler { func CompressionMiddleware(manager compression.Manager) func(http.Handler) http.Handler {
return func(next http.Handler) http.Handler { return func(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@ -106,15 +113,21 @@ func (cw *CompressResponseWriter) Write(b []byte) (int, error) {
// 延迟初始化压缩写入器 // 延迟初始化压缩写入器
if cw.writer == nil { if cw.writer == nil {
var err error writer, err := cw.compressor.Compress(cw.ResponseWriter)
cw.writer, err = cw.compressor.Compress(cw.ResponseWriter)
if err != nil { if err != nil {
return 0, err return 0, err
} }
cw.bufferedWriter = bufio.NewWriterSize(cw.writer, defaultBufferSize) cw.writer = writer
bw := writerPool.Get().(*bufio.Writer)
bw.Reset(writer)
cw.bufferedWriter = bw
} }
return cw.bufferedWriter.Write(b) n, err := cw.bufferedWriter.Write(b)
if err != nil {
writerPool.Put(cw.bufferedWriter)
}
return n, err
} }
// 实现 http.Hijacker 接口 // 实现 http.Hijacker 接口

View File

@ -28,7 +28,7 @@ func main() {
// 创建代理处理器 // 创建代理处理器
mirrorHandler := handler.NewMirrorProxyHandler() mirrorHandler := handler.NewMirrorProxyHandler()
proxyHandler := handler.NewProxyHandler(cfg.MAP) proxyHandler := handler.NewProxyHandler(cfg)
// 创建处理器链 // 创建处理器链
handlers := []struct { handlers := []struct {
@ -80,6 +80,12 @@ func main() {
handler = middleware.CompressionMiddleware(compManager)(handler) handler = middleware.CompressionMiddleware(compManager)(handler)
} }
// 添加监控路由
http.HandleFunc("/metrics", proxyHandler.AuthMiddleware(proxyHandler.MetricsHandler))
http.HandleFunc("/metrics/ui", proxyHandler.MetricsPageHandler)
http.HandleFunc("/metrics/auth", proxyHandler.MetricsAuthHandler)
http.HandleFunc("/metrics/dashboard", proxyHandler.MetricsDashboardHandler)
// 创建服务器 // 创建服务器
server := &http.Server{ server := &http.Server{
Addr: ":80", Addr: ":80",