fix: 修复Agent按IP关联资产
- 统一系统主机名与云资产名称的指标归属 - 修复SQLite更新参数顺序和自动注册冲突错误 - 增加Agent上报关联回归测试
This commit is contained in:
1 parent
827d9a31cd
commit
af98940211
2 files changed
+119
-40
No files matched your search
+41
-40
@@ -30,6 +30,24 @@ func NewAgentHandler(db *gorm.DB, cache ezredis.Cache, hub *MetricsHub) *AgentHa
|
|||||||
return &AgentHandler{db: db, cache: cache, hubMetrics: hub}
|
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 上报的指标数据
|
// ReceiveReport 接收 Agent 上报的指标数据
|
||||||
// POST /api/v1/agent/report
|
// POST /api/v1/agent/report
|
||||||
func (h *AgentHandler) ReceiveReport(c *gin.Context) {
|
func (h *AgentHandler) ReceiveReport(c *gin.Context) {
|
||||||
@@ -47,6 +65,12 @@ func (h *AgentHandler) ReceiveReport(c *gin.Context) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
matchedServer, matchIP, assetMatched := h.resolveAgentAsset(req.Hostname, req.IP, c.ClientIP())
|
||||||
|
canonicalHostname := req.Hostname
|
||||||
|
if assetMatched {
|
||||||
|
canonicalHostname = matchedServer.Hostname
|
||||||
|
}
|
||||||
|
|
||||||
// 序列化磁盘和端口信息
|
// 序列化磁盘和端口信息
|
||||||
diskJSON, _ := json.Marshal(req.Disks)
|
diskJSON, _ := json.Marshal(req.Disks)
|
||||||
portJSON, _ := json.Marshal(req.Ports)
|
portJSON, _ := json.Marshal(req.Ports)
|
||||||
@@ -61,7 +85,7 @@ func (h *AgentHandler) ReceiveReport(c *gin.Context) {
|
|||||||
|
|
||||||
// 存储到数据库
|
// 存储到数据库
|
||||||
report := model.AgentReport{
|
report := model.AgentReport{
|
||||||
Hostname: req.Hostname,
|
Hostname: canonicalHostname,
|
||||||
CPUUsage: req.CPU.UsagePercent,
|
CPUUsage: req.CPU.UsagePercent,
|
||||||
CPUIOwait: req.CPU.IOWait,
|
CPUIOwait: req.CPU.IOWait,
|
||||||
CPUCores: req.CPU.CoreCount,
|
CPUCores: req.CPU.CoreCount,
|
||||||
@@ -103,14 +127,6 @@ func (h *AgentHandler) ReceiveReport(c *gin.Context) {
|
|||||||
memMB := int64(memGB * 1024) // GB → MB
|
memMB := int64(memGB * 1024) // GB → MB
|
||||||
diskGB := int64(totalDiskGB + 0.5) // 39.99 → 40
|
diskGB := int64(totalDiskGB + 0.5) // 39.99 → 40
|
||||||
|
|
||||||
clientIP := c.ClientIP()
|
|
||||||
|
|
||||||
// 三级匹配:优先用 Agent 上报的 IP,其次用请求源 IP,最后用报文的 hostname
|
|
||||||
matchIP := req.IP
|
|
||||||
if matchIP == "" {
|
|
||||||
matchIP = clientIP
|
|
||||||
}
|
|
||||||
|
|
||||||
// 直接 SQL 更新 asset_servers 监控字段
|
// 直接 SQL 更新 asset_servers 监控字段
|
||||||
// 若 AgentVersion 非空则同步写入,否则保留数据库中已有版本号(防止旧版Agent空字段覆盖已设置的版本)
|
// 若 AgentVersion 非空则同步写入,否则保留数据库中已有版本号(防止旧版Agent空字段覆盖已设置的版本)
|
||||||
updateSQL := `UPDATE asset_servers SET
|
updateSQL := `UPDATE asset_servers SET
|
||||||
@@ -121,9 +137,7 @@ func (h *AgentHandler) ReceiveReport(c *gin.Context) {
|
|||||||
return `, agent_version = ?`
|
return `, agent_version = ?`
|
||||||
}
|
}
|
||||||
return ``
|
return ``
|
||||||
}() + `,
|
}()
|
||||||
hostname = CASE WHEN hostname != ? AND ip = ? THEN ? ELSE hostname END,
|
|
||||||
updated_at = ?`
|
|
||||||
args := []interface{}{
|
args := []interface{}{
|
||||||
req.CPU.UsagePercent, req.CPU.IOWait, req.Memory.UsagePercent, diskAvg,
|
req.CPU.UsagePercent, req.CPU.IOWait, req.Memory.UsagePercent, diskAvg,
|
||||||
req.CPU.CoreCount, memMB, diskGB,
|
req.CPU.CoreCount, memMB, diskGB,
|
||||||
@@ -132,10 +146,6 @@ func (h *AgentHandler) ReceiveReport(c *gin.Context) {
|
|||||||
if req.AgentVersion != "" {
|
if req.AgentVersion != "" {
|
||||||
args = append(args, req.AgentVersion)
|
args = append(args, req.AgentVersion)
|
||||||
}
|
}
|
||||||
args = append(args,
|
|
||||||
// CASE WHEN hostname != ? AND ip = ? THEN ? ELSE hostname END
|
|
||||||
req.Hostname, matchIP, req.Hostname,
|
|
||||||
)
|
|
||||||
if req.System != nil {
|
if req.System != nil {
|
||||||
if req.System.OSVersion != "" {
|
if req.System.OSVersion != "" {
|
||||||
updateSQL += ", os = ?"
|
updateSQL += ", os = ?"
|
||||||
@@ -150,29 +160,28 @@ func (h *AgentHandler) ReceiveReport(c *gin.Context) {
|
|||||||
args = append(args, req.System.Uptime)
|
args = append(args, req.System.Uptime)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
updateSQL += ", updated_at = ?"
|
||||||
args = append(args, time.Now())
|
args = append(args, time.Now())
|
||||||
|
if assetMatched {
|
||||||
|
updateSQL += " WHERE id = ?"
|
||||||
|
args = append(args, matchedServer.ID)
|
||||||
|
} else {
|
||||||
updateSQL += " WHERE hostname = ? OR ip = ?"
|
updateSQL += " WHERE hostname = ? OR ip = ?"
|
||||||
args = append(args, req.Hostname, matchIP)
|
args = append(args, req.Hostname, matchIP)
|
||||||
|
}
|
||||||
|
|
||||||
result := h.db.Exec(updateSQL, args...)
|
result := h.db.Exec(updateSQL, args...)
|
||||||
|
|
||||||
// 如果 hostname 和 IP 都没匹配到,尝试用 hostname 反向匹配 IP 字段
|
|
||||||
if result.RowsAffected == 0 && req.Hostname != matchIP {
|
|
||||||
updateSQL2 := updateSQL + " OR ip = ?"
|
|
||||||
args2 := append(args, req.Hostname)
|
|
||||||
result = h.db.Exec(updateSQL2, args2...)
|
|
||||||
}
|
|
||||||
|
|
||||||
// 如果还是没匹配到 —— 按 hostname 显式查重后再注册 // Logan.Li
|
// 如果还是没匹配到 —— 按 hostname 显式查重后再注册 // Logan.Li
|
||||||
// asset_servers.hostname 无 UNIQUE 索引,OnConflict 不可靠,故先 First 查重,
|
// asset_servers.hostname 无 UNIQUE 索引,OnConflict 不可靠,故先 First 查重,
|
||||||
// 避免 Agent 每次上报(10s)都新建记录导致资产越来越多。
|
// 避免 Agent 每次上报(10s)都新建记录导致资产越来越多。
|
||||||
if result.RowsAffected == 0 {
|
if result.Error == nil && result.RowsAffected == 0 {
|
||||||
osName := ""
|
osName := ""
|
||||||
if req.System != nil {
|
if req.System != nil {
|
||||||
osName = req.System.OSVersion
|
osName = req.System.OSVersion
|
||||||
}
|
}
|
||||||
var existing model.Server
|
var existing model.Server
|
||||||
if h.db.Where("hostname = ?", req.Hostname).First(&existing).Error == nil {
|
if h.db.Where("hostname = ?", req.Hostname).Take(&existing).Error == nil {
|
||||||
// 已自动注册过 → 仅更新实时指标,不重复创建
|
// 已自动注册过 → 仅更新实时指标,不重复创建
|
||||||
h.db.Model(&existing).Updates(map[string]interface{}{
|
h.db.Model(&existing).Updates(map[string]interface{}{
|
||||||
"ip": matchIP, "os": osName, "cpu": req.CPU.CoreCount,
|
"ip": matchIP, "os": osName, "cpu": req.CPU.CoreCount,
|
||||||
@@ -182,7 +191,7 @@ func (h *AgentHandler) ReceiveReport(c *gin.Context) {
|
|||||||
"disk_usage": diskAvg,
|
"disk_usage": diskAvg,
|
||||||
})
|
})
|
||||||
} else {
|
} else {
|
||||||
// 真正的新主机 → 注册(OnConflict 作为并发双保险)
|
// 真正的新主机 → 注册,hostname 无唯一索引时不使用 OnConflict。 // Logan.Li
|
||||||
newServer := model.Server{
|
newServer := model.Server{
|
||||||
Hostname: req.Hostname,
|
Hostname: req.Hostname,
|
||||||
IP: matchIP,
|
IP: matchIP,
|
||||||
@@ -200,15 +209,7 @@ func (h *AgentHandler) ReceiveReport(c *gin.Context) {
|
|||||||
MemoryUsage: req.Memory.UsagePercent,
|
MemoryUsage: req.Memory.UsagePercent,
|
||||||
DiskUsage: diskAvg,
|
DiskUsage: diskAvg,
|
||||||
}
|
}
|
||||||
if err := h.db.Clauses(clause.OnConflict{
|
if err := h.db.Create(&newServer).Error; err != nil {
|
||||||
Columns: []clause.Column{{Name: "hostname"}},
|
|
||||||
DoUpdates: clause.AssignmentColumns([]string{
|
|
||||||
"ip", "os", "cpu", "memory", "disk",
|
|
||||||
"status", "agent_status",
|
|
||||||
"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)
|
fmt.Printf("[agent] 自动注册主机 %q (IP: %s) 失败: %v\n", req.Hostname, matchIP, err)
|
||||||
} else {
|
} else {
|
||||||
fmt.Printf("[agent] 自动注册新主机: %q (IP: %s, CPU=%d核, 内存=%dMB, 磁盘=%dGB, OS=%q)\n",
|
fmt.Printf("[agent] 自动注册新主机: %q (IP: %s, CPU=%d核, 内存=%dMB, 磁盘=%dGB, OS=%q)\n",
|
||||||
@@ -221,7 +222,7 @@ func (h *AgentHandler) ReceiveReport(c *gin.Context) {
|
|||||||
if h.hubMetrics != nil && result.RowsAffected > 0 {
|
if h.hubMetrics != nil && result.RowsAffected > 0 {
|
||||||
h.hubMetrics.Broadcast(MetricsUpdate{
|
h.hubMetrics.Broadcast(MetricsUpdate{
|
||||||
Type: "metrics_update",
|
Type: "metrics_update",
|
||||||
Hostname: req.Hostname,
|
Hostname: canonicalHostname,
|
||||||
CPU: req.CPU.UsagePercent,
|
CPU: req.CPU.UsagePercent,
|
||||||
Memory: req.Memory.UsagePercent,
|
Memory: req.Memory.UsagePercent,
|
||||||
Disk: diskAvg,
|
Disk: diskAvg,
|
||||||
@@ -239,13 +240,13 @@ func (h *AgentHandler) ReceiveReport(c *gin.Context) {
|
|||||||
if h.cache != nil && h.cache.Capabilities().SortedSet {
|
if h.cache != nil && h.cache.Capabilities().SortedSet {
|
||||||
timestamp := float64(reportTime.Unix())
|
timestamp := float64(reportTime.Unix())
|
||||||
ttl := 7 * 24 * time.Hour
|
ttl := 7 * 24 * time.Hour
|
||||||
h.writeHotMetric(c, "metrics:cpu:"+req.Hostname, req.CPU.UsagePercent, timestamp, ttl, reportTime)
|
h.writeHotMetric(c, "metrics:cpu:"+canonicalHostname, req.CPU.UsagePercent, timestamp, ttl, reportTime)
|
||||||
h.writeHotMetric(c, "metrics:mem:"+req.Hostname, req.Memory.UsagePercent, timestamp, ttl, reportTime)
|
h.writeHotMetric(c, "metrics:mem:"+canonicalHostname, req.Memory.UsagePercent, timestamp, ttl, reportTime)
|
||||||
h.writeHotMetric(c, "metrics:load1:"+req.Hostname, req.CPU.Load1, timestamp, ttl, reportTime)
|
h.writeHotMetric(c, "metrics:load1:"+canonicalHostname, req.CPU.Load1, timestamp, ttl, reportTime)
|
||||||
}
|
}
|
||||||
|
|
||||||
// 保存成功 — 响应中携带探针配置给 Agent
|
// 保存成功 — 响应中携带探针配置给 Agent
|
||||||
resp := h.makeProbeConfigResponse(req.Hostname, 200, "ok", req.ProbeResults)
|
resp := h.makeProbeConfigResponse(canonicalHostname, 200, "ok", req.ProbeResults)
|
||||||
c.JSON(http.StatusOK, resp)
|
c.JSON(http.StatusOK, resp)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,78 @@
|
|||||||
|
package controller
|
||||||
|
|
||||||
|
import (
|
||||||
|
"bytes"
|
||||||
|
"net/http"
|
||||||
|
"net/http/httptest"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"ezops/model"
|
||||||
|
|
||||||
|
"github.com/gin-gonic/gin"
|
||||||
|
"github.com/glebarez/sqlite"
|
||||||
|
"gorm.io/gorm"
|
||||||
|
gormlogger "gorm.io/gorm/logger"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestReceiveReportMatchesExistingAssetByIP(t *testing.T) {
|
||||||
|
gin.SetMode(gin.TestMode)
|
||||||
|
db, err := gorm.Open(sqlite.Open("file:agent_report_match?mode=memory&cache=shared"), &gorm.Config{
|
||||||
|
Logger: gormlogger.Default.LogMode(gormlogger.Silent),
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("打开测试数据库失败: %v", err)
|
||||||
|
}
|
||||||
|
if err := db.AutoMigrate(&model.Server{}, &model.AgentReport{}, &model.HealthProbe{}); err != nil {
|
||||||
|
t.Fatalf("迁移测试表失败: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
server := model.Server{Hostname: "c-redis-cluster-3", IP: "172.19.16.59", Status: "running"}
|
||||||
|
if err := db.Create(&server).Error; err != nil {
|
||||||
|
t.Fatalf("创建测试资产失败: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
body := []byte(`{
|
||||||
|
"hostname":"VM-16-59-tencentos",
|
||||||
|
"ip":"172.19.16.59",
|
||||||
|
"agent_version":"v2.1.0",
|
||||||
|
"cpu":{"usage_percent":12.5,"iowait":0.5,"core_count":4},
|
||||||
|
"memory":{"total_gb":7.8,"used_gb":3.1,"available_gb":4.7,"usage_percent":39.7},
|
||||||
|
"disks":[{"mount":"/","total_gb":100,"used_gb":25,"usage_percent":25}],
|
||||||
|
"system":{"os_version":"TencentOS Server 4","kernel":"6.6.0","uptime":3600}
|
||||||
|
}`)
|
||||||
|
req := httptest.NewRequest(http.MethodPost, "/api/v1/agent/report", bytes.NewReader(body))
|
||||||
|
req.Header.Set("Content-Type", "application/json")
|
||||||
|
recorder := httptest.NewRecorder()
|
||||||
|
c, _ := gin.CreateTestContext(recorder)
|
||||||
|
c.Request = req
|
||||||
|
|
||||||
|
NewAgentHandler(db, nil, nil).ReceiveReport(c)
|
||||||
|
if recorder.Code != http.StatusOK {
|
||||||
|
t.Fatalf("上报返回码 = %d, body=%s", recorder.Code, recorder.Body.String())
|
||||||
|
}
|
||||||
|
|
||||||
|
var gotServer model.Server
|
||||||
|
if err := db.Take(&gotServer, server.ID).Error; err != nil {
|
||||||
|
t.Fatalf("读取资产失败: %v", err)
|
||||||
|
}
|
||||||
|
if gotServer.Hostname != server.Hostname || gotServer.AgentStatus != "online" {
|
||||||
|
t.Fatalf("资产关联错误: hostname=%q status=%q", gotServer.Hostname, gotServer.AgentStatus)
|
||||||
|
}
|
||||||
|
if gotServer.OS != "TencentOS Server 4" || gotServer.Kernel != "6.6.0" || gotServer.Uptime != 3600 {
|
||||||
|
t.Fatalf("系统字段更新错误: os=%q kernel=%q uptime=%d", gotServer.OS, gotServer.Kernel, gotServer.Uptime)
|
||||||
|
}
|
||||||
|
|
||||||
|
var report model.AgentReport
|
||||||
|
if err := db.Order("id DESC").Take(&report).Error; err != nil {
|
||||||
|
t.Fatalf("读取 Agent 报告失败: %v", err)
|
||||||
|
}
|
||||||
|
if report.Hostname != server.Hostname {
|
||||||
|
t.Fatalf("报告主机名未归一: got=%q want=%q", report.Hostname, server.Hostname)
|
||||||
|
}
|
||||||
|
|
||||||
|
var serverCount int64
|
||||||
|
db.Model(&model.Server{}).Count(&serverCount)
|
||||||
|
if serverCount != 1 {
|
||||||
|
t.Fatalf("上报后产生重复资产: count=%d", serverCount)
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in new issue
Block a user