yuqianhe
3e27017e42
- New CLI commands: key, user - New internal modules: auth (middleware, password, user), db (db, logs, products), email, gateway (gateway, hooks, middlewares), handler (admin, auth, dev, log, product, settings), middleware/geoblock, ratelimit - New demo frontend page - Updated config, server, analytics, stats, and existing frontend pages
662 lines
15 KiB
Go
662 lines
15 KiB
Go
package analytics
|
||
|
||
import (
|
||
"log"
|
||
"math"
|
||
"sort"
|
||
"sync"
|
||
"sync/atomic"
|
||
"time"
|
||
|
||
"api-server/internal/model"
|
||
)
|
||
|
||
// Engine 统计分析引擎（线程安全）
|
||
type Engine struct {
|
||
mu sync.RWMutex
|
||
geoip *GeoIP
|
||
startTime time.Time
|
||
|
||
// 请求记录（环形缓冲区思想，简单用切片 + 定期清理）
|
||
records []model.RequestRecord
|
||
maxRecs int
|
||
|
||
// 并发跟踪
|
||
activeConns int64 // 当前活跃连接（原子操作）
|
||
peakConns int64 // 历史峰值
|
||
connSamples []model.ConcurrencySample // 并发采样
|
||
maxSamples int
|
||
|
||
// 聚合计数器
|
||
totalRequests int64 // 总请求数（原子）
|
||
totalErrors int64 // 总错误数（原子）
|
||
totalTraffic int64 // 总流量字节（原子）
|
||
|
||
// IP 计数器
|
||
ipCount map[string]int64
|
||
|
||
// 端点计数器: "METHOD /path" -> count+latency
|
||
endpointStats map[string]*endpointAgg
|
||
|
||
// 停止信号
|
||
stopCh chan struct{}
|
||
}
|
||
|
||
type endpointAgg struct {
|
||
Count int64
|
||
TotalLatency time.Duration
|
||
Errors int64
|
||
}
|
||
|
||
// NewEngine 创建统计引擎
|
||
func NewEngine(geo *GeoIP) *Engine {
|
||
e := &Engine{
|
||
geoip: geo,
|
||
startTime: time.Now(),
|
||
records: make([]model.RequestRecord, 0, 100000),
|
||
maxRecs: 200000,
|
||
maxSamples: 720, // 默认保留720个采样点（每小时一个，保留30天）
|
||
connSamples: make([]model.ConcurrencySample, 0, 720),
|
||
ipCount: make(map[string]int64),
|
||
endpointStats: make(map[string]*endpointAgg),
|
||
stopCh: make(chan struct{}),
|
||
}
|
||
|
||
// 启动定期清理
|
||
go e.cleanupLoop()
|
||
// 启动并发采样
|
||
go e.concurrencySampler()
|
||
|
||
return e
|
||
}
|
||
|
||
// ── 数据写入 ──────────────────────────────────────────────────
|
||
|
||
// Record 记录一条请求
|
||
func (e *Engine) Record(r model.RequestRecord) {
|
||
// 并发计数
|
||
atomic.AddInt64(&e.totalRequests, 1)
|
||
if r.StatusCode >= 400 {
|
||
atomic.AddInt64(&e.totalErrors, 1)
|
||
}
|
||
atomic.AddInt64(&e.totalTraffic, r.BodyBytes)
|
||
|
||
// IP 地理信息
|
||
if e.geoip != nil {
|
||
info := e.geoip.Lookup(r.ClientIP)
|
||
r.Region = info.Region
|
||
r.ISP = info.ISP
|
||
}
|
||
|
||
e.mu.Lock()
|
||
defer e.mu.Unlock()
|
||
|
||
// 存入记录
|
||
if len(e.records) < e.maxRecs {
|
||
e.records = append(e.records, r)
|
||
}
|
||
|
||
// IP 计数
|
||
e.ipCount[r.ClientIP]++
|
||
|
||
// 端点统计
|
||
key := r.Method + " " + r.Path
|
||
ep, ok := e.endpointStats[key]
|
||
if !ok {
|
||
ep = &endpointAgg{}
|
||
e.endpointStats[key] = ep
|
||
}
|
||
ep.Count++
|
||
ep.TotalLatency += r.Latency
|
||
if r.StatusCode >= 400 {
|
||
ep.Errors++
|
||
}
|
||
}
|
||
|
||
// ── 并发管理 ──────────────────────────────────────────────────
|
||
|
||
// ConnEnter 连接进入
|
||
func (e *Engine) ConnEnter() {
|
||
cur := atomic.AddInt64(&e.activeConns, 1)
|
||
// 更新峰值（非精确，竞态可接受）
|
||
for {
|
||
peak := atomic.LoadInt64(&e.peakConns)
|
||
if cur <= peak || atomic.CompareAndSwapInt64(&e.peakConns, peak, cur) {
|
||
break
|
||
}
|
||
}
|
||
}
|
||
|
||
// ConnLeave 连接离开
|
||
func (e *Engine) ConnLeave() {
|
||
atomic.AddInt64(&e.activeConns, -1)
|
||
}
|
||
|
||
// concurrencySampler 定期采样并发数
|
||
func (e *Engine) concurrencySampler() {
|
||
ticker := time.NewTicker(5 * time.Second)
|
||
defer ticker.Stop()
|
||
for {
|
||
select {
|
||
case <-e.stopCh:
|
||
return
|
||
case t := <-ticker.C:
|
||
e.mu.Lock()
|
||
cur := atomic.LoadInt64(&e.activeConns)
|
||
s := model.ConcurrencySample{
|
||
Time: t.Format("15:04:05"),
|
||
Value: cur,
|
||
}
|
||
if len(e.connSamples) >= e.maxSamples {
|
||
e.connSamples = e.connSamples[1:]
|
||
}
|
||
e.connSamples = append(e.connSamples, s)
|
||
e.mu.Unlock()
|
||
}
|
||
}
|
||
}
|
||
|
||
// ── 清理 ──────────────────────────────────────────────────────
|
||
|
||
func (e *Engine) cleanupLoop() {
|
||
ticker := time.NewTicker(10 * time.Minute)
|
||
defer ticker.Stop()
|
||
for {
|
||
select {
|
||
case <-e.stopCh:
|
||
return
|
||
case <-ticker.C:
|
||
e.cleanup(time.Now().Add(-7 * 24 * time.Hour))
|
||
}
|
||
}
|
||
}
|
||
|
||
// cleanup 清理过期记录
|
||
func (e *Engine) cleanup(before time.Time) {
|
||
e.mu.Lock()
|
||
defer e.mu.Unlock()
|
||
|
||
cutoff := 0
|
||
for i, r := range e.records {
|
||
if r.Timestamp.After(before) {
|
||
cutoff = i
|
||
break
|
||
}
|
||
}
|
||
if cutoff > 0 {
|
||
e.records = e.records[cutoff:]
|
||
}
|
||
}
|
||
|
||
// Stop 停止引擎
|
||
func (e *Engine) Stop() {
|
||
close(e.stopCh)
|
||
}
|
||
|
||
// ── 查询 API ──────────────────────────────────────────────────
|
||
|
||
// Overview 总览统计
|
||
func (e *Engine) Overview() model.OverviewStats {
|
||
e.mu.RLock()
|
||
defer e.mu.RUnlock()
|
||
|
||
total := atomic.LoadInt64(&e.totalRequests)
|
||
errors := atomic.LoadInt64(&e.totalErrors)
|
||
active := atomic.LoadInt64(&e.activeConns)
|
||
traffic := atomic.LoadInt64(&e.totalTraffic)
|
||
|
||
// 平均延迟 & QPS
|
||
avgLat := 0.0
|
||
qps := 0.0
|
||
uptime := time.Since(e.startTime)
|
||
if uptime.Seconds() > 0 {
|
||
qps = float64(total) / uptime.Seconds()
|
||
}
|
||
|
||
recs := e.recentRecords(time.Minute)
|
||
if len(recs) > 0 {
|
||
var sum time.Duration
|
||
for _, r := range recs {
|
||
sum += r.Latency
|
||
}
|
||
avgLat = float64(sum.Microseconds()) / float64(len(recs)) / 1000.0
|
||
}
|
||
|
||
lastReqTime := time.Time{}
|
||
if len(e.records) > 0 {
|
||
lastReqTime = e.records[len(e.records)-1].Timestamp
|
||
}
|
||
|
||
errRate := 0.0
|
||
if total > 0 {
|
||
errRate = float64(errors) / float64(total) * 100
|
||
}
|
||
|
||
uniqueIPs := int64(len(e.ipCount))
|
||
|
||
return model.OverviewStats{
|
||
TotalRequests: total,
|
||
ActiveConn: active,
|
||
AvgLatency: math.Round(avgLat*100) / 100,
|
||
ErrorRate: math.Round(errRate*100) / 100,
|
||
RequestsPerSec: math.Round(qps*100) / 100,
|
||
Uptime: uptime,
|
||
UniqueIPs: uniqueIPs,
|
||
TotalTraffic: traffic / (1024 * 1024),
|
||
LastRequestTime: lastReqTime,
|
||
}
|
||
}
|
||
|
||
// TimeSeries 时间序列数据
|
||
func (e *Engine) TimeSeries(period time.Duration, buckets int) []model.TimeSeriesPoint {
|
||
e.mu.RLock()
|
||
defer e.mu.RUnlock()
|
||
|
||
if buckets <= 0 {
|
||
buckets = 60
|
||
}
|
||
now := time.Now()
|
||
start := now.Add(-period)
|
||
interval := period / time.Duration(buckets)
|
||
|
||
// 过滤时间段内的记录
|
||
recs := e.recordsInRange(start, now)
|
||
if len(recs) == 0 {
|
||
return nil
|
||
}
|
||
|
||
bucketsData := make([]struct {
|
||
count int
|
||
errors int
|
||
latSum time.Duration
|
||
}, buckets)
|
||
|
||
for _, r := range recs {
|
||
idx := int(r.Timestamp.Sub(start) / interval)
|
||
if idx < 0 {
|
||
idx = 0
|
||
}
|
||
if idx >= buckets {
|
||
idx = buckets - 1
|
||
}
|
||
bucketsData[idx].count++
|
||
if r.StatusCode >= 400 {
|
||
bucketsData[idx].errors++
|
||
}
|
||
bucketsData[idx].latSum += r.Latency
|
||
}
|
||
|
||
result := make([]model.TimeSeriesPoint, buckets)
|
||
for i := 0; i < buckets; i++ {
|
||
t := start.Add(interval * time.Duration(i))
|
||
b := bucketsData[i]
|
||
avgMs := 0.0
|
||
if b.count > 0 {
|
||
avgMs = float64(b.latSum.Microseconds()) / float64(b.count) / 1000.0
|
||
}
|
||
result[i] = model.TimeSeriesPoint{
|
||
Time: t.Format("15:04"),
|
||
Count: int64(b.count),
|
||
Errors: int64(b.errors),
|
||
AvgMs: math.Round(avgMs*100) / 100,
|
||
}
|
||
}
|
||
return result
|
||
}
|
||
|
||
// GeoDistribution 地理分布 Top N
|
||
func (e *Engine) GeoDistribution(topN int) []model.GeoDistribution {
|
||
e.mu.RLock()
|
||
defer e.mu.RUnlock()
|
||
|
||
regionCount := make(map[string]int64)
|
||
|
||
// 只统计最近1小时
|
||
cutoff := time.Now().Add(-1 * time.Hour)
|
||
for _, r := range e.records {
|
||
if r.Timestamp.After(cutoff) {
|
||
regionCount[r.Region]++
|
||
}
|
||
}
|
||
|
||
total := int64(0)
|
||
for _, c := range regionCount {
|
||
total += c
|
||
}
|
||
|
||
type kv struct {
|
||
k string
|
||
v int64
|
||
}
|
||
var sorted []kv
|
||
for k, v := range regionCount {
|
||
sorted = append(sorted, kv{k, v})
|
||
}
|
||
sort.Slice(sorted, func(i, j int) bool { return sorted[i].v > sorted[j].v })
|
||
|
||
if topN > 0 && len(sorted) > topN {
|
||
sorted = sorted[:topN]
|
||
}
|
||
|
||
result := make([]model.GeoDistribution, len(sorted))
|
||
for i, kv := range sorted {
|
||
pct := 0.0
|
||
if total > 0 {
|
||
pct = float64(kv.v) / float64(total) * 100
|
||
}
|
||
result[i] = model.GeoDistribution{
|
||
Region: kv.k,
|
||
Count: kv.v,
|
||
Pct: math.Round(pct*100) / 100,
|
||
}
|
||
}
|
||
return result
|
||
}
|
||
|
||
// Concurrency 并发统计
|
||
func (e *Engine) Concurrency() model.ConcurrencyStats {
|
||
e.mu.RLock()
|
||
defer e.mu.RUnlock()
|
||
|
||
cur := atomic.LoadInt64(&e.activeConns)
|
||
peak := atomic.LoadInt64(&e.peakConns)
|
||
|
||
avg := 0.0
|
||
if len(e.connSamples) > 0 {
|
||
var sum int64
|
||
for _, s := range e.connSamples {
|
||
sum += s.Value
|
||
}
|
||
avg = float64(sum) / float64(len(e.connSamples))
|
||
}
|
||
|
||
return model.ConcurrencyStats{
|
||
Current: cur,
|
||
Peak: peak,
|
||
Avg: math.Round(avg*100) / 100,
|
||
Samples: e.connSamples,
|
||
}
|
||
}
|
||
|
||
// Latency 延迟统计
|
||
func (e *Engine) Latency() model.LatencyStats {
|
||
e.mu.RLock()
|
||
defer e.mu.RUnlock()
|
||
|
||
recs := e.recentRecords(5 * time.Minute)
|
||
if len(recs) == 0 {
|
||
return model.LatencyStats{}
|
||
}
|
||
|
||
latencies := make([]float64, len(recs))
|
||
var sum, maxVal, minVal float64
|
||
minVal = math.MaxFloat64
|
||
for i, r := range recs {
|
||
ms := float64(r.Latency.Microseconds()) / 1000.0
|
||
latencies[i] = ms
|
||
sum += ms
|
||
if ms > maxVal {
|
||
maxVal = ms
|
||
}
|
||
if ms < minVal {
|
||
minVal = ms
|
||
}
|
||
}
|
||
|
||
sort.Float64s(latencies)
|
||
n := len(latencies)
|
||
|
||
percentile := func(p float64) float64 {
|
||
idx := int(math.Ceil(p*float64(n))) - 1
|
||
if idx < 0 {
|
||
idx = 0
|
||
}
|
||
if idx >= n {
|
||
idx = n - 1
|
||
}
|
||
return latencies[idx]
|
||
}
|
||
|
||
return model.LatencyStats{
|
||
Avg: math.Round(sum/float64(n)*100) / 100,
|
||
P50: math.Round(percentile(0.50)*100) / 100,
|
||
P90: math.Round(percentile(0.90)*100) / 100,
|
||
P95: math.Round(percentile(0.95)*100) / 100,
|
||
P99: math.Round(percentile(0.99)*100) / 100,
|
||
Max: math.Round(maxVal*100) / 100,
|
||
Min: math.Round(minVal*100) / 100,
|
||
Count: int64(n),
|
||
}
|
||
}
|
||
|
||
// StatusDist 状态码分布
|
||
func (e *Engine) StatusDist(period ...time.Duration) []model.StatusDistribution {
|
||
p := time.Hour
|
||
if len(period) > 0 && period[0] > 0 {
|
||
p = period[0]
|
||
}
|
||
|
||
e.mu.RLock()
|
||
defer e.mu.RUnlock()
|
||
|
||
counts := map[string]int64{"2xx": 0, "3xx": 0, "4xx": 0, "5xx": 0}
|
||
var total int64
|
||
|
||
recs := e.recentRecords(p)
|
||
for _, r := range recs {
|
||
switch {
|
||
case r.StatusCode >= 500:
|
||
counts["5xx"]++
|
||
case r.StatusCode >= 400:
|
||
counts["4xx"]++
|
||
case r.StatusCode >= 300:
|
||
counts["3xx"]++
|
||
default:
|
||
counts["2xx"]++
|
||
}
|
||
total++
|
||
}
|
||
|
||
order := []string{"2xx", "3xx", "4xx", "5xx"}
|
||
result := make([]model.StatusDistribution, 0, 4)
|
||
for _, code := range order {
|
||
ct := counts[code]
|
||
if ct > 0 || total > 0 {
|
||
pct := 0.0
|
||
if total > 0 {
|
||
pct = float64(ct) / float64(total) * 100
|
||
}
|
||
result = append(result, model.StatusDistribution{
|
||
Code: code,
|
||
Count: ct,
|
||
Pct: math.Round(pct*100) / 100,
|
||
})
|
||
}
|
||
}
|
||
return result
|
||
}
|
||
|
||
// TopEndpoints Top N 端点（period=0 表示全部时间）
|
||
func (e *Engine) TopEndpoints(topN int, period ...time.Duration) []model.EndpointStats {
|
||
e.mu.RLock()
|
||
defer e.mu.RUnlock()
|
||
|
||
recs := e.records
|
||
if len(period) > 0 && period[0] > 0 {
|
||
recs = e.recentRecords(period[0])
|
||
}
|
||
|
||
// 聚合端点数据（考虑时间范围过滤）
|
||
type kv struct {
|
||
k string
|
||
v *endpointAgg
|
||
}
|
||
epMap := make(map[string]*endpointAgg)
|
||
for _, r := range recs {
|
||
key := r.Method + " " + r.Path
|
||
ep, ok := epMap[key]
|
||
if !ok {
|
||
ep = &endpointAgg{}
|
||
epMap[key] = ep
|
||
}
|
||
ep.Count++
|
||
ep.TotalLatency += r.Latency
|
||
if r.StatusCode >= 400 {
|
||
ep.Errors++
|
||
}
|
||
}
|
||
var pairs []kv
|
||
for k, v := range epMap {
|
||
pairs = append(pairs, kv{k, v})
|
||
}
|
||
sort.Slice(pairs, func(i, j int) bool { return pairs[i].v.Count > pairs[j].v.Count })
|
||
|
||
if topN > 0 && len(pairs) > topN {
|
||
pairs = pairs[:topN]
|
||
}
|
||
|
||
result := make([]model.EndpointStats, len(pairs))
|
||
for i, p := range pairs {
|
||
avgLat := 0.0
|
||
if p.v.Count > 0 {
|
||
avgLat = float64(p.v.TotalLatency.Microseconds()) / float64(p.v.Count) / 1000.0
|
||
}
|
||
errRate := 0.0
|
||
if p.v.Count > 0 {
|
||
errRate = float64(p.v.Errors) / float64(p.v.Count) * 100
|
||
}
|
||
result[i] = model.EndpointStats{
|
||
Path: p.k,
|
||
Count: p.v.Count,
|
||
AvgLatency: math.Round(avgLat*100) / 100,
|
||
ErrorRate: math.Round(errRate*100) / 100,
|
||
}
|
||
}
|
||
return result
|
||
}
|
||
|
||
// TopIPs Top N IP（period=0 表示全部时间）
|
||
func (e *Engine) TopIPs(topN int, period ...time.Duration) []model.TopIP {
|
||
e.mu.RLock()
|
||
defer e.mu.RUnlock()
|
||
|
||
ipCount := e.ipCount
|
||
if len(period) > 0 && period[0] > 0 {
|
||
ipCount = make(map[string]int64)
|
||
for _, r := range e.recentRecords(period[0]) {
|
||
ipCount[r.ClientIP]++
|
||
}
|
||
}
|
||
|
||
type kv struct {
|
||
ip string
|
||
ct int64
|
||
}
|
||
var pairs []kv
|
||
for ip, ct := range ipCount {
|
||
pairs = append(pairs, kv{ip, ct})
|
||
}
|
||
sort.Slice(pairs, func(i, j int) bool { return pairs[i].ct > pairs[j].ct })
|
||
|
||
if topN > 0 && len(pairs) > topN {
|
||
pairs = pairs[:topN]
|
||
}
|
||
|
||
result := make([]model.TopIP, len(pairs))
|
||
for i, p := range pairs {
|
||
region := "未知"
|
||
if e.geoip != nil {
|
||
region = e.geoip.Lookup(p.ip).Region
|
||
}
|
||
result[i] = model.TopIP{
|
||
IP: p.ip,
|
||
Region: region,
|
||
Count: p.ct,
|
||
}
|
||
}
|
||
return result
|
||
}
|
||
|
||
// Realtime 实时统计
|
||
func (e *Engine) Realtime() model.RealtimeStats {
|
||
e.mu.RLock()
|
||
defer e.mu.RUnlock()
|
||
|
||
active := atomic.LoadInt64(&e.activeConns)
|
||
recs := e.recentRecords(time.Minute)
|
||
|
||
var totalLatMs float64
|
||
errors := int64(0)
|
||
for _, r := range recs {
|
||
totalLatMs += float64(r.Latency.Microseconds()) / 1000.0
|
||
if r.StatusCode >= 400 {
|
||
errors++
|
||
}
|
||
}
|
||
|
||
avgLat := 0.0
|
||
if len(recs) > 0 {
|
||
avgLat = totalLatMs / float64(len(recs))
|
||
}
|
||
|
||
qps := float64(len(recs)) / 60.0
|
||
|
||
errRate := 0.0
|
||
if len(recs) > 0 {
|
||
errRate = float64(errors) / float64(len(recs)) * 100
|
||
}
|
||
|
||
return model.RealtimeStats{
|
||
QPS: math.Round(qps*100) / 100,
|
||
ActiveConn: active,
|
||
AvgLatencyMs: math.Round(avgLat*100) / 100,
|
||
LastMinuteCount: int64(len(recs)),
|
||
ErrorRate: math.Round(errRate*100) / 100,
|
||
}
|
||
}
|
||
|
||
// ── 内部辅助 ──────────────────────────────────────────────────
|
||
|
||
// recentRecords 获取最近 duration 内的记录（调用方需持有锁）
|
||
func (e *Engine) recentRecords(d time.Duration) []model.RequestRecord {
|
||
cutoff := time.Now().Add(-d)
|
||
for i := len(e.records) - 1; i >= 0; i-- {
|
||
if e.records[i].Timestamp.Before(cutoff) {
|
||
return e.records[i+1:]
|
||
}
|
||
}
|
||
return e.records
|
||
}
|
||
|
||
// recordsInRange 获取时间范围内的记录（调用方需持有锁）
|
||
func (e *Engine) recordsInRange(start, end time.Time) []model.RequestRecord {
|
||
if len(e.records) == 0 {
|
||
return nil
|
||
}
|
||
|
||
// 二分查找起始位置
|
||
startIdx := sort.Search(len(e.records), func(i int) bool {
|
||
return !e.records[i].Timestamp.Before(start)
|
||
})
|
||
if startIdx >= len(e.records) {
|
||
return nil
|
||
}
|
||
|
||
// 收集范围内的记录
|
||
var result []model.RequestRecord
|
||
for i := startIdx; i < len(e.records); i++ {
|
||
if e.records[i].Timestamp.After(end) {
|
||
break
|
||
}
|
||
result = append(result, e.records[i])
|
||
}
|
||
return result
|
||
}
|
||
|
||
// LogStats 定期输出统计摘要到日志
|
||
func (e *Engine) LogStats() {
|
||
o := e.Overview()
|
||
log.Printf("[stats] 总请求=%d | 活跃=%d | QPS=%.1f | 错误率=%.1f%% | 延迟=%.1fms | 独立IP=%d",
|
||
o.TotalRequests, o.ActiveConn, o.RequestsPerSec, o.ErrorRate, o.AvgLatency, o.UniqueIPs)
|
||
}
|