attacker
f3a9e5e8d0
- 主菜单「监控告警」→「监控中心」 - 「告警中心」→「告警服务」移至链路追踪下方 - 新增资源监控 Tab 页（主机/中间件/SNMP/站点拨测） - Agent 新增网络采集器(network)和磁盘IO采集器(diskio) - 后端扩展 Agent 上报模型，支持 NetworkStat/DiskIOStat - 前端 ServerTableRow 展示网络和磁盘IO实时指标 - WebSocket MetricsUpdate 推送扩展
364 lines
10 KiB
Go
364 lines
10 KiB
Go
package handlers
|
|
|
|
import (
|
|
"encoding/json"
|
|
"fmt"
|
|
"math"
|
|
"net/http"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"ezops/internal/models"
|
|
ezredis "ezops/pkg/redis"
|
|
|
|
"github.com/gin-gonic/gin"
|
|
"gorm.io/gorm"
|
|
"gorm.io/gorm/clause"
|
|
)
|
|
|
|
// AgentHandler 处理 Agent 上报数据 // Logan.Li
|
|
type AgentHandler struct {
|
|
db *gorm.DB
|
|
cache ezredis.Cache // 可选 Redis 缓存（用于热数据存储）
|
|
hubMetrics *MetricsHub // WebSocket 广播 metrics 更新
|
|
}
|
|
|
|
func NewAgentHandler(db *gorm.DB, cache ezredis.Cache, hub *MetricsHub) *AgentHandler {
|
|
return &AgentHandler{db: db, cache: cache, hubMetrics: hub}
|
|
}
|
|
|
|
// ReceiveReport 接收 Agent 上报的指标数据
|
|
// POST /api/v1/agent/report
|
|
func (h *AgentHandler) ReceiveReport(c *gin.Context) {
|
|
var req models.AgentReportRequest
|
|
if err := c.ShouldBindJSON(&req); err != nil {
|
|
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "请求格式错误: " + err.Error()})
|
|
return
|
|
}
|
|
|
|
// 解析上报时间
|
|
reportTime := time.Now()
|
|
if req.Timestamp != "" {
|
|
if t, err := time.Parse(time.RFC3339, req.Timestamp); err == nil {
|
|
reportTime = t
|
|
}
|
|
}
|
|
|
|
// 序列化磁盘和端口信息
|
|
diskJSON, _ := json.Marshal(req.Disks)
|
|
portJSON, _ := json.Marshal(req.Ports)
|
|
netJSON, _ := json.Marshal(req.Network)
|
|
dioJSON, _ := json.Marshal(req.DiskIO)
|
|
|
|
// 存储到数据库
|
|
report := models.AgentReport{
|
|
Hostname: req.Hostname,
|
|
CPUUsage: req.CPU.UsagePercent,
|
|
CPUCores: req.CPU.CoreCount,
|
|
Load1: req.CPU.Load1,
|
|
Load5: req.CPU.Load5,
|
|
Load15: req.CPU.Load15,
|
|
MemTotal: req.Memory.TotalGB,
|
|
MemUsed: req.Memory.UsedGB,
|
|
MemUsage: req.Memory.UsagePercent,
|
|
DiskInfo: string(diskJSON),
|
|
PortInfo: string(portJSON),
|
|
NetInfo: string(netJSON),
|
|
DiskIOInfo: string(dioJSON),
|
|
ReportTime: reportTime,
|
|
}
|
|
|
|
if err := h.db.Create(&report).Error; err != nil {
|
|
c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "保存失败: " + err.Error()})
|
|
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())
|
|
ttl := 7 * 24 * time.Hour // 7天
|
|
|
|
// 使用 reportTime.UnixNano() 确保 member 唯一（避免相同值碰撞）
|
|
h.writeHotMetric(c, "metrics:cpu:"+req.Hostname, req.CPU.UsagePercent, timestamp, ttl, reportTime)
|
|
h.writeHotMetric(c, "metrics:mem:"+req.Hostname, req.Memory.UsagePercent, timestamp, ttl, reportTime)
|
|
h.writeHotMetric(c, "metrics:load1:"+req.Hostname, req.CPU.Load1, timestamp, ttl, reportTime)
|
|
}
|
|
|
|
// 保存成功
|
|
c.JSON(http.StatusOK, models.Response{Code: 200, Message: "ok"})
|
|
}
|
|
|
|
// writeHotMetric 将单个指标写入 Redis 有序集合并维护 TTL 和过期数据清理
|
|
func (h *AgentHandler) writeHotMetric(c *gin.Context, key string, value float64, score float64, ttl time.Duration, reportTime time.Time) {
|
|
ctx := c.Request.Context()
|
|
|
|
// member 格式: "value:nano_timestamp" 保证唯一性
|
|
member := fmt.Sprintf("%.4f:%d", value, reportTime.UnixNano())
|
|
|
|
if err := h.cache.ZAdd(ctx, key, ezredis.SortedSetMember{
|
|
Score: score,
|
|
Member: member,
|
|
}); err != nil {
|
|
// 静默失败，不影响主流程
|
|
return
|
|
}
|
|
|
|
// 设置 key 过期时间
|
|
_ = h.cache.Expire(ctx, key, ttl)
|
|
|
|
// 清理 7 天前的旧数据
|
|
cutoff := float64(time.Now().Add(-ttl).Unix())
|
|
_ = h.cache.ZRemRangeByScore(ctx, key, "-inf", fmt.Sprintf("%.0f", cutoff))
|
|
}
|
|
|
|
// GetLatestReports 获取各主机最新上报
|
|
// GET /api/v1/agent/latest
|
|
func (h *AgentHandler) GetLatestReports(c *gin.Context) {
|
|
// 子查询：每个 hostname 最新的一条记录
|
|
var reports []models.AgentReport
|
|
h.db.Raw(`
|
|
SELECT * FROM agent_reports
|
|
WHERE id IN (
|
|
SELECT MAX(id) FROM agent_reports GROUP BY hostname
|
|
)
|
|
ORDER BY report_time DESC
|
|
`).Scan(&reports)
|
|
|
|
c.JSON(http.StatusOK, models.Response{
|
|
Code: 200,
|
|
Message: "success",
|
|
Data: reports,
|
|
})
|
|
}
|
|
|
|
// GetHostHistory 获取指定主机的历史指标
|
|
// GET /api/v1/agent/history/:hostname?hours=1
|
|
func (h *AgentHandler) GetHostHistory(c *gin.Context) {
|
|
hostname := c.Param("hostname")
|
|
hours := 1
|
|
if v := c.Query("hours"); v != "" {
|
|
fmt.Sscanf(v, "%d", &hours)
|
|
}
|
|
since := time.Now().Add(-time.Duration(hours) * time.Hour)
|
|
|
|
var reports []models.AgentReport
|
|
h.db.Where("hostname = ? AND report_time >= ?", hostname, since).
|
|
Order("report_time ASC").
|
|
Find(&reports)
|
|
|
|
c.JSON(http.StatusOK, models.Response{
|
|
Code: 200,
|
|
Message: "success",
|
|
Data: reports,
|
|
})
|
|
}
|
|
|
|
// AggregateMetrics 聚合 Redis 热数据到 MySQL klines
|
|
// POST /api/v1/agent/aggregate
|
|
func (h *AgentHandler) AggregateMetrics(c *gin.Context) {
|
|
var req models.AggregateKlineReq
|
|
if err := c.ShouldBindJSON(&req); err != nil {
|
|
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "请求格式错误: " + err.Error()})
|
|
return
|
|
}
|
|
|
|
if h.cache == nil {
|
|
c.JSON(http.StatusServiceUnavailable, models.Response{Code: 503, Message: "Redis 缓存不可用"})
|
|
return
|
|
}
|
|
|
|
// 支持的聚合间隔及对应的回溯时长
|
|
intervals := []struct {
|
|
name string
|
|
duration time.Duration
|
|
}{
|
|
{"1m", 1 * time.Minute},
|
|
{"5m", 5 * time.Minute},
|
|
{"3m", 3 * time.Minute},
|
|
{"1h", 1 * time.Hour},
|
|
{"6h", 6 * time.Hour},
|
|
{"1d", 24 * time.Hour},
|
|
}
|
|
|
|
redisKey := fmt.Sprintf("metrics:%s:%s", req.Metric, req.Hostname)
|
|
ctx := c.Request.Context()
|
|
var totalUpserted int64
|
|
|
|
for _, iv := range intervals {
|
|
// 回溯窗口：取 interval × 200 倍，确保覆盖足够多的数据点
|
|
lookback := iv.duration * 200
|
|
now := time.Now()
|
|
since := now.Add(-lookback)
|
|
minScore := fmt.Sprintf("%d", since.Unix())
|
|
maxScore := fmt.Sprintf("%d", now.Unix()+1)
|
|
|
|
members, err := h.cache.ZRangeByScore(ctx, redisKey, minScore, maxScore)
|
|
if err != nil || len(members) == 0 {
|
|
continue
|
|
}
|
|
|
|
// 按间隔分桶
|
|
buckets := make(map[int64][]float64) // bucketTime (unix) → values
|
|
for _, m := range members {
|
|
ts := int64(m.Score)
|
|
bucketStart := (ts / int64(iv.duration.Seconds())) * int64(iv.duration.Seconds())
|
|
val := parseMemberValue(m.Member)
|
|
buckets[bucketStart] = append(buckets[bucketStart], val)
|
|
}
|
|
|
|
// 为每个桶计算 OHLC + Avg
|
|
for bucketTs, values := range buckets {
|
|
if len(values) == 0 {
|
|
continue
|
|
}
|
|
open := values[0]
|
|
closeV := values[len(values)-1]
|
|
high := values[0]
|
|
low := values[0]
|
|
sum := 0.0
|
|
|
|
for _, v := range values {
|
|
sum += v
|
|
if v > high {
|
|
high = v
|
|
}
|
|
if v < low {
|
|
low = v
|
|
}
|
|
}
|
|
avg := sum / float64(len(values))
|
|
|
|
kline := models.MetricsKline{
|
|
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,
|
|
Count: len(values),
|
|
BucketTime: time.Unix(bucketTs, 0),
|
|
}
|
|
|
|
// Upsert: hostname + metric + interval + bucket_time 唯一
|
|
result := h.db.Clauses(clause.OnConflict{
|
|
Columns: []clause.Column{
|
|
{Name: "hostname"},
|
|
{Name: "metric"},
|
|
{Name: "interval"},
|
|
{Name: "bucket_time"},
|
|
},
|
|
DoUpdates: clause.AssignmentColumns([]string{
|
|
"open", "high", "low", "close", "avg", "count",
|
|
}),
|
|
}).Create(&kline)
|
|
|
|
if result.Error == nil {
|
|
totalUpserted++
|
|
}
|
|
}
|
|
}
|
|
|
|
c.JSON(http.StatusOK, models.Response{
|
|
Code: 200,
|
|
Message: fmt.Sprintf("聚合完成，写入/更新 %d 条记录", totalUpserted),
|
|
Data: gin.H{"upserted": totalUpserted},
|
|
})
|
|
}
|
|
|
|
// GetKline 查询 K 线数据
|
|
// GET /api/v1/agent/kline?hostname=&metric=&interval=&from=&to=
|
|
func (h *AgentHandler) GetKline(c *gin.Context) {
|
|
var req models.KlineQueryReq
|
|
if err := c.ShouldBindQuery(&req); err != nil {
|
|
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "参数错误: " + err.Error()})
|
|
return
|
|
}
|
|
|
|
from, err := time.Parse(time.RFC3339, req.From)
|
|
if err != nil {
|
|
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "from 格式错误，需 RFC3339: " + err.Error()})
|
|
return
|
|
}
|
|
to, err := time.Parse(time.RFC3339, req.To)
|
|
if err != nil {
|
|
c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "to 格式错误，需 RFC3339: " + err.Error()})
|
|
return
|
|
}
|
|
|
|
var klines []models.MetricsKline
|
|
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",
|
|
Data: klines,
|
|
})
|
|
}
|
|
|
|
// parseMemberValue 从 member 字符串 "45.2300:1234567890123456789" 中提取浮点值
|
|
func parseMemberValue(member string) float64 {
|
|
idx := strings.Index(member, ":")
|
|
if idx < 0 {
|
|
v, _ := strconv.ParseFloat(member, 64)
|
|
return v
|
|
}
|
|
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
|
|
}
|