feat(telemetry): agent 心跳遥测上报+节点列表覆写(第13条)

agent 每 60s 心跳捎带遥测: agentVersion/xrayVersion(按 mtime 缓存重读)/
uptime/最近一次主控请求 RTT/cpu%/mem%/公网 v4+v6(公网优先, ULA/私网兜底);
主控心跳处理按指针语义只存上报字段, 旧 agent 无遥测行为不变;
agent_nodes 加法迁移存遥测, 节点列表对 agent 行覆写真值,
panelVersion 显示 "agent <版本>", NodeView 新增 ipv4/ipv6。
This commit is contained in:
Muse
2026-10-22 11:33:56 +00:00
parent 84d7b85be7
commit 3536d4174e
10 changed files with 604 additions and 6 deletions
+28 -4
View File
@@ -14,6 +14,8 @@ import (
"net/http" "net/http"
"os" "os"
"path/filepath" "path/filepath"
"sync"
"sync/atomic"
"time" "time"
"github.com/mhsanaei/3x-ui/v3/internal/config" "github.com/mhsanaei/3x-ui/v3/internal/config"
@@ -64,6 +66,15 @@ type Agent struct {
xrayAPI xray.XrayAPI xrayAPI xray.XrayAPI
apiPort int apiPort int
baselineLoaded bool baselineLoaded bool
// Heartbeat telemetry state (P13): last measured master-request RTT
// and the mtime-keyed xray version cache.
lastRTTms atomic.Int64
hasRTT atomic.Bool
xrayVerMu sync.Mutex
xrayVer string
xrayVerFor string // binary path the cached version was read from
xrayVerMod time.Time
} }
// New creates a new agent instance // New creates a new agent instance
@@ -110,7 +121,7 @@ func (a *Agent) Register() error {
req.Header.Set("Content-Type", "application/json") req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", "Bearer "+a.config.Token) req.Header.Set("Authorization", "Bearer "+a.config.Token)
resp, err := a.httpClient.Do(req) resp, err := a.do(req)
if err != nil { if err != nil {
return fmt.Errorf("register failed: %w", err) return fmt.Errorf("register failed: %w", err)
} }
@@ -241,12 +252,25 @@ func (a *Agent) Stop() {
close(a.stopCh) close(a.stopCh)
} }
// do performs one master request and records its round-trip time as the
// node->master latency sample reported by the next heartbeat (P13).
func (a *Agent) do(req *http.Request) (*http.Response, error) {
start := time.Now()
resp, err := a.httpClient.Do(req)
if err == nil {
a.lastRTTms.Store(time.Since(start).Milliseconds())
a.hasRTT.Store(true)
}
return resp, err
}
// sendHeartbeat sends a heartbeat to the master // sendHeartbeat sends a heartbeat to the master
func (a *Agent) sendHeartbeat() error { func (a *Agent) sendHeartbeat() error {
url := fmt.Sprintf("%s/panel/api/agent/heartbeat", a.config.MasterURL) url := fmt.Sprintf("%s/panel/api/agent/heartbeat", a.config.MasterURL)
payload := map[string]interface{}{ payload := map[string]interface{}{
"token": a.config.Token, "token": a.config.Token,
"timestamp": time.Now().Unix(), "timestamp": time.Now().Unix(),
"telemetry": a.collectTelemetry(),
} }
body, _ := json.Marshal(payload) body, _ := json.Marshal(payload)
req, err := http.NewRequest("POST", url, bytes.NewReader(body)) req, err := http.NewRequest("POST", url, bytes.NewReader(body))
@@ -256,7 +280,7 @@ func (a *Agent) sendHeartbeat() error {
req.Header.Set("Content-Type", "application/json") req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", "Bearer "+a.config.Token) req.Header.Set("Authorization", "Bearer "+a.config.Token)
resp, err := a.httpClient.Do(req) resp, err := a.do(req)
if err != nil { if err != nil {
return err return err
} }
@@ -276,7 +300,7 @@ func (a *Agent) fetchAndApplyConfig() error {
} }
req.Header.Set("Authorization", "Bearer "+a.config.Token) req.Header.Set("Authorization", "Bearer "+a.config.Token)
resp, err := a.httpClient.Do(req) resp, err := a.do(req)
if err != nil { if err != nil {
return err return err
} }
@@ -509,7 +533,7 @@ func (a *Agent) postTrafficReport(report trafficReport) error {
req.Header.Set("Content-Type", "application/json") req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", "Bearer "+a.config.Token) req.Header.Set("Authorization", "Bearer "+a.config.Token)
resp, err := a.httpClient.Do(req) resp, err := a.do(req)
if err != nil { if err != nil {
return err return err
} }
+203
View File
@@ -0,0 +1,203 @@
package agent
import (
"context"
"math"
"net"
"os"
"os/exec"
"path/filepath"
"strings"
"time"
"github.com/mhsanaei/3x-ui/v3/internal/config"
"github.com/mhsanaei/3x-ui/v3/internal/xray"
"github.com/shirou/gopsutil/v4/cpu"
"github.com/shirou/gopsutil/v4/host"
"github.com/shirou/gopsutil/v4/mem"
)
// telemetryPayload is the best-effort host/runtime snapshot piggybacked on
// every heartbeat (P13). Pointer fields are omitted when their collector
// failed, so the master keeps the last stored value instead of a zero.
type telemetryPayload struct {
AgentVersion string `json:"agentVersion"`
XrayVersion string `json:"xrayVersion,omitempty"`
UptimeSeconds *uint64 `json:"uptimeSeconds,omitempty"`
LatencyMs *int64 `json:"latencyMs,omitempty"`
CpuPercent *float64 `json:"cpuPercent,omitempty"`
MemPercent *float64 `json:"memPercent,omitempty"`
IPv4 string `json:"ipv4,omitempty"`
IPv6 string `json:"ipv6,omitempty"`
}
// collectTelemetry samples every source independently; a failing source
// only blanks its own field, never the whole report.
func (a *Agent) collectTelemetry() *telemetryPayload {
t := &telemetryPayload{AgentVersion: config.GetBaseVersion()}
if v, ok := a.cachedXrayVersion(); ok {
t.XrayVersion = v
}
if up, err := host.Uptime(); err == nil {
t.UptimeSeconds = &up
}
if pcts, err := cpu.Percent(0, false); err == nil && len(pcts) > 0 {
v := math.Round(pcts[0]*10) / 10
t.CpuPercent = &v
}
if vm, err := mem.VirtualMemory(); err == nil {
v := math.Round(vm.UsedPercent*10) / 10
t.MemPercent = &v
}
if a.hasRTT.Load() {
v := a.lastRTTms.Load()
t.LatencyMs = &v
}
t.IPv4, t.IPv6 = hostPublicIPs()
return t
}
// cachedXrayVersion returns the managed xray binary's version, re-reading
// it only when the binary file changed (path or mtime) since the last read,
// so a swapped xray is picked up without restarting the agent.
func (a *Agent) cachedXrayVersion() (string, bool) {
path := xrayBinaryPath()
if path == "" {
return "", false
}
fi, err := os.Stat(path)
if err != nil {
return "", false
}
a.xrayVerMu.Lock()
defer a.xrayVerMu.Unlock()
if a.xrayVer != "" && a.xrayVerFor == path && fi.ModTime().Equal(a.xrayVerMod) {
return a.xrayVer, true
}
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel()
out, err := exec.CommandContext(ctx, path, "version").Output()
if err != nil {
return "", false
}
v, ok := parseXrayVersion(string(out))
if !ok {
return "", false
}
a.xrayVer, a.xrayVerFor, a.xrayVerMod = v, path, fi.ModTime()
return v, true
}
// xrayBinaryPath resolves the managed xray binary: the arch-specific name
// the process manager uses, falling back to a plain "xray" in the same bin
// folder (mirrors ensureXrayBinary).
func xrayBinaryPath() string {
if p := xray.GetBinaryPath(); fileExists(p) {
return p
}
if p := filepath.Join(config.GetBinFolderPath(), "xray"); fileExists(p) {
return p
}
return ""
}
func fileExists(p string) bool {
fi, err := os.Stat(p)
return err == nil && !fi.IsDir()
}
// parseXrayVersion extracts the version from the first line of
// `xray version` output, e.g. "Xray 26.6.27 (Xray, Penetrates Everything.)".
func parseXrayVersion(out string) (string, bool) {
line := out
if i := strings.IndexByte(line, '\n'); i >= 0 {
line = line[:i]
}
fields := strings.Fields(line)
if len(fields) < 2 || !strings.EqualFold(fields[0], "xray") {
return "", false
}
v := fields[1]
if v == "" || v[0] < '0' || v[0] > '9' {
return "", false
}
return v, true
}
// hostPublicIPs picks one IPv4 and one IPv6 address from the host's active
// interfaces. Loopback, link-local, multicast and down interfaces are
// excluded; globally routable addresses win over private/ULA ones, which
// are only used when no public address exists.
func hostPublicIPs() (string, string) {
ifaces, err := net.Interfaces()
if err != nil {
return "", ""
}
var ips []net.IP
for _, ifc := range ifaces {
if ifc.Flags&net.FlagUp == 0 || ifc.Flags&net.FlagLoopback != 0 {
continue
}
addrs, err := ifc.Addrs()
if err != nil {
continue
}
for _, addr := range addrs {
var ip net.IP
switch v := addr.(type) {
case *net.IPNet:
ip = v.IP
case *net.IPAddr:
ip = v.IP
}
if ip != nil {
ips = append(ips, ip)
}
}
}
return chooseHostIPs(ips)
}
// chooseHostIPs applies the public-over-private preference to a candidate
// address list. Kept pure for testing.
func chooseHostIPs(ips []net.IP) (string, string) {
var v4Pub, v4Priv, v6Pub, v6Priv net.IP
for _, ip := range ips {
if ip == nil || ip.IsLoopback() || ip.IsUnspecified() ||
ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast() || ip.IsMulticast() {
continue
}
if ip4 := ip.To4(); ip4 != nil {
if ip4.IsPrivate() {
if v4Priv == nil {
v4Priv = ip4
}
} else if v4Pub == nil {
v4Pub = ip4
}
continue
}
ip6 := ip.To16()
if ip6 == nil {
continue
}
if ip6.IsPrivate() { // unique-local fc00::/7
if v6Priv == nil {
v6Priv = ip6
}
} else if v6Pub == nil {
v6Pub = ip6
}
}
pick := func(pub, priv net.IP) string {
if pub != nil {
return pub.String()
}
if priv != nil {
return priv.String()
}
return ""
}
return pick(v4Pub, v4Priv), pick(v6Pub, v6Priv)
}
+75
View File
@@ -0,0 +1,75 @@
package agent
import (
"net"
"testing"
"github.com/mhsanaei/3x-ui/v3/internal/config"
)
func TestParseXrayVersion(t *testing.T) {
cases := []struct {
in string
want string
ok bool
}{
{"Xray 26.6.27 (Xray, Penetrates Everything.) 45cf289 (go1.26.4 linux/amd64)\nA unified platform for anti-censorship.", "26.6.27", true},
{"Xray 1.8.4", "1.8.4", true},
{"V2Ray 4.45.4 (V2Fly, a community-driven edition of V2Ray.)", "", false},
{"", "", false},
{"Xray", "", false},
{"not a version banner", "", false},
{"Xray (no version here)", "", false},
}
for _, c := range cases {
got, ok := parseXrayVersion(c.in)
if got != c.want || ok != c.ok {
t.Errorf("parseXrayVersion(%q) = %q,%v want %q,%v", c.in, got, ok, c.want, c.ok)
}
}
}
func TestChooseHostIPs(t *testing.T) {
mk := func(ss ...string) []net.IP {
var out []net.IP
for _, s := range ss {
out = append(out, net.ParseIP(s))
}
return out
}
cases := []struct {
name string
in []net.IP
v4 string
v6 string
}{
{"public wins over private", mk("10.0.0.5", "8.8.4.4", "2001:db8::1", "fd00::1"), "8.8.4.4", "2001:db8::1"},
{"private fallback", mk("192.168.1.2", "fd00::9"), "192.168.1.2", "fd00::9"},
{"loopback and link-local excluded", mk("127.0.0.1", "169.254.1.1", "fe80::1", "::1"), "", ""},
{"empty", nil, "", ""},
{"multicast excluded", mk("224.0.0.1", "ff02::1"), "", ""},
}
for _, c := range cases {
v4, v6 := chooseHostIPs(c.in)
if v4 != c.v4 || v6 != c.v6 {
t.Errorf("%s: chooseHostIPs = %q,%q want %q,%q", c.name, v4, v6, c.v4, c.v6)
}
}
}
// TestCollectTelemetryShape pins the always-on fields on any Linux host:
// the agent's own build version and a positive host uptime. Source-specific
// fields (xray version, IPs) are environment-dependent and not asserted.
func TestCollectTelemetryShape(t *testing.T) {
a := New(&Config{MasterURL: "http://127.0.0.1:1", Token: "t", NodeName: "n"})
tel := a.collectTelemetry()
if tel.AgentVersion != config.GetBaseVersion() || tel.AgentVersion == "" {
t.Errorf("AgentVersion = %q, want config.GetBaseVersion() %q", tel.AgentVersion, config.GetBaseVersion())
}
if tel.UptimeSeconds == nil || *tel.UptimeSeconds == 0 {
t.Errorf("UptimeSeconds = %v, want non-nil positive", tel.UptimeSeconds)
}
if tel.MemPercent == nil || *tel.MemPercent < 0 || *tel.MemPercent > 100 {
t.Errorf("MemPercent = %v, want within [0,100]", tel.MemPercent)
}
}
+16
View File
@@ -22,4 +22,20 @@ type AgentNode struct {
LastSeen int64 `json:"lastSeen"` LastSeen int64 `json:"lastSeen"`
CreatedAt int64 `json:"createdAt" gorm:"autoCreateTime:milli"` CreatedAt int64 `json:"createdAt" gorm:"autoCreateTime:milli"`
UpdatedAt int64 `json:"updatedAt" gorm:"autoUpdateTime:milli"` UpdatedAt int64 `json:"updatedAt" gorm:"autoUpdateTime:milli"`
// Heartbeat telemetry (P13): the agent piggybacks host/runtime facts on
// its 60s heartbeat and the panel overlays them onto the node list row.
// All fields are best-effort: an agent that never reported telemetry has
// TelemetryAt == 0 and the UI keeps showing "-" for its row.
AgentVersion string `json:"agentVersion"`
XrayVersion string `json:"xrayVersion"`
UptimeSecs uint64 `json:"uptimeSecs"`
LatencyMs int `json:"latencyMs"`
CpuPct float64 `json:"cpuPct"`
MemPct float64 `json:"memPct"`
IPv4 string `json:"ipv4"`
IPv6 string `json:"ipv6"`
// TelemetryAt is the unix-milli time of the last heartbeat that carried
// a telemetry object.
TelemetryAt int64 `json:"telemetryAt"`
} }
+55
View File
@@ -117,13 +117,68 @@ func (a *AgentController) register(c *gin.Context) {
jsonObj(c, gin.H{"message": "registered"}, nil) jsonObj(c, gin.H{"message": "registered"}, nil)
} }
// heartbeatTelemetry is the optional telemetry object an agent piggybacks
// on its heartbeat (P13). Every field is a pointer: a missing field means
// "not reported this round" and must not overwrite the stored value. Older
// agents send no telemetry object at all and are handled exactly as before.
type heartbeatTelemetry struct {
AgentVersion *string `json:"agentVersion"`
XrayVersion *string `json:"xrayVersion"`
UptimeSeconds *uint64 `json:"uptimeSeconds"`
LatencyMs *int `json:"latencyMs"`
CpuPercent *float64 `json:"cpuPercent"`
MemPercent *float64 `json:"memPercent"`
IPv4 *string `json:"ipv4"`
IPv6 *string `json:"ipv6"`
}
// 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()}
if t.AgentVersion != nil {
updates["agent_version"] = *t.AgentVersion
}
if t.XrayVersion != nil {
updates["xray_version"] = *t.XrayVersion
}
if t.UptimeSeconds != nil {
updates["uptime_secs"] = *t.UptimeSeconds
}
if t.LatencyMs != nil {
updates["latency_ms"] = *t.LatencyMs
}
if t.CpuPercent != nil {
updates["cpu_pct"] = *t.CpuPercent
}
if t.MemPercent != nil {
updates["mem_pct"] = *t.MemPercent
}
if t.IPv4 != nil {
updates["ipv4"] = *t.IPv4
}
if t.IPv6 != nil {
updates["ipv6"] = *t.IPv6
}
_ = database.GetDB().Model(&model.AgentNode{}).Where("id = ?", id).Updates(updates).Error
}
func (a *AgentController) heartbeat(c *gin.Context) { func (a *AgentController) heartbeat(c *gin.Context) {
agent, ok := a.findAgent(bearerToken(c)) agent, ok := a.findAgent(bearerToken(c))
if !ok { if !ok {
c.JSON(401, gin.H{"message": "unauthorized"}) c.JSON(401, gin.H{"message": "unauthorized"})
return return
} }
// Telemetry is optional and best-effort: a body that fails to parse
// (or no body at all) must never fail the heartbeat itself.
var req struct {
Telemetry *heartbeatTelemetry `json:"telemetry"`
}
_ = c.ShouldBindJSON(&req)
a.touchAgent(uint(agent.Id), "") a.touchAgent(uint(agent.Id), "")
if req.Telemetry != nil {
a.updateAgentTelemetry(uint(agent.Id), req.Telemetry)
}
jsonObj(c, gin.H{"message": "ok"}, nil) jsonObj(c, gin.H{"message": "ok"}, nil)
} }
@@ -0,0 +1,96 @@
package controller
import (
"bytes"
"net/http"
"net/http/httptest"
"path/filepath"
"testing"
"github.com/gin-gonic/gin"
"github.com/mhsanaei/3x-ui/v3/internal/database"
"github.com/mhsanaei/3x-ui/v3/internal/database/model"
)
// P13: the heartbeat optionally carries a telemetry object. Reported fields
// are persisted onto the agent row; a heartbeat without telemetry (older
// agents) must not clear previously stored values, and a partial report
// only overwrites the fields it carries.
func TestAgentHeartbeatTelemetry(t *testing.T) {
gin.SetMode(gin.TestMode)
dbDir := t.TempDir()
t.Setenv("XUI_DB_FOLDER", dbDir)
if err := database.InitDB(filepath.Join(dbDir, "x-ui.db")); err != nil {
t.Fatalf("InitDB: %v", err)
}
t.Cleanup(func() { _ = database.CloseDB() })
db := database.GetDB()
node := &model.Node{Name: "telemetry-node", Address: "10.0.9.1"}
if err := db.Create(node).Error; err != nil {
t.Fatalf("seed node: %v", err)
}
const token = "telemetry-test-token"
agent := &model.AgentNode{
TokenHash: hashAgentToken(token),
TokenPrefix: "telemetry-",
Name: "telemetry-node",
NodeID: &node.Id,
}
if err := db.Create(agent).Error; err != nil {
t.Fatalf("seed agent: %v", err)
}
engine := gin.New()
NewAgentController(engine.Group("/panel/api/agent"))
post := func(body string) int {
req := httptest.NewRequest(http.MethodPost, "/panel/api/agent/heartbeat", bytes.NewBufferString(body))
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", "Bearer "+token)
rec := httptest.NewRecorder()
engine.ServeHTTP(rec, req)
return rec.Code
}
load := func() model.AgentNode {
var row model.AgentNode
if err := db.First(&row, agent.Id).Error; err != nil {
t.Fatalf("reload agent: %v", err)
}
return row
}
full := `{"token":"` + token + `","timestamp":1,"telemetry":{` +
`"agentVersion":"3.7.0-agent.1","xrayVersion":"26.6.27","uptimeSeconds":90061,` +
`"latencyMs":12,"cpuPercent":3.5,"memPercent":41.2,` +
`"ipv4":"203.0.113.7","ipv6":"2001:db8::7"}}`
if code := post(full); code != http.StatusOK {
t.Fatalf("heartbeat with telemetry: status %d", code)
}
row := load()
if row.AgentVersion != "3.7.0-agent.1" || row.XrayVersion != "26.6.27" ||
row.UptimeSecs != 90061 || row.LatencyMs != 12 ||
row.CpuPct != 3.5 || row.MemPct != 41.2 ||
row.IPv4 != "203.0.113.7" || row.IPv6 != "2001:db8::7" || row.TelemetryAt == 0 {
t.Fatalf("telemetry not stored: %+v", row)
}
// No telemetry object (old agent): stored values survive.
if code := post(`{"token":"` + token + `","timestamp":2}`); code != http.StatusOK {
t.Fatalf("bare heartbeat: status %d", code)
}
row = load()
if row.XrayVersion != "26.6.27" || row.IPv4 != "203.0.113.7" || row.UptimeSecs != 90061 {
t.Fatalf("bare heartbeat cleared telemetry: %+v", row)
}
// Partial report: only the carried field changes.
if code := post(`{"token":"` + token + `","timestamp":3,"telemetry":{"cpuPercent":9.25}}`); code != http.StatusOK {
t.Fatalf("partial heartbeat: status %d", code)
}
row = load()
if row.CpuPct != 9.25 || row.MemPct != 41.2 || row.LatencyMs != 12 {
t.Fatalf("partial telemetry merged wrong: %+v", row)
}
}
+3 -1
View File
@@ -362,7 +362,9 @@ func (s *NodeService) GetViewById(id int) (*NodeView, error) {
return nil, err return nil, err
} }
view := toNodeView(n) view := toNodeView(n)
annotateNodeTypes([]*NodeView{view}) views := []*NodeView{view}
annotateNodeTypes(views)
annotateAgentTelemetry(views)
return view, nil return view, nil
} }
@@ -0,0 +1,72 @@
package service
import (
"testing"
"github.com/mhsanaei/3x-ui/v3/internal/database"
"github.com/mhsanaei/3x-ui/v3/internal/database/model"
)
// P13: annotateAgentTelemetry overlays the agent-reported heartbeat
// telemetry onto agent node views only — panel version becomes
// "agent <version>", xray version/uptime/latency/CPU/mem/IPs take the
// reported values. Traditional rows and never-reporting agents are
// left untouched.
func TestAnnotateAgentTelemetry(t *testing.T) {
setupConflictDB(t)
db := database.GetDB()
agentNode := &model.Node{Id: 7401, Name: "agent-node", Address: ""}
tradNode := &model.Node{Id: 7402, Name: "trad-node", Address: "198.51.100.2", PanelVersion: "3.4.2", XrayVersion: "26.6.27"}
silentNode := &model.Node{Id: 7403, Name: "silent-agent-node", Address: ""}
for _, n := range []*model.Node{agentNode, tradNode, silentNode} {
if err := db.Create(n).Error; err != nil {
t.Fatalf("create node %d: %v", n.Id, err)
}
}
reporting := &model.AgentNode{
TokenHash: "hash-reporting",
Name: "agent-node", NodeID: &agentNode.Id,
AgentVersion: "3.7.0-agent.1", XrayVersion: "26.6.27",
UptimeSecs: 90061, LatencyMs: 12, CpuPct: 3.5, MemPct: 41.2,
IPv4: "203.0.113.7", IPv6: "2001:db8::7", TelemetryAt: 1700000000000,
}
silent := &model.AgentNode{TokenHash: "hash-silent", Name: "silent-agent-node", NodeID: &silentNode.Id}
// The traditional node has no agent row at all — by the codebase's
// definition (any binding makes a node agent-managed) it must pass
// through the overlay untouched.
for _, ag := range []*model.AgentNode{reporting, silent} {
if err := db.Create(ag).Error; err != nil {
t.Fatalf("create agent row: %v", err)
}
}
views := toNodeViews([]*model.Node{agentNode, tradNode, silentNode})
annotateNodeTypes(views)
annotateAgentTelemetry(views)
byID := map[int]*NodeView{}
for _, v := range views {
byID[v.Id] = v
}
av := byID[7401]
if av.NodeType != "agent" {
t.Fatalf("node 7401 type = %q, want agent", av.NodeType)
}
if av.XrayVersion != "26.6.27" || av.PanelVersion != "agent 3.7.0-agent.1" ||
av.UptimeSecs != 90061 || av.LatencyMs != 12 ||
av.CpuPct != 3.5 || av.MemPct != 41.2 ||
av.Ipv4 != "203.0.113.7" || av.Ipv6 != "2001:db8::7" {
t.Fatalf("agent view not overlaid: %+v", av)
}
tv := byID[7402]
if tv.PanelVersion != "3.4.2" || tv.XrayVersion != "26.6.27" || tv.Ipv4 != "" || tv.Ipv6 != "" {
t.Fatalf("traditional view mutated: %+v", tv)
}
sv := byID[7403]
if sv.XrayVersion != "" || sv.PanelVersion != "" || sv.Ipv4 != "" || sv.UptimeSecs != 0 {
t.Fatalf("silent agent view should stay empty: %+v", sv)
}
}
+55 -1
View File
@@ -31,7 +31,13 @@ type NodeView struct {
// agent binding, not stored on the node row; read paths that serve the // agent binding, not stored on the node row; read paths that serve the
// panel UI annotate it via annotateNodeTypes. Empty on views built // panel UI annotate it via annotateNodeTypes. Empty on views built
// outside those paths. // outside those paths.
NodeType string `json:"nodeType,omitempty" example:"traditional"` NodeType string `json:"nodeType,omitempty" example:"traditional"`
// Ipv4/Ipv6 carry the agent-reported host addresses for agent nodes
// (P13). The nodes page renders them as plain text on agent rows
// instead of the panel-URL link used for traditional nodes. Empty for
// traditional nodes and for agents that never reported telemetry.
Ipv4 string `json:"ipv4,omitempty" example:"203.0.113.7"`
Ipv6 string `json:"ipv6,omitempty" example:"2001:db8::7"`
Guid string `json:"guid" example:"node-guid"` Guid string `json:"guid" example:"node-guid"`
Status string `json:"status" example:"online"` Status string `json:"status" example:"online"`
LastHeartbeat int64 `json:"lastHeartbeat" example:"1700000000"` LastHeartbeat int64 `json:"lastHeartbeat" example:"1700000000"`
@@ -156,6 +162,54 @@ func annotateNodeTypes(views []*NodeView) {
} }
} }
// annotateAgentTelemetry overlays the latest heartbeat telemetry reported
// by each node's bound agent onto its view (P13): xray version, the
// "agent <version>" panel-version label, host uptime, node->master latency,
// CPU/memory percentages and the host IPv4/IPv6 pair. Views for traditional
// nodes (and agent nodes whose agent never reported telemetry) are left
// exactly as built. Must run after annotateNodeTypes.
func annotateAgentTelemetry(views []*NodeView) {
if len(views) == 0 {
return
}
var agents []model.AgentNode
// Latest reporter wins when more than one token is bound to a node.
_ = database.GetDB().Model(&model.AgentNode{}).
Where("node_id IS NOT NULL").
Order("last_seen DESC").
Find(&agents).Error
byNode := make(map[int]*model.AgentNode, len(agents))
for i := range agents {
if agents[i].NodeID == nil {
continue
}
if _, ok := byNode[*agents[i].NodeID]; !ok {
byNode[*agents[i].NodeID] = &agents[i]
}
}
for _, v := range views {
if v.NodeType != "agent" {
continue
}
ag, ok := byNode[v.Id]
if !ok || ag.TelemetryAt == 0 {
continue
}
if ag.XrayVersion != "" {
v.XrayVersion = ag.XrayVersion
}
if ag.AgentVersion != "" {
v.PanelVersion = "agent " + ag.AgentVersion
}
v.UptimeSecs = ag.UptimeSecs
v.LatencyMs = ag.LatencyMs
v.CpuPct = ag.CpuPct
v.MemPct = ag.MemPct
v.Ipv4 = ag.IPv4
v.Ipv6 = ag.IPv6
}
}
// NodeMutationRequest is the node write/probe contract. ApiToken is accepted // NodeMutationRequest is the node write/probe contract. ApiToken is accepted
// only as input. On update, nil means keep the stored token; replacement and // only as input. On update, nil means keep the stored token; replacement and
// clearing are explicit and mutually exclusive. // clearing are explicit and mutually exclusive.
+1
View File
@@ -164,6 +164,7 @@ func (s *NodeService) GetNodeTreeView() ([]*NodeView, error) {
} }
views := toNodeViews(nodes) views := toNodeViews(nodes)
annotateNodeTypes(views) annotateNodeTypes(views)
annotateAgentTelemetry(views)
return views, nil return views, nil
} }