diff --git a/internal/web/controller/agent.go b/internal/web/controller/agent.go index ae15599..2ffdf53 100644 --- a/internal/web/controller/agent.go +++ b/internal/web/controller/agent.go @@ -3,6 +3,7 @@ package controller import ( "crypto/rand" "crypto/sha256" + "database/sql" "encoding/hex" "encoding/json" "errors" @@ -146,46 +147,91 @@ func (a *AgentController) captureSeenIP(c *gin.Context, agent *model.AgentNode) if ip == nil || agent == nil { return } - updates := map[string]any{"seen_ip_at": time.Now().UnixMilli()} + now := time.Now().UnixMilli() + cols := []agentColumn{{"seen_ip_at", now}, {"updated_at", now}} if v4 := ip.To4(); v4 != nil { if agent.SeenIPv4 == v4.String() { return } - updates["seen_ipv4"] = v4.String() + cols = append(cols, agentColumn{"seen_ipv4", v4.String()}) } else { if agent.SeenIPv6 == ip.String() { return } - updates["seen_ipv6"] = ip.String() + cols = append(cols, agentColumn{"seen_ipv6", ip.String()}) } - if err := database.GetDB().Model(&model.AgentNode{}). - Where("id = ?", agent.Id).Updates(updates).Error; err != nil { + if err := execAgentColumns(uint(agent.Id), cols); err != nil { return } - if v, ok := updates["seen_ipv4"]; ok { - agent.SeenIPv4 = v.(string) + for _, c := range cols { + switch c.name { + case "seen_ipv4": + agent.SeenIPv4 = c.value.(string) + case "seen_ipv6": + agent.SeenIPv6 = c.value.(string) + } } - if v, ok := updates["seen_ipv6"]; ok { - agent.SeenIPv6 = v.(string) + agent.SeenIPAt = now +} + +// agentColumn is one column assignment for execAgentColumns. +type agentColumn struct { + name string + value any +} + +// execAgentColumns writes exactly the given columns of one +// agent_nodes row with a single raw UPDATE (ITEM-P3 M4). The ingest +// hot paths — heartbeats, traffic reports, probe fast-reports, for +// every agent every few seconds — previously paid gorm's statement +// build/clone/scan machinery per call; the baseline measured that +// family at roughly a third of the master's allocation churn. The +// column sets and values written are unchanged from the gorm calls +// this replaces, including updated_at, which gorm's autoUpdateTime +// stamped on every one of them. +func execAgentColumns(id uint, cols []agentColumn) error { + if len(cols) == 0 { + return nil } - agent.SeenIPAt = updates["seen_ip_at"].(int64) + sqlDB, err := database.GetDB().DB() + if err != nil { + return err + } + var sb strings.Builder + sb.WriteString("UPDATE agent_nodes SET ") + args := make([]any, 0, len(cols)+1) + for i, c := range cols { + if i > 0 { + sb.WriteString(", ") + } + sb.WriteString(c.name) + sb.WriteString(" = ?") + args = append(args, c.value) + } + sb.WriteString(" WHERE id = ?") + args = append(args, id) + _, err = sqlDB.Exec(sb.String(), args...) + return err } func (a *AgentController) touchAgent(id uint, name string) { - updates := map[string]any{"last_seen": time.Now().UnixMilli()} + now := time.Now().UnixMilli() + cols := []agentColumn{{"last_seen", now}, {"updated_at", now}} if name != "" { - updates["name"] = name + cols = append(cols, agentColumn{"name", name}) } - db := database.GetDB() - _ = db.Model(&model.AgentNode{}).Where("id = ?", id).Updates(updates).Error + _ = execAgentColumns(id, cols) // If bound to a node, refresh the node's heartbeat so the legacy node // list shows it online (the legacy probe can't reach agent-only nodes). - var nodeID *int - _ = db.Model(&model.AgentNode{}).Select("node_id").Where("id = ?", id).Scan(&nodeID).Error - if nodeID != nil { - now := time.Now().UnixMilli() - _ = db.Exec("UPDATE nodes SET last_heartbeat = ?, status = 'online' WHERE id = ?", now, *nodeID).Error + sqlDB, err := database.GetDB().DB() + if err != nil { + return } + var nodeID sql.NullInt64 + if err := sqlDB.QueryRow("SELECT node_id FROM agent_nodes WHERE id = ?", id).Scan(&nodeID); err != nil || !nodeID.Valid { + return + } + _ = database.GetDB().Exec("UPDATE nodes SET last_heartbeat = ?, status = 'online' WHERE id = ?", now, nodeID.Int64).Error } // register activates a pre-issued token. Tokens are issued bound to a node via @@ -233,35 +279,36 @@ type heartbeatTelemetry struct { // updateAgentTelemetry persists the reported telemetry fields onto the // agent row. Only fields actually present in the report are written. func (a *AgentController) updateAgentTelemetry(id uint, t *heartbeatTelemetry) { - updates := map[string]any{"telemetry_at": time.Now().UnixMilli()} + now := time.Now().UnixMilli() + cols := []agentColumn{{"telemetry_at", now}, {"updated_at", now}} if t.AgentVersion != nil { - updates["agent_version"] = *t.AgentVersion + cols = append(cols, agentColumn{"agent_version", *t.AgentVersion}) } if t.XrayVersion != nil { - updates["xray_version"] = *t.XrayVersion + cols = append(cols, agentColumn{"xray_version", *t.XrayVersion}) } if t.SingboxVersion != nil { - updates["singbox_version"] = *t.SingboxVersion + cols = append(cols, agentColumn{"singbox_version", *t.SingboxVersion}) } if t.UptimeSeconds != nil { - updates["uptime_secs"] = *t.UptimeSeconds + cols = append(cols, agentColumn{"uptime_secs", *t.UptimeSeconds}) } if t.LatencyMs != nil { - updates["latency_ms"] = *t.LatencyMs + cols = append(cols, agentColumn{"latency_ms", *t.LatencyMs}) } if t.CpuPercent != nil { - updates["cpu_pct"] = *t.CpuPercent + cols = append(cols, agentColumn{"cpu_pct", *t.CpuPercent}) } if t.MemPercent != nil { - updates["mem_pct"] = *t.MemPercent + cols = append(cols, agentColumn{"mem_pct", *t.MemPercent}) } if t.IPv4 != nil { - updates["ipv4"] = *t.IPv4 + cols = append(cols, agentColumn{"ipv4", *t.IPv4}) } if t.IPv6 != nil { - updates["ipv6"] = *t.IPv6 + cols = append(cols, agentColumn{"ipv6", *t.IPv6}) } - _ = database.GetDB().Model(&model.AgentNode{}).Where("id = ?", id).Updates(updates).Error + _ = execAgentColumns(id, cols) } // heartbeatProbeFacts / heartbeatProbeMetrics are the monitor-probe-style @@ -309,99 +356,101 @@ type heartbeatProbeMetrics struct { // metrics share one update so a heartbeat carries a single write; register // calls it with metrics == nil. func (a *AgentController) updateAgentProbe(id uint, f *heartbeatProbeFacts, m *heartbeatProbeMetrics) { - updates := map[string]any{} + var cols []agentColumn if f != nil { if f.Hostname != nil { - updates["probe_hostname"] = *f.Hostname + cols = append(cols, agentColumn{"probe_hostname", *f.Hostname}) } if f.OS != nil { - updates["probe_os"] = *f.OS + cols = append(cols, agentColumn{"probe_os", *f.OS}) } if f.Kernel != nil { - updates["probe_kernel"] = *f.Kernel + cols = append(cols, agentColumn{"probe_kernel", *f.Kernel}) } if f.Arch != nil { - updates["probe_arch"] = *f.Arch + cols = append(cols, agentColumn{"probe_arch", *f.Arch}) } if f.Virt != nil { - updates["probe_virt"] = *f.Virt + cols = append(cols, agentColumn{"probe_virt", *f.Virt}) } if f.CPUName != nil { - updates["probe_cpu_name"] = *f.CPUName + cols = append(cols, agentColumn{"probe_cpu_name", *f.CPUName}) } if f.CPUCores != nil { - updates["probe_cpu_cores"] = *f.CPUCores + cols = append(cols, agentColumn{"probe_cpu_cores", *f.CPUCores}) } if f.MemTotal != nil { - updates["probe_mem_total"] = *f.MemTotal + cols = append(cols, agentColumn{"probe_mem_total", *f.MemTotal}) } if f.SwapTotal != nil { - updates["probe_swap_total"] = *f.SwapTotal + cols = append(cols, agentColumn{"probe_swap_total", *f.SwapTotal}) } if f.DiskTotal != nil { - updates["probe_disk_total"] = *f.DiskTotal + cols = append(cols, agentColumn{"probe_disk_total", *f.DiskTotal}) } } if m != nil { - updates["probe_at"] = time.Now().UnixMilli() + cols = append(cols, agentColumn{"probe_at", time.Now().UnixMilli()}) if m.BootID != nil { - updates["probe_boot_id"] = *m.BootID + cols = append(cols, agentColumn{"probe_boot_id", *m.BootID}) } if m.Iface != nil { - updates["probe_iface"] = *m.Iface + cols = append(cols, agentColumn{"probe_iface", *m.Iface}) } if m.Load != nil { - updates["probe_load1"] = (*m.Load)[0] - updates["probe_load5"] = (*m.Load)[1] - updates["probe_load15"] = (*m.Load)[2] + cols = append(cols, + agentColumn{"probe_load1", (*m.Load)[0]}, + agentColumn{"probe_load5", (*m.Load)[1]}, + agentColumn{"probe_load15", (*m.Load)[2]}) } if m.MemTotal != nil { - updates["probe_mem_total"] = *m.MemTotal + cols = append(cols, agentColumn{"probe_mem_total", *m.MemTotal}) } if m.MemUsed != nil { - updates["probe_mem_used"] = *m.MemUsed + cols = append(cols, agentColumn{"probe_mem_used", *m.MemUsed}) } if m.SwapTotal != nil { - updates["probe_swap_total"] = *m.SwapTotal + cols = append(cols, agentColumn{"probe_swap_total", *m.SwapTotal}) } if m.SwapUsed != nil { - updates["probe_swap_used"] = *m.SwapUsed + cols = append(cols, agentColumn{"probe_swap_used", *m.SwapUsed}) } if m.DiskTotal != nil { - updates["probe_disk_total"] = *m.DiskTotal + cols = append(cols, agentColumn{"probe_disk_total", *m.DiskTotal}) } if m.DiskUsed != nil { - updates["probe_disk_used"] = *m.DiskUsed + cols = append(cols, agentColumn{"probe_disk_used", *m.DiskUsed}) } if m.NetRxTotal != nil { - updates["probe_net_rx_tot"] = *m.NetRxTotal + cols = append(cols, agentColumn{"probe_net_rx_tot", *m.NetRxTotal}) } if m.NetTxTotal != nil { - updates["probe_net_tx_tot"] = *m.NetTxTotal + cols = append(cols, agentColumn{"probe_net_tx_tot", *m.NetTxTotal}) } if m.NetRx != nil { - updates["probe_net_rx"] = *m.NetRx + cols = append(cols, agentColumn{"probe_net_rx", *m.NetRx}) } if m.NetTx != nil { - updates["probe_net_tx"] = *m.NetTx + cols = append(cols, agentColumn{"probe_net_tx", *m.NetTx}) } if m.TCP != nil { - updates["probe_tcp"] = *m.TCP + cols = append(cols, agentColumn{"probe_tcp", *m.TCP}) } if m.UDP != nil { - updates["probe_udp"] = *m.UDP + cols = append(cols, agentColumn{"probe_udp", *m.UDP}) } if m.Procs != nil { - updates["probe_procs"] = *m.Procs + cols = append(cols, agentColumn{"probe_procs", *m.Procs}) } if m.CPU != nil { - updates["probe_cpu"] = *m.CPU + cols = append(cols, agentColumn{"probe_cpu", *m.CPU}) } } - if len(updates) == 0 { + if len(cols) == 0 { return } - _ = database.GetDB().Model(&model.AgentNode{}).Where("id = ?", id).Updates(updates).Error + cols = append(cols, agentColumn{"updated_at", time.Now().UnixMilli()}) + _ = execAgentColumns(id, cols) } // probeSample flattens a heartbeat metrics block into the probe @@ -467,31 +516,40 @@ type heartbeatUpdateStatus struct { // clear the intent so the config poll stops re-offering it; the recorded // state stays visible on the node row. func (a *AgentController) applyAgentUpdateStatus(id uint, st *heartbeatUpdateStatus) { - db := database.GetDB() - var row model.AgentNode - if err := db.First(&row, id).Error; err != nil { + // Only the intent stamp is needed to gate the report (ITEM-P3 + // M4: a two-column read instead of a whole-row First). + sqlDB, err := database.GetDB().DB() + if err != nil { return } - if row.UpdateRequestedAt == 0 || st.RequestedAt != row.UpdateRequestedAt { + var requestedAt sql.NullInt64 + if err := sqlDB.QueryRow("SELECT update_requested_at FROM agent_nodes WHERE id = ?", id).Scan(&requestedAt); err != nil { return } - updates := map[string]any{"update_state_at": time.Now().UnixMilli()} + if !requestedAt.Valid || requestedAt.Int64 == 0 || st.RequestedAt != requestedAt.Int64 { + return + } + now := time.Now().UnixMilli() + cols := []agentColumn{{"update_state_at", now}, {"updated_at", now}} switch st.State { case "downloading", "swapped": - updates["update_state"] = st.State - updates["update_error"] = "" + cols = append(cols, + agentColumn{"update_state", st.State}, + agentColumn{"update_error", ""}) case "failed": - updates["update_state"] = "failed" - updates["update_error"] = st.Error - updates["update_requested_at"] = int64(0) + cols = append(cols, + agentColumn{"update_state", "failed"}, + agentColumn{"update_error", st.Error}, + agentColumn{"update_requested_at", int64(0)}) case "current": - updates["update_state"] = "current" - updates["update_error"] = "" - updates["update_requested_at"] = int64(0) + cols = append(cols, + agentColumn{"update_state", "current"}, + agentColumn{"update_error", ""}, + agentColumn{"update_requested_at", int64(0)}) default: return } - _ = db.Model(&model.AgentNode{}).Where("id = ?", id).Updates(updates).Error + _ = execAgentColumns(id, cols) } // confirmAgentUpdate marks a pending intent successful once the agent @@ -501,23 +559,31 @@ func (a *AgentController) confirmAgentUpdate(id uint, reportedVersion string) { if reportedVersion == "" { return } - db := database.GetDB() - var row model.AgentNode - if err := db.First(&row, id).Error; err != nil { + // Only the pending intent's stamp and target gate the confirm + // (ITEM-P3 M4: a two-column read instead of a whole-row First). + sqlDB, err := database.GetDB().DB() + if err != nil { return } - if row.UpdateRequestedAt == 0 || row.UpdateTarget == "" { + var requestedAt sql.NullInt64 + var target sql.NullString + if err := sqlDB.QueryRow("SELECT update_requested_at, update_target FROM agent_nodes WHERE id = ?", id).Scan(&requestedAt, &target); err != nil { return } - if cmp, ok := config.CompareDisplayVersion(reportedVersion, row.UpdateTarget); !ok || cmp < 0 { + if !requestedAt.Valid || requestedAt.Int64 == 0 || !target.Valid || target.String == "" { return } - _ = db.Model(&model.AgentNode{}).Where("id = ?", id).Updates(map[string]any{ - "update_state": "success", - "update_error": "", - "update_requested_at": int64(0), - "update_state_at": time.Now().UnixMilli(), - }).Error + if cmp, ok := config.CompareDisplayVersion(reportedVersion, target.String); !ok || cmp < 0 { + return + } + now := time.Now().UnixMilli() + _ = execAgentColumns(id, []agentColumn{ + {"update_state", "success"}, + {"update_error", ""}, + {"update_requested_at", int64(0)}, + {"update_state_at", now}, + {"updated_at", now}, + }) } // heartbeatProbeResult is one latency sample an agent piggybacks on