Compare commits

...

5 Commits

4 changed files with 263 additions and 337 deletions

View File

@ -10,7 +10,6 @@ import (
"proxy-go/internal/utils"
"runtime"
"sort"
"strings"
"sync"
"sync/atomic"
"time"
@ -38,6 +37,17 @@ type Collector struct {
config *config.Config
}
type RequestMetric struct {
Path string
Status int
Latency time.Duration
Bytes int64
ClientIP string
Request *http.Request
}
var requestChan chan RequestMetric
var (
instance *Collector
once sync.Once
@ -66,6 +76,10 @@ func InitCollector(cfg *config.Config) error {
instance.latencyBuckets.Store(bucket, counter)
}
// 初始化异步指标收集通道
requestChan = make(chan RequestMetric, 10000)
instance.startAsyncMetricsUpdater()
// 启动数据一致性检查器
instance.startConsistencyChecker()
@ -90,107 +104,22 @@ func (c *Collector) EndRequest() {
atomic.AddInt64(&c.activeRequests, -1)
}
// RecordRequest 记录请求
// RecordRequest 记录请求异步写入channel
func (c *Collector) RecordRequest(path string, status int, latency time.Duration, bytes int64, clientIP string, r *http.Request) {
// 更新状态码统计
statusKey := fmt.Sprintf("%d", status)
if counter, ok := c.statusCodeStats.Load(statusKey); ok {
atomic.AddInt64(counter.(*int64), 1)
} else {
counter := new(int64)
*counter = 1
c.statusCodeStats.Store(statusKey, counter)
metric := RequestMetric{
Path: path,
Status: status,
Latency: latency,
Bytes: bytes,
ClientIP: clientIP,
Request: r,
}
// 更新总字节数和带宽统计
atomic.AddInt64(&c.totalBytes, bytes)
c.updateBandwidthStats(bytes)
// 更新延迟统计
atomic.AddInt64(&c.latencySum, int64(latency))
latencyNanos := int64(latency)
for {
oldMin := atomic.LoadInt64(&c.minLatency)
if oldMin <= latencyNanos {
break
}
if atomic.CompareAndSwapInt64(&c.minLatency, oldMin, latencyNanos) {
break
}
}
for {
oldMax := atomic.LoadInt64(&c.maxLatency)
if oldMax >= latencyNanos {
break
}
if atomic.CompareAndSwapInt64(&c.maxLatency, oldMax, latencyNanos) {
break
}
}
// 更新延迟分布
latencyMs := latency.Milliseconds()
var bucketKey string
switch {
case latencyMs < 10:
bucketKey = "lt10ms"
case latencyMs < 50:
bucketKey = "10-50ms"
case latencyMs < 200:
bucketKey = "50-200ms"
case latencyMs < 1000:
bucketKey = "200-1000ms"
select {
case requestChan <- metric:
// ok
default:
bucketKey = "gt1s"
// channel 满了,丢弃或降级处理
}
if counter, ok := c.latencyBuckets.Load(bucketKey); ok {
atomic.AddInt64(counter.(*int64), 1)
} else {
counter := new(int64)
*counter = 1
c.latencyBuckets.Store(bucketKey, counter)
}
// 更新引用来源统计
if r != nil {
referer := r.Header.Get("Referer")
if referer != "" {
// 简化引用来源,只保留域名部分
referer = simplifyReferer(referer)
if value, ok := c.refererStats.Load(referer); ok {
stat := value.(*models.PathMetrics)
stat.AddRequest()
if status >= 400 {
stat.AddError()
}
stat.AddLatency(int64(latency))
stat.AddBytes(bytes)
} else {
newStat := &models.PathMetrics{
Path: referer,
}
newStat.RequestCount.Store(1)
if status >= 400 {
newStat.ErrorCount.Store(1)
}
newStat.TotalLatency.Store(int64(latency))
newStat.BytesTransferred.Store(bytes)
c.refererStats.Store(referer, newStat)
}
}
}
// 更新最近请求记录
c.recentRequests.Push(models.RequestLog{
Time: time.Now(),
Path: path,
Status: status,
Latency: int64(latency),
BytesSent: bytes,
ClientIP: clientIP,
})
}
// FormatUptime 格式化运行时间
@ -478,126 +407,158 @@ func (c *Collector) getBandwidthHistory() map[string]string {
return history
}
// simplifyReferer 简化引用来源URL只保留域名部分
func simplifyReferer(referer string) string {
// 移除协议部分
if idx := strings.Index(referer, "://"); idx != -1 {
referer = referer[idx+3:]
}
// 只保留域名部分
if idx := strings.Index(referer, "/"); idx != -1 {
referer = referer[:idx]
}
return referer
}
// startCleanupTask 启动定期清理任务
func (c *Collector) startCleanupTask() {
go func() {
// 先立即执行一次清理
c.cleanupOldData()
ticker := time.NewTicker(15 * time.Minute) // 每15分钟清理一次
ticker := time.NewTicker(1 * time.Hour)
defer ticker.Stop()
for range ticker.C {
c.cleanupOldData()
for {
<-ticker.C
oneDayAgo := time.Now().Add(-24 * time.Hour).Unix()
// 清理超过24小时的引用来源统计
var keysToDelete []interface{}
c.refererStats.Range(func(key, value interface{}) bool {
metrics := value.(*models.PathMetrics)
if metrics.LastAccessTime.Load() < oneDayAgo {
keysToDelete = append(keysToDelete, key)
}
return true
})
for _, key := range keysToDelete {
c.refererStats.Delete(key)
}
if len(keysToDelete) > 0 {
log.Printf("[Collector] 已清理 %d 条过期的引用来源统计", len(keysToDelete))
}
// 强制GC
runtime.GC()
}
}()
}
// cleanupOldData 清理旧数据
func (c *Collector) cleanupOldData() {
log.Printf("[Metrics] 开始清理旧数据...")
// 清理引用来源统计 - 类似地处理
var referersToRemove []string
var referersCount int
var totalRefererRequests int64
// 先收集所有引用来源及其请求数
type refererInfo struct {
referer string
count int64
}
var referers []refererInfo
c.refererStats.Range(func(key, value interface{}) bool {
referer := key.(string)
stats := value.(*models.PathMetrics)
count := stats.GetRequestCount()
referersCount++
totalRefererRequests += count
referers = append(referers, refererInfo{referer, count})
return true
})
// 按请求数排序
sort.Slice(referers, func(i, j int) bool {
return referers[i].count > referers[j].count
})
// 只保留前50个请求数最多的引用来源或者请求数占总请求数2%以上的引用来源
refThreshold := totalRefererRequests / 50 // 2%的阈值
if refThreshold < 5 {
refThreshold = 5 // 至少保留请求数>=5的引用来源
}
// 标记要删除的引用来源
for _, ri := range referers {
if len(referers)-len(referersToRemove) <= 50 {
// 已经只剩下50个引用来源了不再删除
break
}
if ri.count < refThreshold {
referersToRemove = append(referersToRemove, ri.referer)
}
}
// 删除标记的引用来源
for _, referer := range referersToRemove {
c.refererStats.Delete(referer)
}
// 清理带宽历史 - 只保留最近的记录
c.bandwidthStats.Lock()
if len(c.bandwidthStats.history) > 10 {
// 找出最旧的记录并删除
var oldestKeys []string
var oldestTimes []time.Time
for k := range c.bandwidthStats.history {
t, err := time.Parse("01-02 15:04", k)
if err != nil {
continue
// 异步批量处理请求指标
func (c *Collector) startAsyncMetricsUpdater() {
go func() {
batch := make([]RequestMetric, 0, 1000)
ticker := time.NewTicker(time.Second)
defer ticker.Stop()
for {
select {
case metric := <-requestChan:
batch = append(batch, metric)
if len(batch) >= 1000 {
c.updateMetricsBatch(batch)
batch = batch[:0]
}
case <-ticker.C:
if len(batch) > 0 {
c.updateMetricsBatch(batch)
batch = batch[:0]
}
}
oldestTimes = append(oldestTimes, t)
oldestKeys = append(oldestKeys, k)
}
// 按时间排序
sort.Slice(oldestKeys, func(i, j int) bool {
return oldestTimes[i].Before(oldestTimes[j])
})
// 删除最旧的记录只保留最近10条
for i := 0; i < len(oldestKeys)-10; i++ {
delete(c.bandwidthStats.history, oldestKeys[i])
}
}
c.bandwidthStats.Unlock()
// 强制进行一次GC
runtime.GC()
// 打印内存使用情况
var mem runtime.MemStats
runtime.ReadMemStats(&mem)
log.Printf("[Metrics] 清理完成: 删除了 %d/%d 个引用来源, 当前内存使用: %s",
len(referersToRemove), referersCount,
utils.FormatBytes(int64(mem.Alloc)))
}()
}
// 批量更新指标
func (c *Collector) updateMetricsBatch(batch []RequestMetric) {
for _, m := range batch {
// 更新状态码统计
statusKey := fmt.Sprintf("%d", m.Status)
if counter, ok := c.statusCodeStats.Load(statusKey); ok {
atomic.AddInt64(counter.(*int64), 1)
} else {
counter := new(int64)
*counter = 1
c.statusCodeStats.Store(statusKey, counter)
}
// 更新总字节数和带宽统计
atomic.AddInt64(&c.totalBytes, m.Bytes)
c.updateBandwidthStats(m.Bytes)
// 更新延迟统计
atomic.AddInt64(&c.latencySum, int64(m.Latency))
latencyNanos := int64(m.Latency)
for {
oldMin := atomic.LoadInt64(&c.minLatency)
if oldMin <= latencyNanos {
break
}
if atomic.CompareAndSwapInt64(&c.minLatency, oldMin, latencyNanos) {
break
}
}
for {
oldMax := atomic.LoadInt64(&c.maxLatency)
if oldMax >= latencyNanos {
break
}
if atomic.CompareAndSwapInt64(&c.maxLatency, oldMax, latencyNanos) {
break
}
}
// 更新延迟分布
latencyMs := m.Latency.Milliseconds()
var bucketKey string
switch {
case latencyMs < 10:
bucketKey = "lt10ms"
case latencyMs < 50:
bucketKey = "10-50ms"
case latencyMs < 200:
bucketKey = "50-200ms"
case latencyMs < 1000:
bucketKey = "200-1000ms"
default:
bucketKey = "gt1s"
}
if counter, ok := c.latencyBuckets.Load(bucketKey); ok {
atomic.AddInt64(counter.(*int64), 1)
} else {
counter := new(int64)
*counter = 1
c.latencyBuckets.Store(bucketKey, counter)
}
// 记录引用来源
if m.Request != nil {
referer := m.Request.Referer()
if referer != "" {
var refererMetrics *models.PathMetrics
if existingMetrics, ok := c.refererStats.Load(referer); ok {
refererMetrics = existingMetrics.(*models.PathMetrics)
} else {
refererMetrics = &models.PathMetrics{Path: referer}
c.refererStats.Store(referer, refererMetrics)
}
refererMetrics.AddRequest()
if m.Status >= 400 {
refererMetrics.AddError()
}
refererMetrics.AddBytes(m.Bytes)
refererMetrics.AddLatency(m.Latency.Nanoseconds())
// 更新最后访问时间
refererMetrics.LastAccessTime.Store(time.Now().Unix())
}
}
// 更新最近请求记录
c.recentRequests.Push(models.RequestLog{
Time: time.Now(),
Path: m.Path,
Status: m.Status,
Latency: int64(m.Latency),
BytesSent: m.Bytes,
ClientIP: m.ClientIP,
})
}
}

View File

@ -6,7 +6,6 @@ import (
"log"
"os"
"path/filepath"
"proxy-go/internal/models"
"proxy-go/internal/utils"
"runtime"
"sync"
@ -16,16 +15,14 @@ import (
// MetricsStorage 指标存储结构
type MetricsStorage struct {
collector *Collector
saveInterval time.Duration
dataDir string
stopChan chan struct{}
wg sync.WaitGroup
lastSaveTime time.Time
mutex sync.RWMutex
metricsFile string
statusCodeFile string
refererStatsFile string
collector *Collector
saveInterval time.Duration
dataDir string
stopChan chan struct{}
wg sync.WaitGroup
lastSaveTime time.Time
mutex sync.RWMutex
statusCodeFile string
}
// NewMetricsStorage 创建新的指标存储
@ -35,13 +32,11 @@ func NewMetricsStorage(collector *Collector, dataDir string, saveInterval time.D
}
return &MetricsStorage{
collector: collector,
saveInterval: saveInterval,
dataDir: dataDir,
stopChan: make(chan struct{}),
metricsFile: filepath.Join(dataDir, "metrics.json"),
statusCodeFile: filepath.Join(dataDir, "status_codes.json"),
refererStatsFile: filepath.Join(dataDir, "referer_stats.json"),
collector: collector,
saveInterval: saveInterval,
dataDir: dataDir,
stopChan: make(chan struct{}),
statusCodeFile: filepath.Join(dataDir, "status_codes.json"),
}
}
@ -104,34 +99,12 @@ func (ms *MetricsStorage) SaveMetrics() error {
// 获取当前指标数据
stats := ms.collector.GetStats()
// 保存基本指标 - 只保存必要的字段
basicMetrics := map[string]interface{}{
"uptime": stats["uptime"],
"total_bytes": stats["total_bytes"],
"avg_response_time": stats["avg_response_time"],
"save_time": time.Now().Format(time.RFC3339),
}
// 单独保存延迟统计,避免嵌套结构导致的内存占用
if latencyStats, ok := stats["latency_stats"].(map[string]interface{}); ok {
basicMetrics["latency_min"] = latencyStats["min"]
basicMetrics["latency_max"] = latencyStats["max"]
}
if err := saveJSONToFile(ms.metricsFile, basicMetrics); err != nil {
return fmt.Errorf("保存基本指标失败: %v", err)
}
// 保存状态码统计
if err := saveJSONToFile(ms.statusCodeFile, stats["status_code_stats"]); err != nil {
return fmt.Errorf("保存状态码统计失败: %v", err)
}
// 保存引用来源统计 - 限制数量
topReferers := stats["top_referers"]
if err := saveJSONToFile(ms.refererStatsFile, topReferers); err != nil {
return fmt.Errorf("保存引用来源统计失败: %v", err)
}
// 不再保存引用来源统计,因为它现在只保存在内存中
// 单独保存延迟分布
if latencyStats, ok := stats["latency_stats"].(map[string]interface{}); ok {
@ -142,10 +115,6 @@ func (ms *MetricsStorage) SaveMetrics() error {
}
}
ms.mutex.Lock()
ms.lastSaveTime = time.Now()
ms.mutex.Unlock()
// 强制进行一次GC
runtime.GC()
@ -163,24 +132,9 @@ func (ms *MetricsStorage) LoadMetrics() error {
start := time.Now()
log.Printf("[MetricsStorage] 开始加载指标数据...")
// 检查文件是否存在
if !fileExists(ms.metricsFile) {
return fmt.Errorf("指标数据文件不存在")
}
// 不再加载 basicMetricsmetrics.json
// 加载基本指标
var basicMetrics map[string]interface{}
if err := loadJSONFromFile(ms.metricsFile, &basicMetrics); err != nil {
return fmt.Errorf("加载基本指标失败: %v", err)
}
// 将加载的数据应用到收集器
// 1. 应用总字节数
if totalBytes, ok := basicMetrics["total_bytes"].(float64); ok {
atomic.StoreInt64(&ms.collector.totalBytes, int64(totalBytes))
}
// 2. 加载状态码统计(如果文件存在)
// 1. 加载状态码统计(如果文件存在)
if fileExists(ms.statusCodeFile) {
var statusCodeStats map[string]interface{}
if err := loadJSONFromFile(ms.statusCodeFile, &statusCodeStats); err != nil {
@ -205,47 +159,9 @@ func (ms *MetricsStorage) LoadMetrics() error {
}
}
// 3. 加载引用来源统计(如果文件存在)
if fileExists(ms.refererStatsFile) {
var refererStats []map[string]interface{}
if err := loadJSONFromFile(ms.refererStatsFile, &refererStats); err != nil {
log.Printf("[MetricsStorage] 加载引用来源统计失败: %v", err)
} else {
// 只加载前20个引用来源统计
maxReferers := 20
if len(refererStats) > maxReferers {
refererStats = refererStats[:maxReferers]
}
// 不再加载引用来源统计,因为它现在只保存在内存中
for _, refererStat := range refererStats {
referer, ok := refererStat["path"].(string)
if !ok {
continue
}
requestCount, _ := refererStat["request_count"].(float64)
errorCount, _ := refererStat["error_count"].(float64)
bytesTransferred, _ := refererStat["bytes_transferred"].(float64)
// 创建或更新引用来源统计
var refererMetrics *models.PathMetrics
if existingMetrics, ok := ms.collector.refererStats.Load(referer); ok {
refererMetrics = existingMetrics.(*models.PathMetrics)
} else {
refererMetrics = &models.PathMetrics{Path: referer}
ms.collector.refererStats.Store(referer, refererMetrics)
}
// 设置统计值
refererMetrics.RequestCount.Store(int64(requestCount))
refererMetrics.ErrorCount.Store(int64(errorCount))
refererMetrics.BytesTransferred.Store(int64(bytesTransferred))
}
log.Printf("[MetricsStorage] 加载了 %d 条引用来源统计", len(refererStats))
}
}
// 5. 加载延迟分布(如果文件存在)
// 3. 加载延迟分布(如果文件存在)
latencyDistributionFile := filepath.Join(ms.dataDir, "latency_distribution.json")
if fileExists(latencyDistributionFile) {
var distribution map[string]interface{}
@ -265,15 +181,6 @@ func (ms *MetricsStorage) LoadMetrics() error {
log.Printf("[MetricsStorage] 加载了延迟分布数据")
}
}
ms.mutex.Lock()
if saveTime, ok := basicMetrics["save_time"].(string); ok {
if t, err := time.Parse(time.RFC3339, saveTime); err == nil {
ms.lastSaveTime = t
}
}
ms.mutex.Unlock()
// 强制进行一次GC
runtime.GC()

View File

@ -19,6 +19,7 @@ type PathMetrics struct {
TotalLatency atomic.Int64 `json:"-"`
BytesTransferred atomic.Int64 `json:"bytes_transferred"`
AvgLatency string `json:"avg_latency"`
LastAccessTime atomic.Int64 `json:"last_access_time"` // 最后访问时间戳
}
// PathMetricsJSON 用于 JSON 序列化的路径统计信息
@ -28,6 +29,7 @@ type PathMetricsJSON struct {
ErrorCount int64 `json:"error_count"`
BytesTransferred int64 `json:"bytes_transferred"`
AvgLatency string `json:"avg_latency"`
LastAccessTime int64 `json:"last_access_time"` // 最后访问时间戳
}
// GetRequestCount 获取请求数
@ -77,7 +79,8 @@ func (p *PathMetrics) ToJSON() PathMetricsJSON {
RequestCount: p.RequestCount.Load(),
ErrorCount: p.ErrorCount.Load(),
BytesTransferred: p.BytesTransferred.Load(),
AvgLatency: p.AvgLatency,
AvgLatency: p.AvgLatency,
LastAccessTime: p.LastAccessTime.Load(),
}
}

View File

@ -40,6 +40,7 @@ interface Metrics {
error_count: number
avg_latency: string
bytes_transferred: number
last_access_time: number // 添加最后访问时间字段
}>
}
@ -188,7 +189,10 @@ export default function DashboardPage() {
<Card>
<CardHeader>
<CardTitle></CardTitle>
<CardTitle>
<span className="ml-2 text-sm font-normal text-gray-500 align-middle">(: {Object.values(metrics.status_code_stats || {}).reduce((a, b) => a + (b as number), 0)})</span>
</CardTitle>
</CardHeader>
<CardContent>
<div className="grid grid-cols-2 md:grid-cols-5 gap-4">
@ -318,7 +322,12 @@ export default function DashboardPage() {
{metrics.top_referers && metrics.top_referers.length > 0 && (
<Card>
<CardHeader>
<CardTitle> (Top {metrics.top_referers.length})</CardTitle>
<CardTitle>
<span className="ml-2 text-sm font-normal text-gray-500 align-middle">
(24, {metrics.top_referers.length} )
</span>
</CardTitle>
</CardHeader>
<CardContent>
<div className="overflow-x-auto">
@ -328,24 +337,49 @@ export default function DashboardPage() {
<th className="text-left p-2"></th>
<th className="text-left p-2"></th>
<th className="text-left p-2"></th>
<th className="text-left p-2"></th>
<th className="text-left p-2"></th>
<th className="text-left p-2"></th>
<th className="text-left p-2">访</th>
</tr>
</thead>
<tbody>
{metrics.top_referers.map((referer, index) => (
<tr key={index} className="border-b">
<td className="p-2 max-w-xs truncate">
<span className="text-blue-600">
{referer.path}
</span>
</td>
<td className="p-2">{referer.request_count}</td>
<td className="p-2">{referer.error_count}</td>
<td className="p-2">{referer.avg_latency}</td>
<td className="p-2">{formatBytes(referer.bytes_transferred)}</td>
</tr>
))}
{metrics.top_referers
.sort((a, b) => b.request_count - a.request_count)
.map((referer, index) => {
const errorRate = ((referer.error_count / referer.request_count) * 100).toFixed(1);
const lastAccessTime = new Date(referer.last_access_time * 1000);
const timeAgo = getTimeAgo(lastAccessTime);
return (
<tr key={index} className="border-b hover:bg-gray-50">
<td className="p-2 max-w-xs truncate">
<a
href={referer.path}
target="_blank"
rel="noopener noreferrer"
className="text-blue-600 hover:text-blue-800 hover:underline"
>
{referer.path}
</a>
</td>
<td className="p-2">{referer.request_count}</td>
<td className="p-2">{referer.error_count}</td>
<td className="p-2">
<span className={errorRate === "0.0" ? "text-green-600" : "text-red-600"}>
{errorRate}%
</span>
</td>
<td className="p-2">{referer.avg_latency}</td>
<td className="p-2">{formatBytes(referer.bytes_transferred)}</td>
<td className="p-2">
<span title={lastAccessTime.toLocaleString()}>
{timeAgo}
</span>
</td>
</tr>
);
})}
</tbody>
</table>
</div>
@ -441,9 +475,30 @@ function formatLatency(nanoseconds: number) {
}
}
function getTimeAgo(date: Date) {
const now = new Date();
const diffInSeconds = Math.floor((now.getTime() - date.getTime()) / 1000);
if (diffInSeconds < 60) {
return `${diffInSeconds}秒前`;
}
const diffInMinutes = Math.floor(diffInSeconds / 60);
if (diffInMinutes < 60) {
return `${diffInMinutes}分钟前`;
}
const diffInHours = Math.floor(diffInMinutes / 60);
if (diffInHours < 24) {
return `${diffInHours}小时前`;
}
return date.toLocaleString();
}
function getStatusColor(status: number) {
if (status >= 500) return "bg-red-100 text-red-800"
if (status >= 400) return "bg-yellow-100 text-yellow-800"
if (status >= 300) return "bg-blue-100 text-blue-800"
return "bg-green-100 text-green-800"
}
}