Files
aiops/backend/internal/api/handlers/metrics_ws.go
T
attacker f3a9e5e8d0 feat: 重构监控中心导航,新增资源监控4大模块
- 主菜单「监控告警」→「监控中心」
- 「告警中心」→「告警服务」移至链路追踪下方
- 新增资源监控 Tab 页(主机/中间件/SNMP/站点拨测)
- Agent 新增网络采集器(network)和磁盘IO采集器(diskio)
- 后端扩展 Agent 上报模型,支持 NetworkStat/DiskIOStat
- 前端 ServerTableRow 展示网络和磁盘IO实时指标
- WebSocket MetricsUpdate 推送扩展
2026-06-18 09:30:45 +00:00

121 lines
3.1 KiB
Go

package handlers
import (
"encoding/json"
"log"
"sync"
"github.com/gin-gonic/gin"
"github.com/gorilla/websocket"
)
// MetricsUpdate 单个主机的指标更新 // Logan.Li
type MetricsUpdate struct {
Type string `json:"type"` // "metrics_update"
Hostname string `json:"hostname"`
CPU float64 `json:"cpu_usage"`
Memory float64 `json:"memory_usage"`
Disk float64 `json:"disk_usage"`
Status string `json:"status"`
TCPEstablished int `json:"tcp_established,omitempty"`
TCPTimeWait int `json:"tcp_time_wait,omitempty"`
NetRxBPS float64 `json:"net_rx_bps,omitempty"`
NetTxBPS float64 `json:"net_tx_bps,omitempty"`
DiskReadBPS float64 `json:"disk_read_bps,omitempty"`
DiskWriteBPS float64 `json:"disk_write_bps,omitempty"`
}
// MetricsHub 管理所有 WebSocket 客户端连接,广播指标更新 // Logan.Li
type MetricsHub struct {
mu sync.RWMutex
clients map[*websocket.Conn]bool
}
// NewMetricsHub 创建 MetricsHub 实例
func NewMetricsHub() *MetricsHub {
return &MetricsHub{
clients: make(map[*websocket.Conn]bool),
}
}
// Register 注册新的 WebSocket 连接
func (h *MetricsHub) Register(conn *websocket.Conn) {
h.mu.Lock()
h.clients[conn] = true
h.mu.Unlock()
log.Printf("[MetricsWS] 客户端已连接,当前连接数: %d", h.Count())
}
// Unregister 移除断开的 WebSocket 连接
func (h *MetricsHub) Unregister(conn *websocket.Conn) {
h.mu.Lock()
delete(h.clients, conn)
h.mu.Unlock()
log.Printf("[MetricsWS] 客户端已断开,当前连接数: %d", h.Count())
}
// Count 返回当前连接数
func (h *MetricsHub) Count() int {
h.mu.RLock()
defer h.mu.RUnlock()
return len(h.clients)
}
// Broadcast 向所有连接的客户端广播指标更新
func (h *MetricsHub) Broadcast(update MetricsUpdate) {
data, err := json.Marshal(update)
if err != nil {
log.Printf("[MetricsWS] JSON 序列化失败: %v", err)
return
}
h.mu.RLock()
defer h.mu.RUnlock()
for conn := range h.clients {
if err := conn.WriteMessage(websocket.TextMessage, data); err != nil {
log.Printf("[MetricsWS] 发送失败: %v", err)
go func(c *websocket.Conn) {
h.Unregister(c)
c.Close()
}(conn)
}
}
}
// MetricsWSHandler 处理 /ws/metrics 的 WebSocket 连接 // Logan.Li
type MetricsWSHandler struct {
hub *MetricsHub
}
// NewMetricsWSHandler 创建 WebSocket metrics 处理器
func NewMetricsWSHandler(hub *MetricsHub) *MetricsWSHandler {
return &MetricsWSHandler{hub: hub}
}
// ServeWS 升级 HTTP → WebSocket,注册到 Hub,保持连接
func (h *MetricsWSHandler) ServeWS(c *gin.Context) {
conn, err := upgrader.Upgrade(c.Writer, c.Request, nil)
if err != nil {
log.Printf("[MetricsWS] WebSocket 升级失败: %v", err)
return
}
h.hub.Register(conn)
defer func() {
h.hub.Unregister(conn)
conn.Close()
}()
// 保持连接,读取消息(客户端可能发 ping)
for {
_, _, err := conn.ReadMessage()
if err != nil {
if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseNormalClosure) {
log.Printf("[MetricsWS] 连接异常关闭: %v", err)
}
break
}
}
}