P3 M4: column-level raw updates on agent ingest hot paths

2026-10-23 21:59:05 +00:00
parent 5a7f89ee21
commit d70fd38978
1 changed files with 154 additions and 88 deletions
@@ -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