Logan.Li
610402d47f
现象：/servers 资产里有 31 台，/dashboards 大屏一台都不显示。
根因：GetLatestReports 的 SQL 中 JOIN 了不存在的表 asset_regions，
实际表名是 asset_cloud_regions（见 model/server.go 的 Region.TableName()）。
查询直接报 Error 1146，但 Scan() 的错误未校验，函数静默返回空数组，
接口始终 200 + data:[]，因此前端无任何报错可循。
- agent.go: JOIN 改为 asset_cloud_regions，并补上错误校验与 500 响应
- agent_report_test.go: 移除手工创建的 asset_regions，
改由 AutoMigrate(&model.Server{}) 级联建立 asset_cloud_regions，
避免测试用错误表名掩盖生产问题
验证：/agent/latest 由 0 台恢复为 31 台（含未装 Agent 的 not_installed 主机）。
// Logan.Li
796 lines
24 KiB
Go
796 lines
24 KiB
Go
package controller
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"log"
|
||
"math"
|
||
"net/http"
|
||
"sort"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
"ezops/middleware"
|
||
"ezops/model"
|
||
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}
|
||
}
|
||
|
||
// resolveAgentAsset 优先按上报主机名或内网 IP 关联现有资产，避免云实例名与系统 hostname 不一致。 // Logan.Li
|
||
func (h *AgentHandler) resolveAgentAsset(hostname, reportIP, clientIP string) (model.Server, string, bool) {
|
||
matchIP := strings.TrimSpace(reportIP)
|
||
if matchIP == "" || matchIP == "unknown" {
|
||
matchIP = strings.TrimSpace(clientIP)
|
||
}
|
||
|
||
var server model.Server
|
||
query := h.db.Where("hostname = ?", hostname)
|
||
if matchIP != "" {
|
||
query = h.db.Where("hostname = ? OR ip = ?", hostname, matchIP)
|
||
}
|
||
if err := query.Take(&server).Error; err == nil {
|
||
return server, matchIP, true
|
||
}
|
||
return server, matchIP, false
|
||
}
|
||
|
||
// ReceiveReport 接收 Agent 上报的指标数据
|
||
// POST /api/v1/agent/report
|
||
func (h *AgentHandler) ReceiveReport(c *gin.Context) {
|
||
var req model.AgentReportRequest
|
||
if err := c.ShouldBindJSON(&req); err != nil {
|
||
c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: "请求格式错误: " + err.Error()})
|
||
return
|
||
}
|
||
if !middleware.RequireAgentHostname(c, req.Hostname) {
|
||
return
|
||
}
|
||
|
||
// 解析上报时间
|
||
reportTime := time.Now()
|
||
if req.Timestamp != "" {
|
||
if t, err := time.Parse(time.RFC3339, req.Timestamp); err == nil {
|
||
reportTime = t
|
||
}
|
||
}
|
||
// Agent 上报时间统一存储为 UTC，避免 SQLite 按带时区文本比较时误判。 // Logan.Li
|
||
reportTime = reportTime.UTC()
|
||
|
||
matchedServer, matchIP, assetMatched := h.resolveAgentAsset(req.Hostname, req.IP, c.ClientIP())
|
||
canonicalHostname := req.Hostname
|
||
if assetMatched {
|
||
canonicalHostname = matchedServer.Hostname
|
||
}
|
||
|
||
// 序列化磁盘和端口信息
|
||
diskJSON, _ := json.Marshal(req.Disks)
|
||
portJSON, _ := json.Marshal(req.Ports)
|
||
netJSON, _ := json.Marshal(req.Network)
|
||
dioJSON, _ := json.Marshal(req.DiskIO)
|
||
// v2.0 新增
|
||
swapJSON, _ := json.Marshal(req.Swap)
|
||
processTopJSON, _ := json.Marshal(req.ProcessTop)
|
||
fdJSON, _ := json.Marshal(req.FD)
|
||
csJSON, _ := json.Marshal(req.ContextSwitches)
|
||
heartbeatJSON, _ := json.Marshal(req.Heartbeat)
|
||
|
||
// 存储到数据库
|
||
report := model.AgentReport{
|
||
Hostname: canonicalHostname,
|
||
CPUUsage: req.CPU.UsagePercent,
|
||
CPUIOwait: req.CPU.IOWait,
|
||
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,
|
||
MemAvailable: req.Memory.AvailableGB,
|
||
MemSwapUsed: req.Memory.SwapUsedGB,
|
||
DiskInfo: string(diskJSON),
|
||
PortInfo: string(portJSON),
|
||
NetInfo: string(netJSON),
|
||
DiskIOInfo: string(dioJSON),
|
||
// v2.0 新增
|
||
SwapInfo: string(swapJSON),
|
||
ProcessTopInfo: string(processTopJSON),
|
||
FDInfo: string(fdJSON),
|
||
CSInfo: string(csJSON),
|
||
HeartbeatInfo: string(heartbeatJSON),
|
||
ReportTime: reportTime,
|
||
}
|
||
|
||
if err := h.db.Create(&report).Error; err != nil {
|
||
c.JSON(http.StatusInternalServerError, model.Response{Code: 500, Message: "保存失败: " + err.Error()})
|
||
return
|
||
}
|
||
|
||
// 同步更新 servers 表的实时指标（三级匹配：hostname → client IP → hostname LIKE）
|
||
diskAvg := avgDiskUsage(req.Disks)
|
||
|
||
// 计算硬件规格（四舍五入到整数，避免 31.34→31GB 的显示偏差）
|
||
totalDiskGB := 0.0
|
||
for _, d := range req.Disks {
|
||
totalDiskGB += d.TotalGB
|
||
}
|
||
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
|
||
|
||
// 使用 map 按主键更新，避免可选系统字段导致原始 SQL 占位参数错位。 // Logan.Li
|
||
updates := map[string]interface{}{
|
||
"cpu_usage": req.CPU.UsagePercent, "memory_usage": req.Memory.UsagePercent,
|
||
"disk_usage": diskAvg,
|
||
"cpu": req.CPU.CoreCount, "memory": memMB, "disk": diskGB,
|
||
"status": "running", "agent_status": "online", "updated_at": time.Now(),
|
||
}
|
||
if req.AgentVersion != "" {
|
||
updates["agent_version"] = req.AgentVersion
|
||
}
|
||
if req.System != nil {
|
||
if req.System.OSVersion != "" {
|
||
updates["os"] = req.System.OSVersion
|
||
}
|
||
if req.System.Kernel != "" {
|
||
updates["kernel"] = req.System.Kernel
|
||
}
|
||
if req.System.Uptime > 0 {
|
||
updates["uptime"] = req.System.Uptime
|
||
}
|
||
}
|
||
// 兼容旧版 Agent：系统信息缺席时使用心跳中的运行时长补齐资产元数据。 // Logan.Li
|
||
if _, ok := updates["uptime"]; !ok && req.Heartbeat.UptimeSec > 0 {
|
||
updates["uptime"] = req.Heartbeat.UptimeSec
|
||
}
|
||
var result *gorm.DB
|
||
if assetMatched {
|
||
result = h.db.Model(&model.Server{}).Where("id = ?", matchedServer.ID).Updates(updates)
|
||
} else {
|
||
result = h.db.Model(&model.Server{}).Where("hostname = ? OR ip = ?", req.Hostname, matchIP).Updates(updates)
|
||
}
|
||
|
||
// 如果还是没匹配到 —— 按 hostname 显式查重后再注册 // Logan.Li
|
||
// asset_servers.hostname 无 UNIQUE 索引，OnConflict 不可靠，故先 First 查重，
|
||
// 避免 Agent 每次上报（10s）都新建记录导致资产越来越多。
|
||
if result.Error == nil && result.RowsAffected == 0 {
|
||
osName := ""
|
||
kernel := ""
|
||
uptime := req.Heartbeat.UptimeSec
|
||
if req.System != nil {
|
||
osName = req.System.OSVersion
|
||
kernel = req.System.Kernel
|
||
if req.System.Uptime > 0 {
|
||
uptime = req.System.Uptime
|
||
}
|
||
}
|
||
var existing model.Server
|
||
if h.db.Where("hostname = ?", req.Hostname).Take(&existing).Error == nil {
|
||
// 已自动注册过 → 仅更新实时指标，不重复创建
|
||
existingUpdates := map[string]interface{}{
|
||
"ip": matchIP, "os": osName, "cpu": req.CPU.CoreCount,
|
||
"memory": memMB, "disk": diskGB, "status": "running",
|
||
"agent_status": "online", "agent_version": req.AgentVersion,
|
||
"cpu_usage": req.CPU.UsagePercent, "memory_usage": req.Memory.UsagePercent,
|
||
"disk_usage": diskAvg,
|
||
}
|
||
if kernel != "" {
|
||
existingUpdates["kernel"] = kernel
|
||
}
|
||
if uptime > 0 {
|
||
existingUpdates["uptime"] = uptime
|
||
}
|
||
h.db.Model(&existing).Updates(existingUpdates)
|
||
} else {
|
||
// 真正的新主机 → 注册，hostname 无唯一索引时不使用 OnConflict。 // Logan.Li
|
||
newServer := model.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,
|
||
Kernel: kernel,
|
||
Uptime: uptime,
|
||
CPUUsage: req.CPU.UsagePercent,
|
||
MemoryUsage: req.Memory.UsagePercent,
|
||
DiskUsage: diskAvg,
|
||
}
|
||
if err := h.db.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
|
||
if h.hubMetrics != nil && result.RowsAffected > 0 {
|
||
h.hubMetrics.Broadcast(MetricsUpdate{
|
||
Type: "metrics_update",
|
||
Hostname: canonicalHostname,
|
||
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 有序集合热数据（redka/local 无 ZSet，数据已在 monitor_agent_reports）// Logan.Li
|
||
if h.cache != nil && h.cache.Capabilities().SortedSet {
|
||
timestamp := float64(reportTime.Unix())
|
||
ttl := 7 * 24 * time.Hour
|
||
h.writeHotMetric(c, "metrics:cpu:"+canonicalHostname, req.CPU.UsagePercent, timestamp, ttl, reportTime)
|
||
h.writeHotMetric(c, "metrics:mem:"+canonicalHostname, req.Memory.UsagePercent, timestamp, ttl, reportTime)
|
||
h.writeHotMetric(c, "metrics:load1:"+canonicalHostname, req.CPU.Load1, timestamp, ttl, reportTime)
|
||
}
|
||
|
||
// 保存成功 — 响应中携带探针配置给 Agent
|
||
resp := h.makeProbeConfigResponse(canonicalHostname, 200, "ok", req.ProbeResults)
|
||
c.JSON(http.StatusOK, resp)
|
||
}
|
||
|
||
// 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 获取所有服务器的 Agent 最新状态（含未安装 Agent 的主机）
|
||
// GET /api/v1/agent/latest
|
||
func (h *AgentHandler) GetLatestReports(c *gin.Context) {
|
||
var results []model.AgentLatestResponse
|
||
// 查询边界与 Agent 上报的 UTC 时间保持一致。 // Logan.Li
|
||
threeMinAgo := time.Now().UTC().Add(-3 * time.Minute)
|
||
// 以 servers 表为主，LEFT JOIN 最近 3 分钟内的 Agent 上报
|
||
// 未装 Agent 的主机也会显示（agent_status='not_installed'），打破"不显示→无法安装"的循环
|
||
// 必须检查错误：此前未校验导致 asset_regions 表名错误被静默吞掉，
|
||
// 接口稳定返回空数组，表现为"资产有主机、大屏没主机"却无任何报错。 // Logan.Li
|
||
if err := h.db.Raw(`
|
||
SELECT
|
||
s.id as server_id,
|
||
s.hostname,
|
||
COALESCE(s.ip, '') as ip,
|
||
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.swap_info,
|
||
a.process_top_info,
|
||
a.fd_info,
|
||
a.cs_info,
|
||
a.heartbeat_info,
|
||
a.report_time,
|
||
COALESCE(s.agent_version, '') as agent_version,
|
||
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(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_cloud_regions r ON r.id = s.region_id
|
||
ORDER BY a.report_time DESC, s.id ASC
|
||
`, threeMinAgo).Scan(&results).Error; err != nil {
|
||
log.Printf("[GetLatestReports] 查询主机最新上报失败: %v", err)
|
||
c.JSON(http.StatusInternalServerError, model.Response{
|
||
Code: 500, Message: "查询主机列表失败: " + err.Error()})
|
||
return
|
||
}
|
||
|
||
if results == nil {
|
||
results = []model.AgentLatestResponse{}
|
||
}
|
||
|
||
c.JSON(http.StatusOK, model.Response{
|
||
Code: 200,
|
||
Message: "success",
|
||
Data: results,
|
||
})
|
||
}
|
||
|
||
// parseFlexibleTime 解析前端传入的时间（ISO / datetime-local） // Logan.Li
|
||
func parseFlexibleTime(s string) (time.Time, error) {
|
||
s = strings.TrimSpace(s)
|
||
layouts := []string{
|
||
time.RFC3339,
|
||
"2006-01-02T15:04:05Z07:00",
|
||
"2006-01-02 15:04:05",
|
||
"2006-01-02T15:04",
|
||
"2006-01-02 15:04",
|
||
}
|
||
for _, layout := range layouts {
|
||
if t, err := time.ParseInLocation(layout, s, time.Local); err == nil {
|
||
return t, nil
|
||
}
|
||
}
|
||
return time.Time{}, fmt.Errorf("无法解析时间: %s", s)
|
||
}
|
||
|
||
// resolveHostHistoryRange 解析历史查询时间范围（minutes / hours / start+end） // Logan.Li
|
||
func resolveHostHistoryRange(c *gin.Context) (since, until time.Time, err error) {
|
||
until = time.Now()
|
||
if startStr := strings.TrimSpace(c.Query("start")); startStr != "" {
|
||
endStr := strings.TrimSpace(c.Query("end"))
|
||
if endStr == "" {
|
||
return since, until, fmt.Errorf("自定义范围缺少 end 参数")
|
||
}
|
||
since, err = parseFlexibleTime(startStr)
|
||
if err != nil {
|
||
return since, until, err
|
||
}
|
||
until, err = parseFlexibleTime(endStr)
|
||
if err != nil {
|
||
return since, until, err
|
||
}
|
||
if until.Before(since) {
|
||
since, until = until, since
|
||
}
|
||
return since, until, nil
|
||
}
|
||
if minStr := strings.TrimSpace(c.Query("minutes")); minStr != "" {
|
||
mins, _ := strconv.Atoi(minStr)
|
||
if mins <= 0 {
|
||
mins = 60
|
||
}
|
||
if mins > 43200 {
|
||
mins = 43200
|
||
}
|
||
since = until.Add(-time.Duration(mins) * time.Minute)
|
||
return since, until, nil
|
||
}
|
||
hours := 1
|
||
if v := strings.TrimSpace(c.Query("hours")); v != "" {
|
||
hours, _ = strconv.Atoi(v)
|
||
if hours <= 0 {
|
||
hours = 1
|
||
}
|
||
if hours > 720 {
|
||
hours = 720
|
||
}
|
||
}
|
||
since = until.Add(-time.Duration(hours) * time.Hour)
|
||
return since, until, nil
|
||
}
|
||
|
||
// GetHostHistory 获取指定主机的历史指标
|
||
// GET /api/v1/agent/history/:hostname?minutes=60 | ?hours=1 | ?start=&end=
|
||
func (h *AgentHandler) GetHostHistory(c *gin.Context) {
|
||
hostname := c.Param("hostname")
|
||
since, until, err := resolveHostHistoryRange(c)
|
||
if err != nil {
|
||
c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: err.Error()})
|
||
return
|
||
}
|
||
|
||
var reports []model.AgentReport
|
||
// SQLite 会按时间文本比较，查询边界需统一转换为 UTC。 // Logan.Li
|
||
h.db.Where("hostname = ? AND report_time >= ? AND report_time <= ?", hostname, since.UTC(), until.UTC()).
|
||
Order("report_time ASC").
|
||
Find(&reports)
|
||
|
||
c.JSON(http.StatusOK, model.Response{
|
||
Code: 200,
|
||
Message: "success",
|
||
Data: reports,
|
||
})
|
||
}
|
||
|
||
// AggregateMetrics 聚合指标到 monitor_metrics_aggregate（Redis ZSet 或 DB 直读）// Logan.Li
|
||
// POST /api/v1/agent/aggregate（需要控制台 JWT） // Logan.Li
|
||
func (h *AgentHandler) AggregateMetrics(c *gin.Context) {
|
||
var req model.AggregateReq
|
||
if err := c.ShouldBindJSON(&req); err != nil {
|
||
c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: "请求格式错误: " + err.Error()})
|
||
return
|
||
}
|
||
|
||
var totalUpserted int64
|
||
var err error
|
||
if h.cache != nil && h.cache.Capabilities().SortedSet {
|
||
totalUpserted, err = h.aggregateFromRedis(c.Request.Context(), req)
|
||
} else {
|
||
totalUpserted, err = h.aggregateFromReports(req)
|
||
}
|
||
if err != nil {
|
||
c.JSON(http.StatusInternalServerError, model.Response{Code: 500, Message: err.Error()})
|
||
return
|
||
}
|
||
|
||
c.JSON(http.StatusOK, model.Response{
|
||
Code: 200,
|
||
Message: fmt.Sprintf("聚合完成，写入/更新 %d 条记录", totalUpserted),
|
||
Data: gin.H{"upserted": totalUpserted, "source": aggregateSource(h.cache)},
|
||
})
|
||
}
|
||
|
||
func aggregateSource(cache ezredis.Cache) string {
|
||
if cache != nil && cache.Capabilities().SortedSet {
|
||
return "redis"
|
||
}
|
||
return "database"
|
||
}
|
||
|
||
func (h *AgentHandler) aggregateFromRedis(ctx context.Context, req model.AggregateReq) (int64, error) {
|
||
if h.cache == nil {
|
||
return 0, fmt.Errorf("Redis 缓存不可用")
|
||
}
|
||
intervals := aggregateIntervals()
|
||
redisKey := fmt.Sprintf("metrics:%s:%s", req.Metric, req.Hostname)
|
||
var total int64
|
||
|
||
for _, iv := range intervals {
|
||
lookback := iv.duration * 200
|
||
now := time.Now()
|
||
since := now.Add(-lookback)
|
||
members, err := h.cache.ZRangeByScore(ctx, redisKey,
|
||
fmt.Sprintf("%d", since.Unix()), fmt.Sprintf("%d", now.Unix()+1))
|
||
if err != nil || len(members) == 0 {
|
||
continue
|
||
}
|
||
buckets := make(map[int64][]float64)
|
||
for _, m := range members {
|
||
ts := int64(m.Score)
|
||
bucketStart := (ts / int64(iv.duration.Seconds())) * int64(iv.duration.Seconds())
|
||
buckets[bucketStart] = append(buckets[bucketStart], parseMemberValue(m.Member))
|
||
}
|
||
n, err := h.upsertMetricBuckets(req, iv.name, buckets)
|
||
if err != nil {
|
||
return total, err
|
||
}
|
||
total += n
|
||
}
|
||
return total, nil
|
||
}
|
||
|
||
// aggregateFromReports local 模式：从 monitor_agent_reports 聚合（无需 Redis ZSet）// Logan.Li
|
||
func (h *AgentHandler) aggregateFromReports(req model.AggregateReq) (int64, error) {
|
||
intervals := aggregateIntervals()
|
||
var total int64
|
||
|
||
for _, iv := range intervals {
|
||
lookback := iv.duration * 200
|
||
// 本地 SQLite 模式下使用 UTC 边界查询 Agent 报告。 // Logan.Li
|
||
since := time.Now().UTC().Add(-lookback)
|
||
var reports []model.AgentReport
|
||
if err := h.db.Where("hostname = ? AND report_time >= ?", req.Hostname, since).
|
||
Order("report_time ASC").Find(&reports).Error; err != nil {
|
||
return total, err
|
||
}
|
||
if len(reports) == 0 {
|
||
continue
|
||
}
|
||
buckets := make(map[int64][]float64)
|
||
sec := int64(iv.duration.Seconds())
|
||
for _, r := range reports {
|
||
ts := r.ReportTime.Unix()
|
||
bucketStart := (ts / sec) * sec
|
||
buckets[bucketStart] = append(buckets[bucketStart], metricValueFromReport(r, req.Metric))
|
||
}
|
||
n, err := h.upsertMetricBuckets(req, iv.name, buckets)
|
||
if err != nil {
|
||
return total, err
|
||
}
|
||
total += n
|
||
}
|
||
return total, nil
|
||
}
|
||
|
||
func aggregateIntervals() []struct {
|
||
name string
|
||
duration time.Duration
|
||
} {
|
||
return []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},
|
||
}
|
||
}
|
||
|
||
func metricValueFromReport(r model.AgentReport, metric string) float64 {
|
||
switch metric {
|
||
case "cpu":
|
||
return r.CPUUsage
|
||
case "mem":
|
||
return r.MemUsage
|
||
case "load1":
|
||
return r.Load1
|
||
default:
|
||
return 0
|
||
}
|
||
}
|
||
|
||
func (h *AgentHandler) upsertMetricBuckets(req model.AggregateReq, interval string, buckets map[int64][]float64) (int64, error) {
|
||
var total int64
|
||
for bucketTs, values := range buckets {
|
||
if len(values) == 0 {
|
||
continue
|
||
}
|
||
maxV, minV, avg, p95 := calcAggregateStats(values)
|
||
agg := model.MetricsAggregate{
|
||
Hostname: req.Hostname,
|
||
Metric: req.Metric,
|
||
Interval: interval,
|
||
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),
|
||
}
|
||
result := h.db.Clauses(clause.OnConflict{
|
||
Columns: []clause.Column{
|
||
{Name: "hostname"},
|
||
{Name: "metric"},
|
||
{Name: "interval"},
|
||
{Name: "bucket_time"},
|
||
},
|
||
DoUpdates: clause.AssignmentColumns([]string{
|
||
"avg", "max_v", "min_v", "p95", "count",
|
||
}),
|
||
}).Create(&agg)
|
||
if result.Error != nil {
|
||
return total, result.Error
|
||
}
|
||
total++
|
||
}
|
||
return total, nil
|
||
}
|
||
|
||
func calcAggregateStats(values []float64) (maxV, minV, avg, p95 float64) {
|
||
maxV = values[0]
|
||
minV = values[0]
|
||
sum := 0.0
|
||
for _, v := range values {
|
||
sum += v
|
||
if v > maxV {
|
||
maxV = v
|
||
}
|
||
if v < minV {
|
||
minV = v
|
||
}
|
||
}
|
||
avg = sum / float64(len(values))
|
||
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]
|
||
}
|
||
return
|
||
}
|
||
|
||
// GetAggregate 查询聚合数据（原 GetKline）
|
||
// GET /api/v1/agent/aggregate?hostname=&metric=&interval=&from=&to=
|
||
func (h *AgentHandler) GetAggregate(c *gin.Context) {
|
||
var req model.AggregateQueryReq
|
||
if err := c.ShouldBindQuery(&req); err != nil {
|
||
c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: "参数错误: " + err.Error()})
|
||
return
|
||
}
|
||
|
||
from, err := time.Parse(time.RFC3339, req.From)
|
||
if err != nil {
|
||
c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: "from 格式错误，需 RFC3339: " + err.Error()})
|
||
return
|
||
}
|
||
to, err := time.Parse(time.RFC3339, req.To)
|
||
if err != nil {
|
||
c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: "to 格式错误，需 RFC3339: " + err.Error()})
|
||
return
|
||
}
|
||
|
||
var items []model.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(&items)
|
||
|
||
if result.Error != nil {
|
||
c.JSON(http.StatusInternalServerError, model.Response{Code: 500, Message: "查询失败: " + result.Error.Error()})
|
||
return
|
||
}
|
||
|
||
if items == nil {
|
||
items = []model.MetricsAggregate{}
|
||
}
|
||
|
||
c.JSON(http.StatusOK, model.Response{
|
||
Code: 200,
|
||
Message: "success",
|
||
Data: items,
|
||
})
|
||
}
|
||
|
||
// 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 []model.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
|
||
}
|
||
|
||
// ── 端口监控策略 ── // Logan.Li
|
||
|
||
// GetPortPolicyForAgent Agent 拉取本机端口策略（无需鉴权，通过 hostname 查询）
|
||
// GET /api/v1/agent/port-policy?hostname=xxx
|
||
func (h *AgentHandler) GetPortPolicyForAgent(c *gin.Context) {
|
||
hostname := c.Query("hostname")
|
||
if !middleware.RequireAgentHostname(c, hostname) {
|
||
return
|
||
}
|
||
if hostname == "" {
|
||
c.JSON(http.StatusOK, model.Response{Code: 200, Message: "success", Data: model.AgentPortPolicyDTO{
|
||
PortFilterMode: "whitelist",
|
||
PortWhitelist: []int{},
|
||
}})
|
||
return
|
||
}
|
||
|
||
var policy model.AgentPortPolicy
|
||
if err := h.db.Where("hostname = ?", hostname).First(&policy).Error; err != nil {
|
||
c.JSON(http.StatusOK, model.Response{Code: 200, Message: "success", Data: model.AgentPortPolicyDTO{
|
||
Hostname: hostname,
|
||
PortFilterMode: "whitelist",
|
||
PortWhitelist: []int{},
|
||
}})
|
||
return
|
||
}
|
||
|
||
var whitelist []int
|
||
json.Unmarshal([]byte(policy.PortWhitelist), &whitelist)
|
||
|
||
c.JSON(http.StatusOK, model.Response{Code: 200, Message: "success", Data: model.AgentPortPolicyDTO{
|
||
Hostname: policy.Hostname,
|
||
PortFilterMode: policy.PortFilterMode,
|
||
PortWhitelist: whitelist,
|
||
}})
|
||
}
|
||
|
||
// GetPortPolicy 前端获取主机的端口监控策略
|
||
// GET /api/v1/agent/port-policies/:hostname
|
||
func (h *AgentHandler) GetPortPolicy(c *gin.Context) {
|
||
hostname := c.Param("hostname")
|
||
|
||
var policy model.AgentPortPolicy
|
||
var whitelist []int
|
||
if err := h.db.Where("hostname = ?", hostname).First(&policy).Error; err != nil {
|
||
c.JSON(http.StatusOK, model.Response{Code: 200, Message: "success", Data: model.AgentPortPolicyDTO{
|
||
Hostname: hostname,
|
||
PortFilterMode: "whitelist",
|
||
PortWhitelist: []int{},
|
||
}})
|
||
return
|
||
}
|
||
|
||
json.Unmarshal([]byte(policy.PortWhitelist), &whitelist)
|
||
c.JSON(http.StatusOK, model.Response{Code: 200, Message: "success", Data: model.AgentPortPolicyDTO{
|
||
Hostname: policy.Hostname,
|
||
PortFilterMode: policy.PortFilterMode,
|
||
PortWhitelist: whitelist,
|
||
}})
|
||
}
|
||
|
||
// SavePortPolicy 更新/创建主机的端口监控策略
|
||
// PUT /api/v1/agent/port-policies/:hostname
|
||
func (h *AgentHandler) SavePortPolicy(c *gin.Context) {
|
||
hostname := c.Param("hostname")
|
||
|
||
var dto model.AgentPortPolicyDTO
|
||
if err := c.ShouldBindJSON(&dto); err != nil {
|
||
c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: "请求格式错误: " + err.Error()})
|
||
return
|
||
}
|
||
|
||
whitelistJSON := "[]"
|
||
if len(dto.PortWhitelist) > 0 {
|
||
b, _ := json.Marshal(dto.PortWhitelist)
|
||
whitelistJSON = string(b)
|
||
}
|
||
|
||
policy := model.AgentPortPolicy{
|
||
Hostname: hostname,
|
||
PortFilterMode: dto.PortFilterMode,
|
||
PortWhitelist: whitelistJSON,
|
||
}
|
||
|
||
if err := h.db.Where(model.AgentPortPolicy{Hostname: hostname}).
|
||
Assign(policy).
|
||
FirstOrCreate(&policy).Error; err != nil {
|
||
c.JSON(http.StatusInternalServerError, model.Response{Code: 500, Message: "保存失败: " + err.Error()})
|
||
return
|
||
}
|
||
|
||
c.JSON(http.StatusOK, model.Response{Code: 200, Message: "端口策略已保存"})
|
||
}
|