feat: Agent采集uptime, 后端持久化kernel/uptime/diskIO, 前端磁盘BPS+IOPS展示

- Agent: system.go 添加 readUptime(), SystemStat 携带 Uptime
- Backend: Server 模型新增 Kernel 字段, ReceiveReport 持久化 kernel+uptime
- Backend: /agent/latest 返回 kernel/uptime/port_info/disk_io_info/disk_total
- Frontend: 磁盘监控改为 BPS+IOPS (4张卡片), 系统信息修复内核/运行时长/监听端口/磁盘容量
- 图表单位从 MB/s 改为 KB/s, 添加 formatBPS 工具函数
This commit is contained in:
attacker committed 2026-06-19 01:10:00 +00:00
1 parent 31c2e338a3
commit 5949ec5d0c
48 files changed
+2182 -956

No files matched your search

+126 -55
View File
@@ -5,6 +5,7 @@ import (
"fmt"
"math"
"net/http"
"sort"
"strconv"
"strings"
"time"
@@ -77,12 +78,14 @@ func (h *AgentHandler) ReceiveReport(c *gin.Context) {
// 同步更新 servers 表的实时指标(三级匹配:hostname → client IP → hostname LIKE)
diskAvg := avgDiskUsage(req.Disks)
// 计算硬件规格
// 计算硬件规格(四舍五入到整数,避免 31.34→31GB 的显示偏差)
totalDiskGB := 0.0
for _, d := range req.Disks {
totalDiskGB += d.TotalGB
}
memMB := int64(req.Memory.TotalGB * 1024) // GB → MB
memGB := math.Ceil(req.Memory.TotalGB) // 31.34 → 32, 62.79 → 63(物理内存向上取整)
memMB := int64(memGB * 1024) // GB → MB
diskGB := int64(totalDiskGB + 0.5) // 39.99 → 40
clientIP := c.ClientIP()
@@ -92,8 +95,8 @@ func (h *AgentHandler) ReceiveReport(c *gin.Context) {
matchIP = clientIP
}
// 直接 SQL 更新,避免 GORM soft delete 自动添加 deleted_at 条件干扰
updateSQL := `UPDATE servers SET
// 直接 SQL 更新 asset_servers 监控字段
updateSQL := `UPDATE asset_servers SET
cpu_usage = ?, memory_usage = ?, disk_usage = ?,
cpu = ?, memory = ?, disk = ?,
status = ?, agent_status = ?, agent_version = ?,
@@ -101,14 +104,24 @@ func (h *AgentHandler) ReceiveReport(c *gin.Context) {
updated_at = NOW()`
args := []interface{}{
req.CPU.UsagePercent, req.Memory.UsagePercent, diskAvg,
req.CPU.CoreCount, memMB, int64(totalDiskGB),
req.CPU.CoreCount, memMB, diskGB,
"running", "online", req.AgentVersion,
// CASE WHEN hostname != ? AND ip = ? THEN ? ELSE hostname END
req.Hostname, matchIP, req.Hostname,
}
if req.System != nil && req.System.OSVersion != "" {
updateSQL += ", os = ?"
args = append(args, req.System.OSVersion)
if req.System != nil {
if req.System.OSVersion != "" {
updateSQL += ", os = ?"
args = append(args, req.System.OSVersion)
}
if req.System.Kernel != "" {
updateSQL += ", kernel = ?"
args = append(args, req.System.Kernel)
}
if req.System.Uptime > 0 {
updateSQL += ", uptime = ?"
args = append(args, req.System.Uptime)
}
}
updateSQL += " WHERE hostname = ? OR ip = ?"
args = append(args, req.Hostname, matchIP)
@@ -122,9 +135,44 @@ func (h *AgentHandler) ReceiveReport(c *gin.Context) {
result = h.db.Exec(updateSQL2, args2...)
}
// 如果还是没匹配到,记录日志(Agent 上报了,但服务器列表里没有对应条目)
// 如果还是没匹配到 —— 自动注册新服务器(UPSERT 防重复)
if result.RowsAffected == 0 {
fmt.Printf("[agent] 上报主机 %q (IP: %s) 未匹配任何 servers 条目,指标仅写入 agent_reports 表\n", req.Hostname, matchIP)
osName := ""
if req.System != nil {
osName = req.System.OSVersion
}
newServer := models.Server{
Hostname: req.Hostname,
IP: matchIP,
Port: 22,
Username: "root",
AuthType: "password",
OS: osName,
CPU: req.CPU.CoreCount,
Memory: memMB,
Disk: diskGB,
Status: "running",
AgentStatus: "online",
AgentVersion: req.AgentVersion,
CPUUsage: req.CPU.UsagePercent,
MemoryUsage: req.Memory.UsagePercent,
DiskUsage: diskAvg,
}
// ON DUPLICATE KEY UPDATE — 若 hostname 已存在则更新字段(防重复注册)
if err := h.db.Clauses(clause.OnConflict{
Columns: []clause.Column{{Name: "hostname"}},
DoUpdates: clause.AssignmentColumns([]string{
"ip", "os", "cpu", "memory", "disk",
"status", "agent_status", "agent_version",
"cpu_usage", "memory_usage", "disk_usage",
"updated_at",
}),
}).Create(&newServer).Error; err != nil {
fmt.Printf("[agent] 自动注册主机 %q (IP: %s) 失败: %v\n", req.Hostname, matchIP, err)
} else {
fmt.Printf("[agent] 自动注册新主机: %q (IP: %s, CPU=%d核, 内存=%dMB, 磁盘=%dGB, OS=%q)\n",
req.Hostname, matchIP, req.CPU.CoreCount, memMB, diskGB, osName)
}
}
// WebSocket 广播:向所有前端客户端推送该主机的最新指标 // Logan.Li
@@ -183,38 +231,51 @@ func (h *AgentHandler) writeHotMetric(c *gin.Context, key string, value float64,
_ = h.cache.ZRemRangeByScore(ctx, key, "-inf", fmt.Sprintf("%.0f", cutoff))
}
// GetLatestReports 获取各主机最新上报(仅 3 分钟内活跃的 hostname)
// GetLatestReports 获取所有服务器的 Agent 最新状态(含未安装 Agent 的主机)
// GET /api/v1/agent/latest
func (h *AgentHandler) GetLatestReports(c *gin.Context) {
var results []models.AgentLatestResponse
threeMinAgo := time.Now().Add(-3 * time.Minute)
// 以 servers 表为主,LEFT JOIN 最近 3 分钟内的 Agent 上报
// 未装 Agent 的主机也会显示(agent_status='not_installed'),打破"不显示→无法安装"的循环
h.db.Raw(`
SELECT
a.hostname,
s.id as server_id,
s.hostname,
COALESCE(s.ip, '') as ip,
a.cpu_usage,
a.cpu_cores,
a.mem_usage,
a.mem_total,
a.mem_used,
COALESCE(a.cpu_usage, 0) as cpu_usage,
COALESCE(a.cpu_cores, 0) as cpu_cores,
COALESCE(a.mem_usage, 0) as mem_usage,
COALESCE(a.mem_total, 0) as mem_total,
COALESCE(a.mem_used, 0) as mem_used,
COALESCE(s.disk_usage, 0) as disk_usage,
a.disk_info,
a.port_info,
a.net_info,
a.disk_io_info,
a.report_time,
COALESCE(s.agent_version, '') as agent_version,
COALESCE(s.agent_status, 'offline') as agent_status,
CASE
WHEN a.hostname IS NOT NULL THEN COALESCE(s.agent_status, 'online')
ELSE 'not_installed'
END as agent_status,
COALESCE(s.os, '') as os,
COALESCE(r.name, '') as region
FROM agent_reports a
LEFT JOIN servers s ON s.hostname = a.hostname
LEFT JOIN regions r ON r.id = s.region_id
WHERE a.id IN (
SELECT MAX(id) FROM agent_reports
WHERE report_time >= ?
GROUP BY hostname
)
AND s.id IS NOT NULL -- 只显示匹配 servers 表的主机(首次上线后自动同步 hostname)
ORDER BY a.report_time DESC
COALESCE(s.kernel, '') as kernel,
COALESCE(s.uptime, 0) as uptime,
COALESCE(s.disk, 0) as disk_total,
COALESCE(r.name, '') as region,
CASE WHEN s.password != '' OR s.ssh_key != '' THEN true ELSE false END as has_auth
FROM asset_servers s
LEFT JOIN (
SELECT * FROM monitor_agent_reports
WHERE id IN (
SELECT MAX(id) FROM monitor_agent_reports
WHERE report_time >= ?
GROUP BY hostname
)
) a ON a.hostname = s.hostname
LEFT JOIN asset_regions r ON r.id = s.region_id
ORDER BY a.report_time DESC, s.id ASC
`, threeMinAgo).Scan(&results)
if results == nil {
@@ -253,7 +314,7 @@ func (h *AgentHandler) GetHostHistory(c *gin.Context) {
// AggregateMetrics 聚合 Redis 热数据到 MySQL klines
// POST /api/v1/agent/aggregate
func (h *AgentHandler) AggregateMetrics(c *gin.Context) {
var req models.AggregateKlineReq
var req models.AggregateReq
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "请求格式错误: " + err.Error()})
return
@@ -303,37 +364,47 @@ func (h *AgentHandler) AggregateMetrics(c *gin.Context) {
buckets[bucketStart] = append(buckets[bucketStart], val)
}
// 为每个桶计算 OHLC + Avg
// 计算 max/min/avg/p95
for bucketTs, values := range buckets {
if len(values) == 0 {
continue
}
open := values[0]
closeV := values[len(values)-1]
high := values[0]
low := values[0]
maxV := values[0]
minV := values[0]
sum := 0.0
for _, v := range values {
sum += v
if v > high {
high = v
if v > maxV {
maxV = v
}
if v < low {
low = v
if v < minV {
minV = v
}
}
avg := sum / float64(len(values))
kline := models.MetricsKline{
// 计算 P95(排序后取 95% 位置的值)
p95 := maxV
if len(values) >= 20 {
sorted := make([]float64, len(values))
copy(sorted, values)
sort.Float64s(sorted)
idx := int(float64(len(sorted)) * 0.95)
if idx >= len(sorted) {
idx = len(sorted) - 1
}
p95 = sorted[idx]
}
agg := models.MetricsAggregate{
Hostname: req.Hostname,
Metric: req.Metric,
Interval: iv.name,
Open: math.Round(open*100) / 100,
High: math.Round(high*100) / 100,
Low: math.Round(low*100) / 100,
Close: math.Round(closeV*100) / 100,
Avg: math.Round(avg*100) / 100,
MaxV: math.Round(maxV*100) / 100,
MinV: math.Round(minV*100) / 100,
P95: math.Round(p95*100) / 100,
Count: len(values),
BucketTime: time.Unix(bucketTs, 0),
}
@@ -347,9 +418,9 @@ func (h *AgentHandler) AggregateMetrics(c *gin.Context) {
{Name: "bucket_time"},
},
DoUpdates: clause.AssignmentColumns([]string{
"open", "high", "low", "close", "avg", "count",
"avg", "max_v", "min_v", "p95", "count",
}),
}).Create(&kline)
}).Create(&agg)
if result.Error == nil {
totalUpserted++
@@ -364,10 +435,10 @@ func (h *AgentHandler) AggregateMetrics(c *gin.Context) {
})
}
// GetKline 查询 K 线数据
// GET /api/v1/agent/kline?hostname=&metric=&interval=&from=&to=
func (h *AgentHandler) GetKline(c *gin.Context) {
var req models.KlineQueryReq
// GetAggregate 查询聚合数据(原 GetKline)
// GET /api/v1/agent/aggregate?hostname=&metric=&interval=&from=&to=
func (h *AgentHandler) GetAggregate(c *gin.Context) {
var req models.AggregateQueryReq
if err := c.ShouldBindQuery(&req); err != nil {
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "参数错误: " + err.Error()})
return
@@ -384,25 +455,25 @@ func (h *AgentHandler) GetKline(c *gin.Context) {
return
}
var klines []models.MetricsKline
var items []models.MetricsAggregate
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)
Find(&items)
if result.Error != nil {
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "查询失败: " + result.Error.Error()})
return
}
if klines == nil {
klines = []models.MetricsKline{}
if items == nil {
items = []models.MetricsAggregate{}
}
c.JSON(http.StatusOK, models.Response{
Code: 200,
Message: "success",
Data: klines,
Data: items,
})
}