Files
kefu_cloud/server/internal/handler/ops_metrics.go
T
yml2213 9251815695 接入运维真实 metrics:负载、请求错误率与服务探测
- 新增 GET /admin/ops/metrics 与进程内请求统计中间件
- 运维页展示 CPU/内存/磁盘、24h 曲线与 WS/DB/存储状态
- 去掉未使用的消息队列占位项
2026-07-15 15:01:18 +08:00

231 lines
6.1 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package handler
import (
"context"
"fmt"
"runtime"
"time"
"github.com/gin-gonic/gin"
"github.com/shirou/gopsutil/v4/cpu"
"github.com/shirou/gopsutil/v4/disk"
"github.com/shirou/gopsutil/v4/mem"
"kefu-sys/server/internal/metrics"
"kefu-sys/server/internal/middleware"
"kefu-sys/server/internal/model"
"kefu-sys/server/internal/storage"
"kefu-sys/server/internal/ws"
)
// OpsMetrics 平台运维实时指标(当前 API 实例视角)。
// GET /api/admin/ops/metrics
func (h *AdminHandler) OpsMetrics(c *gin.Context) {
now := time.Now()
// —— 系统负载 ——
load := collectSystemLoad()
// —— 请求统计 ——
reqSeries, errSeries, hourLabels, winReq, winErr := metrics.Default.Snapshot24h()
totalReq, totalErr, total5xx, startedAt := metrics.Default.Totals()
var errNow float64
if winReq > 0 {
errNow = float64(winErr) / float64(winReq) * 100
}
var qps float64
uptimeSec := now.Sub(startedAt).Seconds()
if uptimeSec > 0 {
qps = float64(totalReq) / uptimeSec
}
// —— 服务探测 ——
services := collectServiceHealth(h.store)
// —— 整体状态 ——
overall := "normal"
for _, s := range services {
if st, _ := s["status"].(string); st == "error" {
overall = "error"
break
}
if st, _ := s["status"].(string); st == "warning" && overall == "normal" {
overall = "warning"
}
}
if load.CPUPercent >= 90 || load.MemPercent >= 90 || load.DiskPercent >= 95 {
if overall == "normal" {
overall = "warning"
}
}
if errNow >= 5 {
overall = "error"
} else if errNow >= 2 && overall == "normal" {
overall = "warning"
}
wsTotal, wsAgents, wsVisitors := ws.DefaultHub.Stats()
middleware.JSON(c, gin.H{
"overall": overall,
"updated_at": now.Format(time.RFC3339),
"instance": "api", // 当前进程实例
"uptime_sec": int64(uptimeSec),
"services": services,
"load": gin.H{
"cpu_percent": load.CPUPercent,
"mem_percent": load.MemPercent,
"disk_percent": load.DiskPercent,
"cpu_cores": load.CPUCores,
"mem_total_mb": load.MemTotalMB,
"mem_used_mb": load.MemUsedMB,
"disk_total_gb": load.DiskTotalGB,
"disk_used_gb": load.DiskUsedGB,
"go_goroutines": runtime.NumGoroutine(),
"go_heap_mb": load.GoHeapMB,
},
"api": gin.H{
"requests_last_24h": reqSeries,
"error_rate_last_24h": errSeries,
"hours": hourLabels,
"qps": round2(qps),
"error_rate_now": round2(errNow),
"window_requests": winReq,
"window_errors": winErr,
"total_requests": totalReq,
"total_errors": totalErr,
"total_5xx": total5xx,
"in_flight": metrics.Default.InFlight(),
},
"websocket": gin.H{
"connections": wsTotal,
"agents": wsAgents,
"visitors": wsVisitors,
},
})
}
type systemLoad struct {
CPUPercent float64
MemPercent float64
DiskPercent float64
CPUCores int
MemTotalMB uint64
MemUsedMB uint64
DiskTotalGB float64
DiskUsedGB float64
GoHeapMB float64
}
func collectSystemLoad() systemLoad {
out := systemLoad{
CPUCores: runtime.NumCPU(),
}
var ms runtime.MemStats
runtime.ReadMemStats(&ms)
out.GoHeapMB = round2(float64(ms.HeapAlloc) / 1024 / 1024)
if percents, err := cpu.Percent(120*time.Millisecond, false); err == nil && len(percents) > 0 {
out.CPUPercent = round2(percents[0])
}
if vm, err := mem.VirtualMemory(); err == nil {
out.MemPercent = round2(vm.UsedPercent)
out.MemTotalMB = vm.Total / 1024 / 1024
out.MemUsedMB = vm.Used / 1024 / 1024
}
if du, err := disk.Usage("/"); err == nil {
out.DiskPercent = round2(du.UsedPercent)
out.DiskTotalGB = round2(float64(du.Total) / 1024 / 1024 / 1024)
out.DiskUsedGB = round2(float64(du.Used) / 1024 / 1024 / 1024)
}
return out
}
func collectServiceHealth(store storage.ObjectStorage) []gin.H {
services := make([]gin.H, 0, 5)
// API:本进程存活
services = append(services, gin.H{
"name": "API服务",
"status": "normal",
"detail": "进程运行中",
"latency_ms": 0,
})
// WebSocket Hub
wsTotal, wsAgents, wsVisitors := ws.DefaultHub.Stats()
services = append(services, gin.H{
"name": "WebSocket",
"status": "normal",
"detail": fmt.Sprintf("连接 %d(坐席 %d / 访客 %d", wsTotal, wsAgents, wsVisitors),
"connections": wsTotal,
})
// 数据库
dbStatus, dbDetail, dbLatency := probeDatabase()
services = append(services, gin.H{
"name": "数据库",
"status": dbStatus,
"detail": dbDetail,
"latency_ms": dbLatency,
})
// 对象存储
stStatus, stDetail, stLatency := probeStorage(store)
services = append(services, gin.H{
"name": "对象存储",
"status": stStatus,
"detail": stDetail,
"latency_ms": stLatency,
})
return services
}
func probeDatabase() (status, detail string, latencyMs int64) {
if model.DB == nil {
return "error", "数据库未初始化", 0
}
sqlDB, err := model.DB.DB()
if err != nil {
return "error", "获取连接失败: " + err.Error(), 0
}
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
start := time.Now()
if err := sqlDB.PingContext(ctx); err != nil {
return "error", "Ping 失败: " + err.Error(), time.Since(start).Milliseconds()
}
lat := time.Since(start).Milliseconds()
status = "normal"
detail = fmt.Sprintf("Ping 正常 · %dms", lat)
if lat >= 200 {
status = "warning"
detail = fmt.Sprintf("Ping 偏慢 · %dms", lat)
}
return status, detail, lat
}
func probeStorage(store storage.ObjectStorage) (status, detail string, latencyMs int64) {
if store == nil {
return "warning", "未配置对象存储", 0
}
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
start := time.Now()
if err := store.HealthCheck(ctx); err != nil {
return "error", "不可达: " + err.Error(), time.Since(start).Milliseconds()
}
lat := time.Since(start).Milliseconds()
status = "normal"
detail = fmt.Sprintf("Bucket 可达 · %dms", lat)
if lat >= 500 {
status = "warning"
detail = fmt.Sprintf("探测偏慢 · %dms", lat)
}
return status, detail, lat
}
func round2(v float64) float64 {
return float64(int(v*100+0.5)) / 100
}