diff --git a/internal/agent/agent.go b/internal/agent/agent.go index c70babc..8e8ceb7 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -14,6 +14,8 @@ import ( "net/http" "os" "path/filepath" + "sync" + "sync/atomic" "time" "github.com/mhsanaei/3x-ui/v3/internal/config" @@ -64,6 +66,15 @@ type Agent struct { xrayAPI xray.XrayAPI apiPort int 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 @@ -110,7 +121,7 @@ func (a *Agent) Register() error { req.Header.Set("Content-Type", "application/json") req.Header.Set("Authorization", "Bearer "+a.config.Token) - resp, err := a.httpClient.Do(req) + resp, err := a.do(req) if err != nil { return fmt.Errorf("register failed: %w", err) } @@ -241,12 +252,25 @@ func (a *Agent) Stop() { 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 func (a *Agent) sendHeartbeat() error { url := fmt.Sprintf("%s/panel/api/agent/heartbeat", a.config.MasterURL) payload := map[string]interface{}{ "token": a.config.Token, "timestamp": time.Now().Unix(), + "telemetry": a.collectTelemetry(), } body, _ := json.Marshal(payload) 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("Authorization", "Bearer "+a.config.Token) - resp, err := a.httpClient.Do(req) + resp, err := a.do(req) if err != nil { return err } @@ -276,7 +300,7 @@ func (a *Agent) fetchAndApplyConfig() error { } req.Header.Set("Authorization", "Bearer "+a.config.Token) - resp, err := a.httpClient.Do(req) + resp, err := a.do(req) if err != nil { return err } @@ -509,7 +533,7 @@ func (a *Agent) postTrafficReport(report trafficReport) error { req.Header.Set("Content-Type", "application/json") req.Header.Set("Authorization", "Bearer "+a.config.Token) - resp, err := a.httpClient.Do(req) + resp, err := a.do(req) if err != nil { return err } diff --git a/internal/agent/telemetry.go b/internal/agent/telemetry.go new file mode 100644 index 0000000..087c58c --- /dev/null +++ b/internal/agent/telemetry.go @@ -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) +} diff --git a/internal/agent/telemetry_test.go b/internal/agent/telemetry_test.go new file mode 100644 index 0000000..b905283 --- /dev/null +++ b/internal/agent/telemetry_test.go @@ -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) + } +} diff --git a/internal/database/model/agent_node.go b/internal/database/model/agent_node.go index dbb1024..1fd6482 100644 --- a/internal/database/model/agent_node.go +++ b/internal/database/model/agent_node.go @@ -22,4 +22,20 @@ type AgentNode struct { LastSeen int64 `json:"lastSeen"` CreatedAt int64 `json:"createdAt" gorm:"autoCreateTime: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"` } diff --git a/internal/web/controller/agent.go b/internal/web/controller/agent.go index b58adfc..c2a69e8 100644 --- a/internal/web/controller/agent.go +++ b/internal/web/controller/agent.go @@ -117,13 +117,68 @@ func (a *AgentController) register(c *gin.Context) { 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) { agent, ok := a.findAgent(bearerToken(c)) if !ok { c.JSON(401, gin.H{"message": "unauthorized"}) 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), "") + if req.Telemetry != nil { + a.updateAgentTelemetry(uint(agent.Id), req.Telemetry) + } jsonObj(c, gin.H{"message": "ok"}, nil) } diff --git a/internal/web/controller/agent_telemetry_test.go b/internal/web/controller/agent_telemetry_test.go new file mode 100644 index 0000000..de93c76 --- /dev/null +++ b/internal/web/controller/agent_telemetry_test.go @@ -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) + } +} diff --git a/internal/web/service/node.go b/internal/web/service/node.go index 5a12b4b..98bcb5a 100644 --- a/internal/web/service/node.go +++ b/internal/web/service/node.go @@ -362,7 +362,9 @@ func (s *NodeService) GetViewById(id int) (*NodeView, error) { return nil, err } view := toNodeView(n) - annotateNodeTypes([]*NodeView{view}) + views := []*NodeView{view} + annotateNodeTypes(views) + annotateAgentTelemetry(views) return view, nil } diff --git a/internal/web/service/node_agent_telemetry_test.go b/internal/web/service/node_agent_telemetry_test.go new file mode 100644 index 0000000..92e4533 --- /dev/null +++ b/internal/web/service/node_agent_telemetry_test.go @@ -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 ", 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) + } +} diff --git a/internal/web/service/node_contract.go b/internal/web/service/node_contract.go index 595c022..d3fcae3 100644 --- a/internal/web/service/node_contract.go +++ b/internal/web/service/node_contract.go @@ -31,7 +31,13 @@ type NodeView struct { // agent binding, not stored on the node row; read paths that serve the // panel UI annotate it via annotateNodeTypes. Empty on views built // 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"` Status string `json:"status" example:"online"` 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 " 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 // only as input. On update, nil means keep the stored token; replacement and // clearing are explicit and mutually exclusive. diff --git a/internal/web/service/node_tree.go b/internal/web/service/node_tree.go index 91727bb..03e9cb8 100644 --- a/internal/web/service/node_tree.go +++ b/internal/web/service/node_tree.go @@ -164,6 +164,7 @@ func (s *NodeService) GetNodeTreeView() ([]*NodeView, error) { } views := toNodeViews(nodes) annotateNodeTypes(views) + annotateAgentTelemetry(views) return views, nil }