diff --git a/agent/collector/diskio.go b/agent/collector/diskio.go new file mode 100644 index 0000000..15c3b1d --- /dev/null +++ b/agent/collector/diskio.go @@ -0,0 +1,121 @@ +// Package collector — 磁盘 IO 指标采集器 +// 通过读取 /proc/diskstats 获取磁盘读写 BPS 和 IOPS。 +// +// 署名 Hermes +package collector + +import ( + "bufio" + "fmt" + "os" + "strconv" + "strings" + "sync" + "time" + + "ezops/agent/model" +) + +// diskIOSample 保存上一次采样，用于计算速率 +type diskIOSample struct { + time time.Time + readSects uint64 + writeSects uint64 + readOps uint64 + writeOps uint64 +} + +// DiskIOCollector 磁盘 IO 采集器 +type DiskIOCollector struct { + mu sync.Mutex + prev *diskIOSample +} + +// NewDiskIOCollector 创建磁盘 IO 采集器 +func NewDiskIOCollector() *DiskIOCollector { + return &DiskIOCollector{} +} + +// Collect 采集磁盘 IO 指标 +func (c *DiskIOCollector) Collect() (model.DiskIOStat, error) { + stat := model.DiskIOStat{} + + rSects, wSects, rOps, wOps, err := readDiskStats() + if err != nil { + return stat, fmt.Errorf("读取磁盘 IO 失败: %w", err) + } + + c.mu.Lock() + defer c.mu.Unlock() + + now := time.Now() + cur := &diskIOSample{ + time: now, + readSects: rSects, + writeSects: wSects, + readOps: rOps, + writeOps: wOps, + } + + if c.prev != nil && now.After(c.prev.time) { + elapsed := now.Sub(c.prev.time).Seconds() + if elapsed > 0 { + sectorSize := uint64(512) // 标准扇区 512 bytes + stat.ReadBPS = float64((cur.readSects-c.prev.readSects)*sectorSize) / elapsed + stat.WriteBPS = float64((cur.writeSects-c.prev.writeSects)*sectorSize) / elapsed + stat.ReadIOPS = float64(cur.readOps-c.prev.readOps) / elapsed + stat.WriteIOPS = float64(cur.writeOps-c.prev.writeOps) / elapsed + } + } + c.prev = cur + + return stat, nil +} + +// readDiskStats 读取 /proc/diskstats，汇总所有物理磁盘的读写数据 +// fields[2]=设备名, fields[3]=读完成次数, fields[5]=读扇区数, fields[7]=写完成次数, fields[9]=写扇区数 +func readDiskStats() (rSects, wSects, rOps, wOps uint64, err error) { + f, err := os.Open("/proc/diskstats") + if err != nil { + return 0, 0, 0, 0, err + } + defer f.Close() + + scanner := bufio.NewScanner(f) + for scanner.Scan() { + line := scanner.Text() + fields := strings.Fields(line) + if len(fields) < 14 { + continue + } + + dev := fields[2] + // 只采集物理磁盘 (sdX, nvmeXn1, vdX)，跳过分区和 loop 设备 + if strings.HasPrefix(dev, "loop") || + (strings.HasPrefix(dev, "sd") && len(dev) > 3) || // sda1, sda2 等分区 + (strings.HasPrefix(dev, "nvme") && strings.Contains(dev, "p")) || // nvme0n1p1 + (strings.HasPrefix(dev, "ram")) || + strings.HasPrefix(dev, "dm-") { + continue + } + + // field 3 = 读完成次数 (rd_ios) + // field 5 = 读扇区数 (rd_sectors) + // field 7 = 写完成次数 (wr_ios) + // field 9 = 写扇区数 (wr_sectors) + if v, e := strconv.ParseUint(fields[3], 10, 64); e == nil { + rOps += v + } + if v, e := strconv.ParseUint(fields[5], 10, 64); e == nil { + rSects += v + } + if v, e := strconv.ParseUint(fields[7], 10, 64); e == nil { + wOps += v + } + if v, e := strconv.ParseUint(fields[9], 10, 64); e == nil { + wSects += v + } + } + + return rSects, wSects, rOps, wOps, scanner.Err() +} diff --git a/agent/collector/network.go b/agent/collector/network.go new file mode 100644 index 0000000..18d7aef --- /dev/null +++ b/agent/collector/network.go @@ -0,0 +1,171 @@ +// Package collector 提供服务器各项系统指标的采集能力。 +// 本文件实现网络连接数、网卡流量等网络指标采集。 +// +// 署名 Hermes (extended from Logan.Li pattern) +package collector + +import ( + "bufio" + "fmt" + "os" + "strconv" + "strings" + "sync" + "time" + + "ezops/agent/model" +) + +// netSample 保存上一次的网络采样数据，用于计算速率 +type netSample struct { + time time.Time + rxBytes uint64 + txBytes uint64 +} + +// NetworkCollector 网络指标采集器 +type NetworkCollector struct { + mu sync.Mutex + prev *netSample +} + +// NewNetworkCollector 创建网络采集器 +func NewNetworkCollector() *NetworkCollector { + return &NetworkCollector{} +} + +// Collect 采集网络指标（并发连接数、TIME_WAIT、网卡 rx/tx B/s） +func (c *NetworkCollector) Collect() (model.NetworkStat, error) { + stat := model.NetworkStat{} + + // 1. 采集 TCP 连接统计 + if tcpStat, err := readSockStat(); err == nil { + stat.TCPEstablished = tcpStat.established + stat.TCPTimeWait = tcpStat.timeWait + stat.TCPTotal = tcpStat.total + } + + // 2. 采集网卡流量（取所有非 lo 网卡合计） + curRx, curTx, nicName, err := readNetDevTotal() + if err != nil { + return stat, fmt.Errorf("读取网卡流量失败: %w", err) + } + stat.NicName = nicName + + c.mu.Lock() + defer c.mu.Unlock() + + now := time.Now() + cur := &netSample{time: now, rxBytes: curRx, txBytes: curTx} + + if c.prev != nil && now.After(c.prev.time) { + elapsed := now.Sub(c.prev.time).Seconds() + if elapsed > 0 && cur.rxBytes >= c.prev.rxBytes && cur.txBytes >= c.prev.txBytes { + stat.NetRxBPS = float64(cur.rxBytes-c.prev.rxBytes) / elapsed + stat.NetTxBPS = float64(cur.txBytes-c.prev.txBytes) / elapsed + } + } + c.prev = cur + + return stat, nil +} + +// tcpSockStat 解析 /proc/net/sockstat 或 /proc/net/tcp 的结果 +type tcpSockStat struct { + established int + timeWait int + total int +} + +// readSockStat 从 /proc/net/sockstat 读取 TCP 连接统计 +func readSockStat() (tcpSockStat, error) { + var result tcpSockStat + + f, err := os.Open("/proc/net/sockstat") + if err != nil { + // Fallback: try /proc/net/sockstat6 + return result, err + } + defer f.Close() + + scanner := bufio.NewScanner(f) + for scanner.Scan() { + line := scanner.Text() + if !strings.HasPrefix(line, "TCP:") { + continue + } + // 格式: "TCP: inuse 123 orphan 0 tw 45 alloc 200 mem 30" + fields := strings.Fields(line) + for i, f := range fields { + switch f { + case "inuse": + if i+1 < len(fields) { + result.established, _ = strconv.Atoi(fields[i+1]) + } + case "tw": + if i+1 < len(fields) { + result.timeWait, _ = strconv.Atoi(fields[i+1]) + } + case "alloc": + if i+1 < len(fields) { + result.total, _ = strconv.Atoi(fields[i+1]) + } + } + } + result.total = result.established // total = established count (inuse) + break + } + + return result, scanner.Err() +} + +// readNetDevTotal 读取 /proc/net/dev，取第一个非 lo 且活跃的物理网卡总流量 +func readNetDevTotal() (rxBytes, txBytes uint64, nicName string, err error) { + f, err := os.Open("/proc/net/dev") + if err != nil { + return 0, 0, "", err + } + defer f.Close() + + scanner := bufio.NewScanner(f) + // Skip header lines (first 2 lines) + lineNum := 0 + for scanner.Scan() { + lineNum++ + if lineNum <= 2 { + continue + } + line := strings.TrimSpace(scanner.Text()) + parts := strings.SplitN(line, ":", 2) + if len(parts) != 2 { + continue + } + name := strings.TrimSpace(parts[0]) + // Skip loopback and virtual interfaces + if name == "lo" || strings.HasPrefix(name, "veth") || strings.HasPrefix(name, "docker") || + strings.HasPrefix(name, "br-") || strings.HasPrefix(name, "virbr") { + continue + } + + fields := strings.Fields(parts[1]) + if len(fields) < 10 { + continue + } + + // fields[0]=rx_bytes, fields[8]=tx_bytes + rb, _ := strconv.ParseUint(fields[0], 10, 64) + tb, _ := strconv.ParseUint(fields[8], 10, 64) + + if rb > 0 || tb > 0 { + nicName = name + rxBytes += rb + txBytes += tb + } + } + + if nicName == "" { + return 0, 0, "eth0", nil + } + + return rxBytes, txBytes, nicName, scanner.Err() +} diff --git a/agent/go.mod b/agent/go.mod index 4012148..40438c6 100644 --- a/agent/go.mod +++ b/agent/go.mod @@ -1,5 +1,5 @@ module ezops/agent -go 1.24 +go 1.22 require golang.org/x/sys v0.28.0 diff --git a/agent/main.go b/agent/main.go index 1be87f3..5bd9974 100644 --- a/agent/main.go +++ b/agent/main.go @@ -64,6 +64,8 @@ func main() { memCollector := collector.NewMemCollector() diskCollector := collector.NewDiskCollector() portCollector := collector.NewPortCollector() + netCollector := collector.NewNetworkCollector() + diskIOCollector := collector.NewDiskIOCollector() rep := reporter.New(*server, *token, logger) // 初始化任务执行器与任务客户端（与 reporter 复用相同的 server/token/hostname） @@ -108,12 +110,27 @@ func main() { report.Ports = ports } + if netStat, err := netCollector.Collect(); err != nil { + logger.Printf("网络采集异常: %v", err) + } else { + report.Network = netStat + } + + if dioStat, err := diskIOCollector.Collect(); err != nil { + logger.Printf("磁盘IO采集异常: %v", err) + } else { + report.DiskIO = dioStat + } + if err := rep.Send(ctx, report); err != nil { logger.Printf("上报失败: %v", err) return } - logger.Printf("上报成功: cpu=%.2f%% mem=%.2f%% disks=%d ports=%d", - report.CPU.UsagePercent, report.Memory.UsagePercent, len(report.Disks), len(report.Ports)) + logger.Printf("上报成功: cpu=%.2f%% mem=%.2f%% disks=%d ports=%d net=%dconn/%dwait netIO=%.0f/%.0fB/s diskIO=%.0f/%.0fB/s", + report.CPU.UsagePercent, report.Memory.UsagePercent, len(report.Disks), len(report.Ports), + report.Network.TCPEstablished, report.Network.TCPTimeWait, + report.Network.NetRxBPS, report.Network.NetTxBPS, + report.DiskIO.ReadBPS, report.DiskIO.WriteBPS) } // pollAndExecute 执行单次任务轮询：拉取待执行任务 → 逐个执行 → 回传结果。 diff --git a/agent/model/model.go b/agent/model/model.go index 20543d4..6f4386f 100644 --- a/agent/model/model.go +++ b/agent/model/model.go @@ -6,12 +6,14 @@ package model // Report 为单次采集周期的完整上报报文 type Report struct { - Hostname string `json:"hostname"` // 主机名 - Timestamp string `json:"timestamp"` // RFC3339 格式的 UTC 时间戳 - CPU CPUStat `json:"cpu"` // CPU 指标 - Memory MemoryStat `json:"memory"` // 内存指标 - Disks []DiskStat `json:"disks"` // 磁盘指标列表 - Ports []PortStat `json:"ports"` // 监听端口列表 + Hostname string `json:"hostname"` // 主机名 + Timestamp string `json:"timestamp"` // RFC3339 格式的 UTC 时间戳 + CPU CPUStat `json:"cpu"` // CPU 指标 + Memory MemoryStat `json:"memory"` // 内存指标 + Disks []DiskStat `json:"disks"` // 磁盘指标列表 + Ports []PortStat `json:"ports"` // 监听端口列表 + Network NetworkStat `json:"network"` // 网络指标 + DiskIO DiskIOStat `json:"disk_io"` // 磁盘 IO 指标 } // CPUStat CPU 相关指标 diff --git a/agent/model/model_ext.go b/agent/model/model_ext.go new file mode 100644 index 0000000..c757b44 --- /dev/null +++ b/agent/model/model_ext.go @@ -0,0 +1,28 @@ +package model + +// NetworkStat 网络相关指标 +type NetworkStat struct { + TCPEstablished int `json:"tcp_established"` // ESTABLISHED 连接数 + TCPTimeWait int `json:"tcp_time_wait"` // TIME_WAIT 连接数 + TCPTotal int `json:"tcp_total"` // TCP 总连接数 + NicName string `json:"nic_name"` // 主网卡名 + NetRxBPS float64 `json:"net_rx_bps"` // 网卡接收速率 (bytes/s) + NetTxBPS float64 `json:"net_tx_bps"` // 网卡发送速率 (bytes/s) +} + +// DiskIOStat 磁盘 IO 相关指标 +type DiskIOStat struct { + ReadBPS float64 `json:"read_bps"` // 读取速率 (bytes/s) + WriteBPS float64 `json:"write_bps"` // 写入速率 (bytes/s) + ReadIOPS float64 `json:"read_iops"` // 读取 IOPS (count/s) + WriteIOPS float64 `json:"write_iops"` // 写入 IOPS (count/s) +} + +// SystemStat 系统信息（启动时采集一次） +type SystemStat struct { + OSVersion string `json:"os_version"` // OS 版本 (如 "Ubuntu 22.04") + Kernel string `json:"kernel"` // 内核版本 + Uptime int64 `json:"uptime"` // 运行秒数 + UlimitNofile int64 `json:"ulimit_nofile"` // ulimit -n (最大文件描述符) + UlimitNproc int64 `json:"ulimit_nproc"` // ulimit -u (最大进程数) +} diff --git a/backend/cmd/main.go b/backend/cmd/main.go index be8057f..941d7fe 100644 --- a/backend/cmd/main.go +++ b/backend/cmd/main.go @@ -125,7 +125,7 @@ func main() { // 初始化缓存（Redka 内嵌 / Redis 外部）// Logan.Li cache, err := ezredis.New(ezredis.Config{ - Driver: "redka", // 默认内嵌 Redka，零依赖 + Driver: cfg.Redis.Driver, // 从配置读取 driver Host: cfg.Redis.Host, Port: cfg.Redis.Port, Password: cfg.Redis.Password, @@ -136,7 +136,7 @@ func main() { log.Warn("缓存初始化失败，继续启动", zap.Error(err)) cache = nil } else { - log.Info("缓存已就绪", zap.String("driver", "redka")) + log.Info("缓存已就绪", zap.String("driver", cfg.Redis.Driver)) defer cache.Close() } @@ -222,6 +222,7 @@ func loadConfig() *config.Config { viper.SetDefault("redis.host", "localhost") viper.SetDefault("redis.port", 6379) + viper.SetDefault("redis.driver", "redka") viper.SetDefault("redis.password", "") viper.SetDefault("redis.db", 0) diff --git a/backend/internal/api/handlers/agent.go b/backend/internal/api/handlers/agent.go index c6df9c2..5f6bd60 100644 --- a/backend/internal/api/handlers/agent.go +++ b/backend/internal/api/handlers/agent.go @@ -19,12 +19,13 @@ import ( // AgentHandler 处理 Agent 上报数据 // Logan.Li type AgentHandler struct { - db *gorm.DB - cache ezredis.Cache // 可选 Redis 缓存（用于热数据存储） + db *gorm.DB + cache ezredis.Cache // 可选 Redis 缓存（用于热数据存储） + hubMetrics *MetricsHub // WebSocket 广播 metrics 更新 } -func NewAgentHandler(db *gorm.DB, cache ezredis.Cache) *AgentHandler { - return &AgentHandler{db: db, cache: cache} +func NewAgentHandler(db *gorm.DB, cache ezredis.Cache, hub *MetricsHub) *AgentHandler { + return &AgentHandler{db: db, cache: cache, hubMetrics: hub} } // ReceiveReport 接收 Agent 上报的指标数据 @@ -47,6 +48,8 @@ func (h *AgentHandler) ReceiveReport(c *gin.Context) { // 序列化磁盘和端口信息 diskJSON, _ := json.Marshal(req.Disks) portJSON, _ := json.Marshal(req.Ports) + netJSON, _ := json.Marshal(req.Network) + dioJSON, _ := json.Marshal(req.DiskIO) // 存储到数据库 report := models.AgentReport{ @@ -61,6 +64,8 @@ func (h *AgentHandler) ReceiveReport(c *gin.Context) { MemUsage: req.Memory.UsagePercent, DiskInfo: string(diskJSON), PortInfo: string(portJSON), + NetInfo: string(netJSON), + DiskIOInfo: string(dioJSON), ReportTime: reportTime, } @@ -69,6 +74,35 @@ func (h *AgentHandler) ReceiveReport(c *gin.Context) { return } + // 同步更新 servers 表的实时指标（通过 hostname 匹配） + diskAvg := avgDiskUsage(req.Disks) + result := h.db.Model(&models.Server{}). + Where("hostname = ? OR ip = ?", req.Hostname, req.Hostname). + Updates(map[string]interface{}{ + "cpu_usage": req.CPU.UsagePercent, + "memory_usage": req.Memory.UsagePercent, + "disk_usage": diskAvg, + "status": "running", + }) + + // WebSocket 广播：向所有前端客户端推送该主机的最新指标 // Logan.Li + if h.hubMetrics != nil && result.RowsAffected > 0 { + h.hubMetrics.Broadcast(MetricsUpdate{ + Type: "metrics_update", + Hostname: req.Hostname, + CPU: req.CPU.UsagePercent, + Memory: req.Memory.UsagePercent, + Disk: diskAvg, + Status: "running", + TCPEstablished: req.Network.TCPEstablished, + TCPTimeWait: req.Network.TCPTimeWait, + NetRxBPS: req.Network.NetRxBPS, + NetTxBPS: req.Network.NetTxBPS, + DiskReadBPS: req.DiskIO.ReadBPS, + DiskWriteBPS: req.DiskIO.WriteBPS, + }) + } + // 写入 Redis 有序集合（热数据，保留7天） if h.cache != nil { timestamp := float64(reportTime.Unix()) @@ -169,6 +203,7 @@ func (h *AgentHandler) AggregateMetrics(c *gin.Context) { duration time.Duration }{ {"1m", 1 * time.Minute}, + {"5m", 5 * time.Minute}, {"3m", 3 * time.Minute}, {"1h", 1 * time.Hour}, {"6h", 6 * time.Hour}, @@ -283,11 +318,20 @@ func (h *AgentHandler) GetKline(c *gin.Context) { } var klines []models.MetricsKline - h.db.Where("hostname = ? AND metric = ? AND interval = ? AND bucket_time >= ? AND bucket_time <= ?", + result := h.db.Where("hostname = ? AND metric = ? AND `interval` = ? AND bucket_time >= ? AND bucket_time <= ?", req.Hostname, req.Metric, req.Interval, from, to). Order("bucket_time ASC"). Find(&klines) + if result.Error != nil { + c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "查询失败: " + result.Error.Error()}) + return + } + + if klines == nil { + klines = []models.MetricsKline{} + } + c.JSON(http.StatusOK, models.Response{ Code: 200, Message: "success", @@ -305,3 +349,15 @@ func parseMemberValue(member string) float64 { v, _ := strconv.ParseFloat(member[:idx], 64) return v } + +// avgDiskUsage 计算磁盘平均使用率 +func avgDiskUsage(disks []models.DiskMetric) float64 { + if len(disks) == 0 { + return 0 + } + var sum float64 + for _, d := range disks { + sum += d.UsagePercent + } + return math.Round(sum/float64(len(disks))*100) / 100 +} diff --git a/backend/internal/api/handlers/metrics_ws.go b/backend/internal/api/handlers/metrics_ws.go new file mode 100644 index 0000000..1602b61 --- /dev/null +++ b/backend/internal/api/handlers/metrics_ws.go @@ -0,0 +1,120 @@ +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 + } + } +} diff --git a/backend/internal/api/routes/routes.go b/backend/internal/api/routes/routes.go index af16739..9a8bcfa 100644 --- a/backend/internal/api/routes/routes.go +++ b/backend/internal/api/routes/routes.go @@ -59,6 +59,11 @@ func Register(r *gin.Engine, db *gorm.DB, log *zap.Logger, cfg *config.Config, c webShellHandler := handlers.NewWebShellHandler(db) r.GET("/ws/terminal", middleware.JWTAuth(&cfg.JWT), webShellHandler.Terminal) + // WebSocket 实时指标推送（JWT 鉴权） // Logan.Li + metricsHub := handlers.NewMetricsHub() + metricsWSHandler := handlers.NewMetricsWSHandler(metricsHub) + r.GET("/ws/metrics", middleware.JWTAuth(&cfg.JWT), metricsWSHandler.ServeWS) + // 公开路由 authHandler := handlers.NewUserHandler(db, cfg) api.POST("/auth/login", authHandler.Login) @@ -414,7 +419,7 @@ func Register(r *gin.Engine, db *gorm.DB, log *zap.Logger, cfg *config.Config, c // Agent 上报路由（无需 JWT） redisCache, _ := cache.(ezredis.Cache) - agentHandler := handlers.NewAgentHandler(db, redisCache) + agentHandler := handlers.NewAgentHandler(db, redisCache, metricsHub) taskHandler := handlers.NewAgentTaskHandler(db) agentGroup := api.Group("/agent") { diff --git a/backend/internal/config/config.go b/backend/internal/config/config.go index b2bc5d2..381741c 100644 --- a/backend/internal/config/config.go +++ b/backend/internal/config/config.go @@ -39,6 +39,7 @@ type JWTConfig struct { // RedisConfig Redis配置 type RedisConfig struct { + Driver string `mapstructure:"driver"` // redka (内嵌) / redis (外部) Host string `mapstructure:"host"` Port int `mapstructure:"port"` Password string `mapstructure:"password"` @@ -72,6 +73,7 @@ func Load() *Config { viper.SetDefault("jwt.expire_hour", 24) viper.SetDefault("redis.host", "localhost") viper.SetDefault("redis.port", 6379) + viper.SetDefault("redis.driver", "redka") viper.SetDefault("websocket.read_buffer_size", 1024) viper.SetDefault("websocket.write_buffer_size", 1024) diff --git a/backend/internal/models/agent.go b/backend/internal/models/agent.go index 4f7eeae..7b5dcaa 100644 --- a/backend/internal/models/agent.go +++ b/backend/internal/models/agent.go @@ -20,6 +20,8 @@ type AgentReport struct { MemUsage float64 `gorm:"type:decimal(6,2)" json:"mem_usage"` // 内存使用率 % DiskInfo string `gorm:"type:text" json:"disk_info"` // 磁盘信息 JSON PortInfo string `gorm:"type:text" json:"port_info"` // 端口信息 JSON + NetInfo string `gorm:"type:text" json:"net_info"` // 网络指标 JSON + DiskIOInfo string `gorm:"type:text" json:"disk_io_info"` // 磁盘 IO JSON ReportTime time.Time `json:"report_time"` // 上报时间 CreatedAt time.Time `json:"created_at"` UpdatedAt time.Time `json:"updated_at"` @@ -32,12 +34,14 @@ func (AgentReport) TableName() string { // AgentReportRequest Agent 上报请求体 // Logan.Li type AgentReportRequest struct { - Hostname string `json:"hostname"` - Timestamp string `json:"timestamp"` - CPU CPUMetric `json:"cpu"` - Memory MemMetric `json:"memory"` + Hostname string `json:"hostname" binding:"required"` + Timestamp string `json:"timestamp"` + CPU CPUMetric `json:"cpu"` + Memory MemMetric `json:"memory"` Disks []DiskMetric `json:"disks"` Ports []PortMetric `json:"ports"` + Network NetworkStat `json:"network"` // 网络指标 + DiskIO DiskIOStat `json:"disk_io"` // 磁盘 IO } type CPUMetric struct { @@ -69,3 +73,21 @@ type PortMetric struct { Process string `json:"process"` PID int `json:"pid"` } + +// NetworkStat 网络指标（Agent 上报） +type NetworkStat struct { + TCPEstablished int `json:"tcp_established"` + TCPTimeWait int `json:"tcp_time_wait"` + TCPTotal int `json:"tcp_total"` + NicName string `json:"nic_name"` + NetRxBPS float64 `json:"net_rx_bps"` + NetTxBPS float64 `json:"net_tx_bps"` +} + +// DiskIOStat 磁盘 IO 指标（Agent 上报） +type DiskIOStat struct { + ReadBPS float64 `json:"read_bps"` + WriteBPS float64 `json:"write_bps"` + ReadIOPS float64 `json:"read_iops"` + WriteIOPS float64 `json:"write_iops"` +} diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index f2e4099..79197f4 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -31,8 +31,7 @@ import { AIPredictiveAnalytics } from '@/components/aiops/AIPredictiveAnalytics' import AlertCenter from '@/components/monitor/alert-center/AlertCenter'; import TraceManager from '@/components/monitor/trace/TraceManager'; import GrafanaDashboard from '@/components/monitor/dashboard/GrafanaDashboard'; -import { AgentMonitor } from '@/components/agent/AgentMonitor'; -import { AgentAlerts } from '@/components/agent/AgentAlerts'; +import { MonitorCenter } from '@/pages/MonitorCenter'; const queryClient = new QueryClient({ defaultOptions: { @@ -59,7 +58,8 @@ function App() { } /> } /> } /> - } /> + {/* 资源监控（Agent 指标 + 中间件 + SNMP + 拨测） */} + } /> } /> } /> } /> @@ -73,7 +73,6 @@ function App() { } /> } /> } /> - } /> } /> } /> } /> diff --git a/frontend/src/components/layout/Sidebar.tsx b/frontend/src/components/layout/Sidebar.tsx index 95b07d5..74f2235 100644 --- a/frontend/src/components/layout/Sidebar.tsx +++ b/frontend/src/components/layout/Sidebar.tsx @@ -59,7 +59,6 @@ const menuItems: MenuItem[] = [ { id: 'regions', name: '区域管理', icon: MapPin, path: '/regions' }, { id: 'platforms', name: '云平台', icon: CloudCog, path: '/platforms' }, { id: 'cloud-billing', name: '云账单', icon: BarChart3, path: '/cloud-billing' }, - { id: 'agent-monitor', name: 'Agent监控', icon: Activity, path: '/agent-monitor' }, ], }, { @@ -84,14 +83,14 @@ const menuItems: MenuItem[] = [ }, { id: 'monitor', - name: '监控告警', + name: '监控中心', icon: Bell, children: [ + { id: 'monitor-center', name: '资源监控', icon: Activity, path: '/monitor-center' }, { id: 'metrics', name: '指标监控', icon: Cpu, path: '/metrics' }, - { id: 'alert-center', name: '告警中心', icon: Bell, path: '/alert-center' }, - { id: 'trace', name: '链路追踪', icon: GitBranch, path: '/trace' }, { id: 'grafana', name: '监控大屏', icon: Monitor, path: '/grafana' }, - { id: 'agent-alerts', name: 'Agent告警', icon: AlertTriangle, path: '/agent-alerts' }, + { id: 'trace', name: '链路追踪', icon: GitBranch, path: '/trace' }, + { id: 'alert-center', name: '告警服务', icon: Bell, path: '/alert-center' }, ], }, { diff --git a/frontend/src/components/server/ServerManager.tsx b/frontend/src/components/server/ServerManager.tsx index fd8dd2a..ad47e5c 100644 --- a/frontend/src/components/server/ServerManager.tsx +++ b/frontend/src/components/server/ServerManager.tsx @@ -1,4 +1,4 @@ -import { useState, useEffect } from 'react'; +import { useState, useEffect, useRef } from 'react'; import { cn } from '@/lib/utils'; import { Card, CardContent } from '@/components/ui/card'; import { Button } from '@/components/ui/button'; @@ -168,6 +168,82 @@ export function ServerManager() { loadData(); }, []); + // WebSocket 实时接收 Agent 指标推送 // Logan.Li + const wsRef = useRef(null); + const reconnectTimerRef = useRef | null>(null); + + useEffect(() => { + const token = useUserStore.getState().token; + if (!token) return; + + const protocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:'; + const wsUrl = `${protocol}//${window.location.host}/ws/metrics?token=${token}`; + + const connect = () => { + if (wsRef.current?.readyState === WebSocket.OPEN) return; + + const ws = new WebSocket(wsUrl); + wsRef.current = ws; + + ws.onopen = () => { + console.log('[MetricsWS] 已连接，实时接收指标推送'); + }; + + ws.onmessage = (event) => { + try { + const msg = JSON.parse(event.data); + if (msg.type === 'metrics_update') { + setServers((prev) => + prev.map((s) => { + // 通过 hostname 匹配（Agent 上报时使用 hostname） + if (s.hostname === msg.hostname) { + return { + ...s, + cpu_usage: msg.cpu_usage, + memory_usage: msg.memory_usage, + disk_usage: msg.disk_usage, + status: msg.status, + tcp_established: msg.tcp_established, + tcp_time_wait: msg.tcp_time_wait, + net_rx_bps: msg.net_rx_bps, + net_tx_bps: msg.net_tx_bps, + disk_read_bps: msg.disk_read_bps, + disk_write_bps: msg.disk_write_bps, + }; + } + return s; + }) + ); + } + } catch (e) { + // 忽略非 JSON 消息 + } + }; + + ws.onclose = () => { + console.log('[MetricsWS] 连接断开，3 秒后重连'); + wsRef.current = null; + reconnectTimerRef.current = setTimeout(connect, 3000); + }; + + ws.onerror = (err) => { + console.error('[MetricsWS] 连接错误:', err); + ws.close(); + }; + }; + + connect(); + + return () => { + if (reconnectTimerRef.current) clearTimeout(reconnectTimerRef.current); + if (wsRef.current) { + wsRef.current.onclose = null; // 阻止自动重连 + wsRef.current.close(); + wsRef.current = null; + } + }; + }, []); + const filteredServers = servers.filter((server) => { const matchesSearch = (server.hostname || '').toLowerCase().includes(searchQuery.toLowerCase()) || diff --git a/frontend/src/components/server/ServerTableRow.tsx b/frontend/src/components/server/ServerTableRow.tsx index 474ce31..bd2d824 100644 --- a/frontend/src/components/server/ServerTableRow.tsx +++ b/frontend/src/components/server/ServerTableRow.tsx @@ -12,7 +12,7 @@ import { DropdownMenuTrigger, } from '@/components/ui/dropdown-menu'; import { TableCell, TableRow } from '@/components/ui/table'; -import { Cpu, Edit, Loader2, MoreHorizontal, Server as ServerIcon, Terminal, Trash2, CheckCircle, XCircle } from "lucide-react"; +import { Cpu, Edit, Loader2, MoreHorizontal, Server as ServerIcon, Terminal, Trash2, CheckCircle, XCircle, Network, HardDrive } from 'lucide-react'; import { getStatusBadge, getTags, getUsageColor } from './serverHelpers'; interface ServerTableRowProps { @@ -42,6 +42,19 @@ export function ServerTableRow({ const memoryUsage = Number(server.memory_usage || 0); const tags = getTags(server); const memGB = server.memory ? Math.round(server.memory / 1024) : 0; + const tcpEst = server.tcp_established || 0; + const tcpTw = server.tcp_time_wait || 0; + const rxBps = server.net_rx_bps || 0; + const txBps = server.net_tx_bps || 0; + const readBps = server.disk_read_bps || 0; + const writeBps = server.disk_write_bps || 0; + + // Format bytes/s to human readable + const fmtBps = (bps: number) => { + if (bps >= 1_000_000) return (bps / 1_000_000).toFixed(1) + 'MB/s'; + if (bps >= 1_000) return (bps / 1_000).toFixed(1) + 'KB/s'; + return bps.toFixed(0) + 'B/s'; + }; return ( @@ -107,6 +120,15 @@ export function ServerTableRow({ {memoryUsage.toFixed(1)}% +
+ + {tcpEst}连接 + {tcpTw > 0 && ({tcpTw}TW)} +
+
+ + 读{fmtBps(readBps)}/写{fmtBps(writeBps)} +
diff --git a/frontend/src/pages/HostMonitor.tsx b/frontend/src/pages/HostMonitor.tsx new file mode 100644 index 0000000..60d1a10 --- /dev/null +++ b/frontend/src/pages/HostMonitor.tsx @@ -0,0 +1,94 @@ +// 主机监控 — Agent 采集指标展示 (CPU/内存/磁盘/网络/磁盘IO) +import { Card, CardContent } from '@/components/ui/card'; +import { Cpu, HardDrive, Network, Activity, Server } from 'lucide-react'; + +export function HostMonitor() { + return ( +
+ {/* Overview Cards */} +
+ + +
+
+ +
+
+

CPU 使用率

+

--

+
+
+ + + + +
+
+ +
+
+

内存使用率

+

--

+
+
+ + + + +
+
+ +
+
+

TCP 连接数

+

--

+
+
+ + + + +
+
+ +
+
+

磁盘 IO

+

--

+
+
+ + +
+ + {/* Detail Sections */} +
+ {/* CPU/Memory Trends */} + + +
+ +

CPU / 内存 趋势

+
+
+ 趋势图表开发中... +
+ + + + {/* Network */} + + +
+ +

网络流量

+
+
+ 网络流量图表开发中... +
+ + +
+
+ ); +} diff --git a/frontend/src/pages/MiddlewareMonitor.tsx b/frontend/src/pages/MiddlewareMonitor.tsx new file mode 100644 index 0000000..f575769 --- /dev/null +++ b/frontend/src/pages/MiddlewareMonitor.tsx @@ -0,0 +1,47 @@ +// 中间件监控 — Redis / MySQL 指标采集（需配置认证信息） +import { Card, CardContent } from '@/components/ui/card'; +import { Database, Server } from 'lucide-react'; + +export function MiddlewareMonitor() { + return ( +
+ {/* Redis */} + + +
+ +

Redis 监控

+
+

需在 Agent 配置 Redis 连接地址和密码后启用采集。

+
+ {['连接数', '内存', '命中率', 'QPS'].map((label) => ( +
+

{label}

+

--

+
+ ))} +
+ + + + {/* MySQL */} + + +
+ +

MySQL 监控

+
+

需在 Agent 配置 MySQL 连接地址、用户名和密码后启用采集。

+
+ {['连接数', 'QPS', '慢查询', '缓冲池命中率'].map((label) => ( +
+

{label}

+

--

+
+ ))} +
+ + +
+ ); +} diff --git a/frontend/src/pages/MonitorCenter.tsx b/frontend/src/pages/MonitorCenter.tsx new file mode 100644 index 0000000..9955a9d --- /dev/null +++ b/frontend/src/pages/MonitorCenter.tsx @@ -0,0 +1,69 @@ +// 监控中心 — 4个监控标签页的容器 +import { useState } from 'react'; +import { cn } from '@/lib/utils'; +import { Cpu, Database, Radio, Globe } from 'lucide-react'; +import { HostMonitor } from './HostMonitor'; +import { MiddlewareMonitor } from './MiddlewareMonitor'; +import { SNMPMonitor } from './SNMPMonitor'; +import { SiteMonitor } from './SiteMonitor'; + +type TabId = 'host' | 'middleware' | 'snmp' | 'site'; + +interface Tab { + id: TabId; + name: string; + icon: React.ElementType; + description: string; +} + +const tabs: Tab[] = [ + { id: 'host', name: '主机监控', icon: Cpu, description: 'CPU / 内存 / 磁盘 / 网络 / 磁盘IO' }, + { id: 'middleware', name: '中间件监控', icon: Database, description: 'Redis / MySQL 指标采集' }, + { id: 'snmp', name: 'SNMP 监控', icon: Radio, description: '网络设备 / 服务器硬件 (OID 模板)' }, + { id: 'site', name: '站点拨测', icon: Globe, description: 'HTTP / PING / DNS / TCP / WS 拨测' }, +]; + +export function MonitorCenter() { + const [activeTab, setActiveTab] = useState('host'); + + return ( +
+
+

监控中心

+

全面的基础设施与应用监控

+
+ + {/* Tab Bar */} +
+ {tabs.map((tab) => ( + + ))} +
+ + {/* Tab Description */} +

+ {tabs.find(t => t.id === activeTab)?.description} +

+ + {/* Tab Content */} +
+ {activeTab === 'host' && } + {activeTab === 'middleware' && } + {activeTab === 'snmp' && } + {activeTab === 'site' && } +
+
+ ); +} diff --git a/frontend/src/pages/SNMPMonitor.tsx b/frontend/src/pages/SNMPMonitor.tsx new file mode 100644 index 0000000..42db342 --- /dev/null +++ b/frontend/src/pages/SNMPMonitor.tsx @@ -0,0 +1,72 @@ +// SNMP 监控 — 网络设备 / 服务器硬件 (OID 模板采集) +import { Card, CardContent } from '@/components/ui/card'; +import { Radio, Router, HardDrive } from 'lucide-react'; + +export function SNMPMonitor() { + return ( +
+ {/* Device Types */} +
+ + +
+ +

网络设备

+
+

交换机 / 路由器 / 防火墙

+
+
+ 设备数 + -- +
+
+ 在线 + -- +
+
+ + + + + +
+ +

服务器硬件

+
+

iLO / iDRAC / IPMI

+
+
+ 设备数 + -- +
+
+ 异常 + -- +
+
+ + +
+ + {/* Template Info */} + + +
+ +

OID 监控模板

+
+

+ 支持 Zabbix 兼容的 SNMP OID 模板（XML 格式），内置思科/H3C/华为交换机模板、服务器硬件监控模板。 +

+
+ {['思科交换机', 'H3C S5000', '华为交换机', 'iLO/Dell'].map((t) => ( +
+ {t} +
+ ))} +
+ + +
+ ); +} diff --git a/frontend/src/pages/SiteMonitor.tsx b/frontend/src/pages/SiteMonitor.tsx new file mode 100644 index 0000000..df3363f --- /dev/null +++ b/frontend/src/pages/SiteMonitor.tsx @@ -0,0 +1,52 @@ +// 站点拨测 — HTTP / PING / DNS / TCP / WebSocket 等协议主动监控 +import { Card, CardContent } from '@/components/ui/card'; +import { Globe, Wifi, Server, Mail, ArrowUpDown } from 'lucide-react'; + +const probeTypes = [ + { icon: Globe, name: 'HTTP(S)', color: 'text-blue-400', desc: 'URL 状态码 & 响应时间' }, + { icon: ArrowUpDown, name: 'PING', color: 'text-emerald-400', desc: 'ICMP 连通性 & 延迟' }, + { icon: Wifi, name: 'WebSocket', color: 'text-purple-400', desc: 'WS/WSS 连接心跳' }, + { icon: Globe, name: 'DNS', color: 'text-cyan-400', desc: '域名解析延迟 & 结果验证' }, + { icon: Server, name: 'TCP', color: 'text-orange-400', desc: '端口连通性检测' }, + { icon: Server, name: 'UDP', color: 'text-yellow-400', desc: '端口连通性检测' }, + { icon: Mail, name: 'SMTP', color: 'text-pink-400', desc: '邮件服务连通性' }, + { icon: Mail, name: 'POP3', color: 'text-indigo-400', desc: '邮件接收服务' }, + { icon: Globe, name: 'FTP', color: 'text-teal-400', desc: '文件传输服务' }, +]; + +export function SiteMonitor() { + return ( +
+ {/* Probe Type Grid */} +
+ {probeTypes.map((p) => ( + + + +

{p.name}

+

{p.desc}

+ + + ))} +
+ + {/* Summary */} + + +
+ +

拨测任务概览

+
+
+ {['总任务数', '正常', '异常', '可用率'].map((label) => ( +
+

{label}

+

--

+
+ ))} +
+ + +
+ ); +}