yml/kefu_cloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

接入运维真实 metrics：负载、请求错误率与服务探测

Browse Source 
- 新增 GET /admin/ops/metrics 与进程内请求统计中间件
- 运维页展示 CPU/内存/磁盘、24h 曲线与 WS/DB/存储状态
- 去掉未使用的消息队列占位项
This commit is contained in:
yml2213
2026-07-15 15:01:18 +08:00
parent a1336c630b
commit 9251815695
14 changed files with 797 additions and 248 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/internal/handler/admin.go
+7 -2
View File
@@ -10,11 +10,16 @@ import (
"golang.org/x/crypto/bcrypt"
"kefu-sys/server/internal/middleware"
"kefu-sys/server/internal/model"
"kefu-sys/server/internal/storage"
)
type AdminHandler struct{}
type AdminHandler struct {
store storage.ObjectStorage
}
func NewAdminHandler() *AdminHandler { return &AdminHandler{} }
func NewAdminHandler(store storage.ObjectStorage) *AdminHandler {
return &AdminHandler{store: store}
}
func (h *AdminHandler) Stats(c *gin.Context) {
var tenantTotal, activeTotal, suspendedTotal, expiringTotal, newThisMonth int64
server/internal/handler/ops_metrics.go
+230
View File
@@ -0,0 +1,230 @@
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
}
server/internal/handler/router.go
+2 -1
View File
@@ -13,7 +13,7 @@ func SetupRoutes(r *gin.Engine, store storage.ObjectStorage, storageCfg config.S
customer := NewCustomerHandler()
knowledge := NewKnowledgeHandler()
stats := NewStatisticsHandler()
admin := NewAdminHandler()
admin := NewAdminHandler(store)
channel := NewChannelHandler()
settings := NewSettingsHandler()
staff := NewStaffHandler()
@@ -116,6 +116,7 @@ func SetupRoutes(r *gin.Engine, store storage.ObjectStorage, storageCfg config.S
adminGroup.Use(middleware.PlatformRequired())
{
adminGroup.GET("/stats", admin.Stats)
adminGroup.GET("/ops/metrics", admin.OpsMetrics)
adminGroup.GET("/tenants", admin.ListTenants)
adminGroup.GET("/tenants/:id", admin.GetTenant)
adminGroup.POST("/tenants", admin.CreateTenant)
server/internal/metrics/metrics.go
+140
View File
@@ -0,0 +1,140 @@
package metrics
import (
"sync"
"sync/atomic"
"time"
"github.com/gin-gonic/gin"
)
// Collector 进程内请求统计：近 24 小时按整点小时桶累加。
type Collector struct {
mu sync.Mutex
byHour map[int64]*hourBucket
// 自进程启动以来的累计
totalReq atomic.Int64
totalErr atomic.Int64 // HTTP status >= 400
total5xx atomic.Int64
inFlight atomic.Int64
startedAt time.Time
}
type hourBucket struct {
requests int64
errors int64
status5 int64
}
// Default 全局采集器
var Default = NewCollector()
func NewCollector() *Collector {
return &Collector{
byHour: make(map[int64]*hourBucket),
startedAt: time.Now(),
}
}
func hourKey(t time.Time) int64 {
return t.Unix() / 3600
}
func (c *Collector) Record(status int) {
c.totalReq.Add(1)
isErr := status >= 400
is5 := status >= 500
if isErr {
c.totalErr.Add(1)
}
if is5 {
c.total5xx.Add(1)
}
h := hourKey(time.Now())
c.mu.Lock()
b := c.byHour[h]
if b == nil {
b = &hourBucket{}
c.byHour[h] = b
c.pruneLocked(h)
}
b.requests++
if isErr {
b.errors++
}
if is5 {
b.status5++
}
c.mu.Unlock()
}
func (c *Collector) pruneLocked(nowHour int64) {
cutoff := nowHour - 25
for k := range c.byHour {
if k < cutoff {
delete(c.byHour, k)
}
}
}
// Snapshot24h 返回近 24 个整点小时（含当前小时）的请求量与错误率（0–100）。
// 下标 0 为 23 小时前，23 为当前小时；hours 为当地小时 0–23。
func (c *Collector) Snapshot24h() (requests []int64, errorRates []float64, hours []int, totalReq, totalErr int64) {
now := time.Now()
nowH := hourKey(now)
loc := now.Location()
requests = make([]int64, 24)
errorRates = make([]float64, 24)
hours = make([]int, 24)
c.mu.Lock()
defer c.mu.Unlock()
for i := 0; i < 24; i++ {
hKey := nowH - int64(23-i)
t := time.Unix(hKey*3600, 0).In(loc)
hours[i] = t.Hour()
b := c.byHour[hKey]
var r, e int64
if b != nil {
r, e = b.requests, b.errors
}
requests[i] = r
if r > 0 {
errorRates[i] = float64(e) / float64(r) * 100
} else {
errorRates[i] = 0
}
totalReq += r
totalErr += e
}
return
}
func (c *Collector) Totals() (req, err4, err5 int64, startedAt time.Time) {
return c.totalReq.Load(), c.totalErr.Load(), c.total5xx.Load(), c.startedAt
}
func (c *Collector) InFlight() int64 {
return c.inFlight.Load()
}
// Middleware 记录每个 HTTP 请求的状态码（跳过 WebSocket 升级失败后的统计由 status 决定）。
func Middleware() gin.HandlerFunc {
return func(c *gin.Context) {
// 健康检查可不计入业务错误曲线，但仍记请求量更直观；这里全部计入
Default.inFlight.Add(1)
c.Next()
Default.inFlight.Add(-1)
// WebSocket 升级成功后 status 可能是 101
status := c.Writer.Status()
if status == 0 {
status = 200
}
Default.Record(status)
}
}
server/internal/metrics/metrics_test.go
+32
View File
@@ -0,0 +1,32 @@
package metrics
import (
"testing"
)
func TestRecordAndSnapshot(t *testing.T) {
c := NewCollector()
c.Record(200)
c.Record(200)
c.Record(500)
c.Record(404)
reqs, rates, hours, totalReq, totalErr := c.Snapshot24h()
if len(reqs) != 24 || len(rates) != 24 || len(hours) != 24 {
t.Fatalf("snapshot 长度应为 24: req=%d rates=%d hours=%d", len(reqs), len(rates), len(hours))
}
if totalReq != 4 || totalErr != 2 {
t.Fatalf("窗口统计错误: totalReq=%d totalErr=%d", totalReq, totalErr)
}
// 当前小时桶应有 4 请求
if reqs[23] != 4 {
t.Fatalf("当前小时请求量 = %d, 期望 4", reqs[23])
}
if rates[23] < 49 || rates[23] > 51 {
t.Fatalf("当前小时错误率 = %v, 期望约 50", rates[23])
}
tr, te, t5, _ := c.Totals()
if tr != 4 || te != 2 || t5 != 1 {
t.Fatalf("累计统计错误: req=%d err=%d 5xx=%d", tr, te, t5)
}
}
server/internal/storage/memory.go
+2
View File
@@ -44,6 +44,8 @@ func (m *MemoryStorage) PublicBase() string { return m.base }
func (m *MemoryStorage) EnsureBucket(context.Context) error { return nil }
func (m *MemoryStorage) HealthCheck(context.Context) error { return nil }
func (m *MemoryStorage) Get(key string) ([]byte, error) {
m.mu.RLock()
defer m.mu.RUnlock()
server/internal/storage/s3.go
+5
View File
@@ -40,6 +40,11 @@ func NewS3(cfg config.StorageConfig) (*S3Storage, error) {
return &S3Storage{client: client, bucket: cfg.Bucket, base: base}, nil
}
func (s *S3Storage) HealthCheck(ctx context.Context) error {
_, err := s.client.BucketExists(ctx, s.bucket)
return err
}
func (s *S3Storage) EnsureBucket(ctx context.Context) error {
exists, err := s.client.BucketExists(ctx, s.bucket)
if err != nil {
server/internal/storage/storage.go
+2
View File
@@ -17,6 +17,8 @@ type ObjectStorage interface {
PublicBase() string
// EnsureBucket 开发环境自动建桶
EnsureBucket(ctx context.Context) error
// HealthCheck 探测存储是否可达（运维监控用）
HealthCheck(ctx context.Context) error
}
// PutResult 上传结果
server/internal/ws/ws.go
+15
View File
@@ -83,6 +83,21 @@ func (h *Hub) Run() {
}
}
// Stats 返回当前连接统计（总连接 / 坐席 / 访客）。
func (h *Hub) Stats() (total, agents, visitors int) {
h.mu.RLock()
defer h.mu.RUnlock()
for client := range h.clients {
total++
if client.Kind == "visitor" {
visitors++
} else {
agents++
}
}
return
}
func (h *Hub) send(client *Client, message []byte) {
select {
case client.Send <- message: