diff --git a/backend/controller/agent.go b/backend/controller/agent.go index a740b48..e11674b 100644 --- a/backend/controller/agent.go +++ b/backend/controller/agent.go @@ -30,6 +30,24 @@ func NewAgentHandler(db *gorm.DB, cache ezredis.Cache, hub *MetricsHub) *AgentHa 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) { @@ -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) portJSON, _ := json.Marshal(req.Ports) @@ -61,7 +85,7 @@ func (h *AgentHandler) ReceiveReport(c *gin.Context) { // 存储到数据库 report := model.AgentReport{ - Hostname: req.Hostname, + Hostname: canonicalHostname, CPUUsage: req.CPU.UsagePercent, CPUIOwait: req.CPU.IOWait, CPUCores: req.CPU.CoreCount, @@ -103,14 +127,6 @@ func (h *AgentHandler) ReceiveReport(c *gin.Context) { memMB := int64(memGB * 1024) // GB → MB 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 监控字段 // 若 AgentVersion 非空则同步写入，否则保留数据库中已有版本号（防止旧版Agent空字段覆盖已设置的版本） updateSQL := `UPDATE asset_servers SET @@ -121,9 +137,7 @@ func (h *AgentHandler) ReceiveReport(c *gin.Context) { return `, agent_version = ?` } return `` - }() + `, - hostname = CASE WHEN hostname != ? AND ip = ? THEN ? ELSE hostname END, - updated_at = ?` + }() args := []interface{}{ req.CPU.UsagePercent, req.CPU.IOWait, req.Memory.UsagePercent, diskAvg, req.CPU.CoreCount, memMB, diskGB, @@ -132,10 +146,6 @@ func (h *AgentHandler) ReceiveReport(c *gin.Context) { if 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.OSVersion != "" { updateSQL += ", os = ?" @@ -150,29 +160,28 @@ func (h *AgentHandler) ReceiveReport(c *gin.Context) { args = append(args, req.System.Uptime) } } + updateSQL += ", updated_at = ?" args = append(args, time.Now()) - updateSQL += " WHERE hostname = ? OR ip = ?" - args = append(args, req.Hostname, matchIP) + if assetMatched { + updateSQL += " WHERE id = ?" + args = append(args, matchedServer.ID) + } else { + updateSQL += " WHERE hostname = ? OR ip = ?" + args = append(args, req.Hostname, matchIP) + } 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 // asset_servers.hostname 无 UNIQUE 索引，OnConflict 不可靠，故先 First 查重， // 避免 Agent 每次上报（10s）都新建记录导致资产越来越多。 - if result.RowsAffected == 0 { + if result.Error == nil && result.RowsAffected == 0 { osName := "" if req.System != nil { osName = req.System.OSVersion } 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{}{ "ip": matchIP, "os": osName, "cpu": req.CPU.CoreCount, @@ -182,7 +191,7 @@ func (h *AgentHandler) ReceiveReport(c *gin.Context) { "disk_usage": diskAvg, }) } else { - // 真正的新主机 → 注册（OnConflict 作为并发双保险） + // 真正的新主机 → 注册，hostname 无唯一索引时不使用 OnConflict。 // Logan.Li newServer := model.Server{ Hostname: req.Hostname, IP: matchIP, @@ -200,15 +209,7 @@ func (h *AgentHandler) ReceiveReport(c *gin.Context) { MemoryUsage: req.Memory.UsagePercent, DiskUsage: diskAvg, } - if err := h.db.Clauses(clause.OnConflict{ - 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 { + 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", @@ -221,7 +222,7 @@ func (h *AgentHandler) ReceiveReport(c *gin.Context) { if h.hubMetrics != nil && result.RowsAffected > 0 { h.hubMetrics.Broadcast(MetricsUpdate{ Type: "metrics_update", - Hostname: req.Hostname, + Hostname: canonicalHostname, CPU: req.CPU.UsagePercent, Memory: req.Memory.UsagePercent, Disk: diskAvg, @@ -239,13 +240,13 @@ func (h *AgentHandler) ReceiveReport(c *gin.Context) { if h.cache != nil && h.cache.Capabilities().SortedSet { timestamp := float64(reportTime.Unix()) ttl := 7 * 24 * time.Hour - 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) + 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(req.Hostname, 200, "ok", req.ProbeResults) + resp := h.makeProbeConfigResponse(canonicalHostname, 200, "ok", req.ProbeResults) c.JSON(http.StatusOK, resp) } diff --git a/backend/controller/agent_report_test.go b/backend/controller/agent_report_test.go new file mode 100644 index 0000000..e0e35c9 --- /dev/null +++ b/backend/controller/agent_report_test.go @@ -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) + } +}