package controller import ( "encoding/json" "log" "sync" "ezops/middleware" "ezops/pkg/wshub" "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 wsMgr *wshub.WSManager } // NewMetricsWSHandler 创建 WebSocket metrics 处理器 func NewMetricsWSHandler(hub *MetricsHub, wsMgr *wshub.WSManager) *MetricsWSHandler { return &MetricsWSHandler{hub: hub, wsMgr: wsMgr} } // ServeWS 升级 HTTP → WebSocket,注册到 Hub,保持连接 func (h *MetricsWSHandler) ServeWS(c *gin.Context) { userID := middleware.GetUserID(c) username := middleware.GetUsername(c) conn, info, err := h.wsMgr.Upgrade(c, wshub.TypeMetrics, userID, username) if err != nil { log.Printf("[MetricsWS] WebSocket 升级失败: %v", err) return } h.hub.Register(conn) defer func() { h.hub.Unregister(conn) h.wsMgr.Unregister(info) }() h.wsMgr.StartHeartbeat(conn, info) // 保持连接,读取消息(客户端可能发 ping) for { _, _, err := conn.ReadMessage() if err != nil { if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseNormalClosure) { log.Printf("[MetricsWS] 连接异常关闭: %v", err) } break } } }