feat(probe): port monitor-probe collector into agent, persist latest snapshot (item 7 B1)
This commit is contained in:
Muse
@@ -118,9 +118,11 @@ func (e *statusError) Error() string {
|
||||
// Register registers the agent with the master
|
||||
func (a *Agent) Register() error {
|
||||
url := fmt.Sprintf("%s/panel/api/agent/register", a.config.MasterURL)
|
||||
payload := map[string]string{
|
||||
payload := map[string]any{
|
||||
"token": a.config.Token,
|
||||
"name": a.config.NodeName,
|
||||
// Probe facts ride along (B1): old masters ignore the extra field.
|
||||
"probeFacts": collectFacts(),
|
||||
}
|
||||
body, _ := json.Marshal(payload)
|
||||
req, err := http.NewRequest("POST", url, bytes.NewReader(body))
|
||||
@@ -280,6 +282,12 @@ func (a *Agent) sendHeartbeat() error {
|
||||
"token": a.config.Token,
|
||||
"timestamp": time.Now().Unix(),
|
||||
"telemetry": a.collectTelemetry(),
|
||||
// Probe snapshot (B1): facts are process-cached so re-sending them
|
||||
// costs nothing and heals a master that missed the register-time
|
||||
// copy; metrics are sampled fresh each heartbeat. Old masters
|
||||
// ignore both fields.
|
||||
"probeFacts": collectFacts(),
|
||||
"probeMetrics": collectMetrics(),
|
||||
}
|
||||
if rep := a.currentUpdateReport(); rep != nil {
|
||||
payload["updateStatus"] = rep
|
||||
|
||||
@@ -0,0 +1,950 @@
|
||||
package agent
|
||||
|
||||
// probe.go ports monitor-probe's agent collector (collect.rs, MIT) to Go:
|
||||
// host Facts (once per process) and per-heartbeat Metrics read directly from
|
||||
// /proc and /sys with the standard library only. The port keeps the source's
|
||||
// accounting discipline:
|
||||
// - memory "used" is free(1)'s column: MemTotal - MemAvailable;
|
||||
// - disk "used" is df(1)'s: sum over real mounts of (f_blocks-f_bfree);
|
||||
// - network totals sum only "counted" interfaces — loopback, container,
|
||||
// tunnel, bridge/bond/VLAN stacked devices are excluded by a three-layer
|
||||
// filter so no byte is counted twice;
|
||||
// - boot epoch = kernel boot_id + FNV-1a fingerprint of the counted set,
|
||||
// so a set change re-baselines exactly like a reboot;
|
||||
// - every source degrades independently: an unreadable file yields a zero
|
||||
// field, never a failed collection round.
|
||||
|
||||
import (
|
||||
"hash/fnv"
|
||||
"net"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"runtime"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"github.com/mhsanaei/3x-ui/v3/internal/config"
|
||||
)
|
||||
|
||||
// ProbeFacts mirrors monitor's Facts frame: static host identity, sampled
|
||||
// once and piggybacked on the register request and (cheaply) heartbeats.
|
||||
type ProbeFacts struct {
|
||||
Hostname string `json:"hostname"`
|
||||
OS string `json:"os"`
|
||||
Kernel string `json:"kernel"`
|
||||
Arch string `json:"arch"`
|
||||
Virt string `json:"virt"`
|
||||
CPUName string `json:"cpuName"`
|
||||
CPUCores uint32 `json:"cpuCores"`
|
||||
MemTotal uint64 `json:"memTotal"`
|
||||
SwapTotal uint64 `json:"swapTotal"`
|
||||
DiskTotal uint64 `json:"diskTotal"`
|
||||
AgentVersion string `json:"agentVersion"`
|
||||
IPv4 string `json:"ipv4"`
|
||||
IPv6 string `json:"ipv6"`
|
||||
}
|
||||
|
||||
// ProbeMetrics mirrors monitor's Metrics frame: one sample per heartbeat.
|
||||
type ProbeMetrics struct {
|
||||
BootID string `json:"bootId"`
|
||||
Iface string `json:"iface"`
|
||||
Uptime uint64 `json:"uptime"`
|
||||
CPU float32 `json:"cpu"`
|
||||
Load [3]float32 `json:"load"`
|
||||
MemTotal uint64 `json:"memTotal"`
|
||||
MemUsed uint64 `json:"memUsed"`
|
||||
SwapTotal uint64 `json:"swapTotal"`
|
||||
SwapUsed uint64 `json:"swapUsed"`
|
||||
DiskTotal uint64 `json:"diskTotal"`
|
||||
DiskUsed uint64 `json:"diskUsed"`
|
||||
NetRxTotal uint64 `json:"netRxTotal"`
|
||||
NetTxTotal uint64 `json:"netTxTotal"`
|
||||
NetRx uint64 `json:"netRx"`
|
||||
NetTx uint64 `json:"netTx"`
|
||||
TCP uint32 `json:"tcp"`
|
||||
UDP uint32 `json:"udp"`
|
||||
Procs uint32 `json:"procs"`
|
||||
}
|
||||
|
||||
// probeCollector holds the cross-sample state the rates need (previous CPU
|
||||
// jiffies and per-interface counters). Kept in memory only, like the source.
|
||||
type probeCollector struct {
|
||||
mu sync.Mutex
|
||||
prevCPU *cpuSample
|
||||
prevNet map[string][2]uint64
|
||||
prevNetT time.Time
|
||||
}
|
||||
|
||||
type cpuSample struct{ total, idle uint64 }
|
||||
|
||||
var (
|
||||
probeColl = &probeCollector{}
|
||||
factsOnce sync.Once
|
||||
factsCache *ProbeFacts
|
||||
)
|
||||
|
||||
// collectFacts returns the process-cached Facts, sampling on first use.
|
||||
func collectFacts() *ProbeFacts {
|
||||
factsOnce.Do(func() { factsCache = sampleFacts() })
|
||||
return factsCache
|
||||
}
|
||||
|
||||
func sampleFacts() *ProbeFacts {
|
||||
mem := parseMeminfo(readProcFile("meminfo"))
|
||||
diskTotal, _ := diskUsage(realMountPoints(readProcFile("self/mounts")))
|
||||
cpuName, cpuCores := cpuInfo(readProcFile("cpuinfo"))
|
||||
v4, v6 := probeAddresses()
|
||||
return &ProbeFacts{
|
||||
Hostname: readTrimFile("/proc/sys/kernel/hostname", "unknown"),
|
||||
OS: osPrettyName(),
|
||||
Kernel: readTrimFile("/proc/sys/kernel/osrelease", "unknown"),
|
||||
Arch: runtime.GOARCH,
|
||||
Virt: detectVirt(readProcFile("self/mounts"), readProcFile("cpuinfo")),
|
||||
CPUName: cpuName,
|
||||
CPUCores: cpuCores,
|
||||
MemTotal: mem["MemTotal"],
|
||||
SwapTotal: mem["SwapTotal"],
|
||||
DiskTotal: diskTotal,
|
||||
AgentVersion: config.GetAgentDisplayVersion(),
|
||||
IPv4: v4,
|
||||
IPv6: v6,
|
||||
}
|
||||
}
|
||||
|
||||
// collectMetrics takes one full sample. Individual sources fail soft.
|
||||
func collectMetrics() *ProbeMetrics {
|
||||
probeColl.mu.Lock()
|
||||
defer probeColl.mu.Unlock()
|
||||
mem := parseMeminfo(readProcFile("meminfo"))
|
||||
memTotal, memUsed := memUsedFree(mem)
|
||||
swapTotal, swapUsed := swapUsedFree(mem)
|
||||
diskTotal, diskUsed := diskUsage(realMountPoints(readProcFile("self/mounts")))
|
||||
counted := countedInterfaces(parseNetDev(readProcFile("net/dev")), sysNetRoot)
|
||||
var rxTotal, txTotal uint64
|
||||
names := make([]string, 0, len(counted))
|
||||
for _, c := range counted {
|
||||
rxTotal += c.rx
|
||||
txTotal += c.tx
|
||||
names = append(names, c.name)
|
||||
}
|
||||
bootID := bootEpoch(readTrimFile("/proc/sys/kernel/random/boot_id", ""), names)
|
||||
rxRate, txRate := probeColl.netRate(counted, time.Now())
|
||||
tcp, udp := connCounts(readProcFile("net/sockstat"), readProcFile("net/sockstat6"))
|
||||
return &ProbeMetrics{
|
||||
BootID: bootID,
|
||||
Uptime: parseUptime(readProcFile("uptime")),
|
||||
CPU: probeColl.cpuPercent(parseCPUJiffies(readProcFile("stat"))),
|
||||
Load: parseLoadavg(readProcFile("loadavg")),
|
||||
MemTotal: memTotal,
|
||||
MemUsed: memUsed,
|
||||
SwapTotal: swapTotal,
|
||||
SwapUsed: swapUsed,
|
||||
DiskTotal: diskTotal,
|
||||
DiskUsed: diskUsed,
|
||||
NetRxTotal: rxTotal,
|
||||
NetTxTotal: txTotal,
|
||||
NetRx: rxRate,
|
||||
NetTx: txRate,
|
||||
TCP: tcp,
|
||||
UDP: udp,
|
||||
Procs: procCount(),
|
||||
}
|
||||
}
|
||||
|
||||
// --- file helpers -----------------------------------------------------------
|
||||
|
||||
const sysNetRoot = "/sys/class/net"
|
||||
|
||||
func readProcFile(name string) string {
|
||||
b, err := os.ReadFile(filepath.Join("/proc", name))
|
||||
if err != nil {
|
||||
return ""
|
||||
}
|
||||
return string(b)
|
||||
}
|
||||
|
||||
func readTrimFile(path, fallback string) string {
|
||||
b, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return fallback
|
||||
}
|
||||
if s := strings.TrimSpace(string(b)); s != "" {
|
||||
return s
|
||||
}
|
||||
return fallback
|
||||
}
|
||||
|
||||
// --- memory (free口径) -------------------------------------------------------
|
||||
|
||||
func parseMeminfo(text string) map[string]uint64 {
|
||||
out := make(map[string]uint64)
|
||||
for _, line := range strings.Split(text, "\n") {
|
||||
key, rest, ok := strings.Cut(line, ":")
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
f := strings.Fields(rest)
|
||||
if len(f) == 0 {
|
||||
continue
|
||||
}
|
||||
kb, err := strconv.ParseUint(f[0], 10, 64)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
out[key] = kb * 1024
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// memUsedFree is free(1)'s used column: total - MemAvailable, with the
|
||||
// source's two fallbacks (no MemAvailable -> Free+Buffers+Cached; an
|
||||
// available above the total, as LXCFS can report, -> total - MemFree).
|
||||
func memUsedFree(m map[string]uint64) (total, used uint64) {
|
||||
total = m["MemTotal"]
|
||||
if total == 0 {
|
||||
return 0, 0
|
||||
}
|
||||
avail, ok := m["MemAvailable"]
|
||||
if !ok {
|
||||
avail = m["MemFree"] + m["Buffers"] + m["Cached"]
|
||||
}
|
||||
if avail > total {
|
||||
avail = m["MemFree"]
|
||||
}
|
||||
if avail > total {
|
||||
avail = total
|
||||
}
|
||||
return total, total - avail
|
||||
}
|
||||
|
||||
func swapUsedFree(m map[string]uint64) (total, used uint64) {
|
||||
total = m["SwapTotal"]
|
||||
free := m["SwapFree"]
|
||||
if free > total {
|
||||
free = total
|
||||
}
|
||||
return total, total - free
|
||||
}
|
||||
|
||||
// --- cpu --------------------------------------------------------------------
|
||||
|
||||
func parseCPUJiffies(stat string) *cpuSample {
|
||||
for _, line := range strings.Split(stat, "\n") {
|
||||
if !strings.HasPrefix(line, "cpu ") {
|
||||
continue
|
||||
}
|
||||
f := strings.Fields(line[4:])
|
||||
if len(f) < 5 {
|
||||
return nil
|
||||
}
|
||||
var v [8]uint64
|
||||
var total uint64
|
||||
for i := 0; i < len(f) && i < 8; i++ {
|
||||
n, err := strconv.ParseUint(f[i], 10, 64)
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
v[i] = n
|
||||
total += n
|
||||
}
|
||||
return &cpuSample{total: total, idle: v[3] + v[4]}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *probeCollector) cpuPercent(now *cpuSample) float32 {
|
||||
if now == nil {
|
||||
return 0
|
||||
}
|
||||
prev := c.prevCPU
|
||||
c.prevCPU = now
|
||||
if prev == nil || now.total <= prev.total {
|
||||
return 0
|
||||
}
|
||||
dt := float32(now.total - prev.total)
|
||||
di := float32(0)
|
||||
if now.idle > prev.idle {
|
||||
di = float32(now.idle - prev.idle)
|
||||
}
|
||||
pct := (dt - di) / dt * 100
|
||||
if pct < 0 {
|
||||
pct = 0
|
||||
}
|
||||
if pct > 100 {
|
||||
pct = 100
|
||||
}
|
||||
return pct
|
||||
}
|
||||
|
||||
func parseLoadavg(text string) [3]float32 {
|
||||
var out [3]float32
|
||||
f := strings.Fields(text)
|
||||
for i := 0; i < 3 && i < len(f); i++ {
|
||||
if v, err := strconv.ParseFloat(f[i], 32); err == nil {
|
||||
out[i] = float32(v)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func parseUptime(text string) uint64 {
|
||||
f := strings.Fields(text)
|
||||
if len(f) == 0 {
|
||||
return 0
|
||||
}
|
||||
v, err := strconv.ParseFloat(f[0], 64)
|
||||
if err != nil || v < 0 {
|
||||
return 0
|
||||
}
|
||||
return uint64(v)
|
||||
}
|
||||
|
||||
// cpuInfo returns the display name and logical core count from /proc/cpuinfo.
|
||||
func cpuInfo(text string) (string, uint32) {
|
||||
name := ""
|
||||
cores := uint32(0)
|
||||
for _, line := range strings.Split(text, "\n") {
|
||||
key, val, ok := strings.Cut(line, ":")
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
switch strings.TrimSpace(key) {
|
||||
case "processor":
|
||||
cores++
|
||||
case "model name", "Model", "cpu model":
|
||||
if name == "" {
|
||||
name = strings.TrimSpace(val)
|
||||
}
|
||||
}
|
||||
}
|
||||
if cores == 0 {
|
||||
cores = uint32(runtime.NumCPU())
|
||||
if cores == 0 {
|
||||
cores = 1
|
||||
}
|
||||
}
|
||||
if name == "" {
|
||||
name = armCPUName(text)
|
||||
}
|
||||
if name == "" {
|
||||
name = "unknown"
|
||||
}
|
||||
return name, cores
|
||||
}
|
||||
|
||||
// armCPUName decodes AArch64 MIDR implementer/part pairs the way lscpu does,
|
||||
// joining distinct core types with "/" (e.g. Cortex-A55/Cortex-A76).
|
||||
func armCPUName(text string) string {
|
||||
var vendor *armVendor
|
||||
var names []string
|
||||
seen := map[string]bool{}
|
||||
for _, line := range strings.Split(text, "\n") {
|
||||
key, val, ok := strings.Cut(line, ":")
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
switch strings.TrimSpace(key) {
|
||||
case "CPU implementer":
|
||||
vendor = nil
|
||||
if id, ok := parseHexID(val); ok {
|
||||
for i := range armVendors {
|
||||
if armVendors[i].id == id {
|
||||
vendor = &armVendors[i]
|
||||
}
|
||||
}
|
||||
}
|
||||
case "CPU part":
|
||||
if vendor == nil {
|
||||
continue
|
||||
}
|
||||
name := vendor.name
|
||||
if id, ok := parseHexID(val); ok {
|
||||
for _, p := range vendor.parts {
|
||||
if p.id == id {
|
||||
name = p.name
|
||||
}
|
||||
}
|
||||
}
|
||||
if !seen[name] {
|
||||
seen[name] = true
|
||||
names = append(names, name)
|
||||
}
|
||||
}
|
||||
}
|
||||
return strings.Join(names, "/")
|
||||
}
|
||||
|
||||
func parseHexID(s string) (uint32, bool) {
|
||||
s = strings.TrimSpace(strings.TrimPrefix(strings.TrimSpace(s), "0x"))
|
||||
v, err := strconv.ParseUint(s, 16, 32)
|
||||
if err != nil {
|
||||
return 0, false
|
||||
}
|
||||
return uint32(v), true
|
||||
}
|
||||
|
||||
type armPart struct {
|
||||
id uint32
|
||||
name string
|
||||
}
|
||||
type armVendor struct {
|
||||
id uint32
|
||||
name string
|
||||
parts []armPart
|
||||
}
|
||||
|
||||
// Subset of the source's MIDR table (util-linux lscpu-arm names): the parts
|
||||
// plausibly met on VPS hardware. An unlisted part falls back to the vendor.
|
||||
var armVendors = []armVendor{
|
||||
{0x41, "ARM", []armPart{
|
||||
{0xd03, "Cortex-A53"}, {0xd04, "Cortex-A35"}, {0xd05, "Cortex-A55"},
|
||||
{0xd07, "Cortex-A57"}, {0xd08, "Cortex-A72"}, {0xd09, "Cortex-A73"},
|
||||
{0xd0a, "Cortex-A75"}, {0xd0b, "Cortex-A76"}, {0xd0c, "Neoverse-N1"},
|
||||
{0xd0d, "Cortex-A77"}, {0xd40, "Neoverse-V1"}, {0xd41, "Cortex-A78"},
|
||||
{0xd44, "Cortex-X1"}, {0xd46, "Cortex-A510"}, {0xd47, "Cortex-A710"},
|
||||
{0xd48, "Cortex-X2"}, {0xd49, "Neoverse-N2"}, {0xd4d, "Cortex-A715"},
|
||||
{0xd4e, "Cortex-X3"}, {0xd4f, "Neoverse-V2"}, {0xd80, "Cortex-A520"},
|
||||
{0xd81, "Cortex-A720"}, {0xd82, "Cortex-X4"}, {0xd84, "Neoverse-V3"},
|
||||
{0xd85, "Cortex-X925"}, {0xd87, "Cortex-A725"}, {0xd8e, "Neoverse-N3"},
|
||||
}},
|
||||
{0x48, "HiSilicon", []armPart{
|
||||
{0xd01, "Kunpeng-920"}, {0xd02, "Kunpeng-920"}, {0xd03, "Kunpeng-920"},
|
||||
{0xd06, "Kunpeng-950"}, {0xd22, "Kunpeng-920"},
|
||||
}},
|
||||
{0x51, "Qualcomm", []armPart{{0x001, "Oryon"}, {0x002, "Oryon-2"}}},
|
||||
{0x50, "APM", []armPart{{0x000, "X-Gene"}}},
|
||||
{0xc0, "Ampere", []armPart{{0xac3, "Ampere-1"}, {0xac4, "Ampere-1a"}}},
|
||||
{0x6d, "Microsoft", []armPart{{0xd49, "Azure-Cobalt-100"}}},
|
||||
}
|
||||
|
||||
// --- os / virt ---------------------------------------------------------------
|
||||
|
||||
func osPrettyName() string {
|
||||
for _, path := range []string{"/etc/os-release", "/usr/lib/os-release"} {
|
||||
b, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
for _, line := range strings.Split(string(b), "\n") {
|
||||
if v, ok := strings.CutPrefix(line, "PRETTY_NAME="); ok {
|
||||
return strings.Trim(v, "\"'")
|
||||
}
|
||||
}
|
||||
}
|
||||
return "Linux"
|
||||
}
|
||||
|
||||
// detectVirt ports the source's chain: containers first, then hypervisor.
|
||||
// CPUID is unreachable from Go's stdlib, so the x86 signature check is
|
||||
// replaced by the DMI strings (the source's own non-x86 fallback) plus the
|
||||
// /proc/cpuinfo hypervisor flag — a guest of a hypervisor that hides both
|
||||
// reads as "none", same caveat the source documents for cleared CPUID bits.
|
||||
func detectVirt(mounts, cpuinfo string) string {
|
||||
if _, err := os.Stat("/proc/vz"); err == nil {
|
||||
if _, err := os.Stat("/proc/bc"); err != nil {
|
||||
return "openvz"
|
||||
}
|
||||
}
|
||||
if s := readTrimFile("/run/systemd/container", ""); s != "" {
|
||||
return s
|
||||
}
|
||||
if _, err := os.Stat("/.dockerenv"); err == nil {
|
||||
return "docker"
|
||||
}
|
||||
if _, err := os.Stat("/run/.containerenv"); err == nil {
|
||||
return "podman"
|
||||
}
|
||||
for _, line := range strings.Split(mounts, "\n") {
|
||||
f := strings.Fields(line)
|
||||
if len(f) >= 3 && f[2] == "fuse.lxcfs" && strings.HasPrefix(f[1], "/proc/") {
|
||||
return "lxc"
|
||||
}
|
||||
}
|
||||
if _, err := os.Stat("/proc/xen"); err == nil {
|
||||
return "xen"
|
||||
}
|
||||
if _, err := os.Stat("/sys/hypervisor/type"); err == nil {
|
||||
return "xen"
|
||||
}
|
||||
dmi := strings.ToLower(readTrimFile("/sys/class/dmi/id/product_name", "") + " " +
|
||||
readTrimFile("/sys/class/dmi/id/sys_vendor", ""))
|
||||
for _, m := range []struct{ word, name string }{
|
||||
{"kvm", "kvm"}, {"qemu", "kvm"}, {"amazon ec2", "kvm"}, {"google", "kvm"},
|
||||
{"microsoft", "hyper-v"}, {"vmware", "vmware"}, {"virtualbox", "virtualbox"},
|
||||
{"bhyve", "bhyve"}, {"xen", "xen"},
|
||||
} {
|
||||
if strings.Contains(dmi, m.word) {
|
||||
return m.name
|
||||
}
|
||||
}
|
||||
if _, err := os.Stat("/sys/bus/acpi/devices/QEMU0002:00"); err == nil {
|
||||
return "kvm"
|
||||
}
|
||||
if strings.Contains(cpuinfo, "hypervisor") {
|
||||
return "vm"
|
||||
}
|
||||
return "none"
|
||||
}
|
||||
|
||||
// --- network: counted set, epoch, rates -------------------------------------
|
||||
|
||||
type netCounters struct {
|
||||
name string
|
||||
rx, tx uint64
|
||||
}
|
||||
|
||||
func parseNetDev(text string) []netCounters {
|
||||
var out []netCounters
|
||||
lines := strings.Split(text, "\n")
|
||||
for _, line := range lines {
|
||||
name, rest, ok := strings.Cut(line, ":")
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
f := strings.Fields(rest)
|
||||
if len(f) < 16 {
|
||||
continue
|
||||
}
|
||||
rx, err1 := strconv.ParseUint(f[0], 10, 64)
|
||||
tx, err2 := strconv.ParseUint(f[8], 10, 64)
|
||||
if err1 != nil || err2 != nil {
|
||||
continue
|
||||
}
|
||||
out = append(out, netCounters{name: strings.TrimSpace(name), rx: rx, tx: tx})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// skipIfacePrefixes is layer 1 of the filter (source SKIP_IFACES).
|
||||
var skipIfacePrefixes = []string{
|
||||
"lo", "docker", "veth", "br-", "virbr", "tap", "tun", "wg", "tailscale",
|
||||
"cni", "flannel", "podman", "fwbr", "fwpr", "fwln", "ifb", "gretap",
|
||||
"erspan", "kube", "cali", "nerdctl", "lxc", "cilium", "zt",
|
||||
}
|
||||
|
||||
func skipIfaceName(name string) bool {
|
||||
for _, p := range skipIfacePrefixes {
|
||||
if strings.HasPrefix(name, p) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// isStackedName is layer 2: bonds, bridges, VLAN children, PPPoE overlays —
|
||||
// the kernel books their bytes again on the device beneath.
|
||||
func isStackedName(name string) bool {
|
||||
if strings.Contains(name, ".") {
|
||||
return true
|
||||
}
|
||||
for _, p := range []string{"bond", "br", "vlan", "vmbr", "pppoe-"} {
|
||||
if strings.HasPrefix(name, p) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
var tunnelLinkTypes = map[string]bool{
|
||||
"65534": true, "768": true, "769": true, "776": true, "778": true, "823": true,
|
||||
}
|
||||
|
||||
// countedElsewhere is layer 3 (sysfs): the kernel shows this device's bytes
|
||||
// are already counted on another device. Hardware (a `device` link) exempts
|
||||
// a device from the tunnel/port rules — a raw-IP LTE modem shares its link
|
||||
// type with WireGuard but carries real traffic.
|
||||
func countedElsewhere(sysRoot, name string) bool {
|
||||
dev := filepath.Join(sysRoot, name)
|
||||
uevent := readTrimFile(filepath.Join(dev, "uevent"), "")
|
||||
devtypeIs := func(set ...string) bool {
|
||||
for _, line := range strings.Split(uevent, "\n") {
|
||||
for _, want := range set {
|
||||
if line == want {
|
||||
return true
|
||||
}
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
if devtypeIs("DEVTYPE=bridge", "DEVTYPE=bond") {
|
||||
return true
|
||||
}
|
||||
if entries, err := os.ReadDir(dev); err == nil {
|
||||
for _, e := range entries {
|
||||
if strings.HasPrefix(e.Name(), "lower_") {
|
||||
return true
|
||||
}
|
||||
}
|
||||
}
|
||||
if _, err := os.Stat(filepath.Join(dev, "device")); err == nil {
|
||||
return false
|
||||
}
|
||||
if _, err := os.Stat(filepath.Join(dev, "brport")); err == nil {
|
||||
return true
|
||||
}
|
||||
if tunnelLinkTypes[readTrimFile(filepath.Join(dev, "type"), "")] {
|
||||
return true
|
||||
}
|
||||
return devtypeIs("DEVTYPE=vxlan", "DEVTYPE=geneve")
|
||||
}
|
||||
|
||||
// ifaceCounts applies the three layers. Exposed for tests with a fake sysfs.
|
||||
func ifaceCounts(sysRoot, name string) bool {
|
||||
if skipIfaceName(name) || isStackedName(name) {
|
||||
return false
|
||||
}
|
||||
return !countedElsewhere(sysRoot, name)
|
||||
}
|
||||
|
||||
func countedInterfaces(all []netCounters, sysRoot string) []netCounters {
|
||||
out := make([]netCounters, 0, len(all))
|
||||
for _, c := range all {
|
||||
if ifaceCounts(sysRoot, c.name) {
|
||||
out = append(out, c)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// bootEpoch = "{kernel boot_id}/{FNV-1a-64 of the sorted counted names}".
|
||||
// The master only tests it for equality: any change re-baselines.
|
||||
func bootEpoch(kernelBootID string, names []string) string {
|
||||
sorted := append([]string(nil), names...)
|
||||
sort.Strings(sorted)
|
||||
h := fnv.New64a()
|
||||
_, _ = h.Write([]byte(strings.Join(sorted, "\n")))
|
||||
return kernelBootID + "/" + hex16(h.Sum64())
|
||||
}
|
||||
|
||||
func hex16(v uint64) string {
|
||||
const digits = "0123456789abcdef"
|
||||
var buf [16]byte
|
||||
for i := 15; i >= 0; i-- {
|
||||
buf[i] = digits[v&0xf]
|
||||
v >>= 4
|
||||
}
|
||||
return string(buf[:])
|
||||
}
|
||||
|
||||
// netRate diffs per interface over those present in both samples
|
||||
// (saturating), divided by the elapsed wall time.
|
||||
func (c *probeCollector) netRate(counted []netCounters, now time.Time) (uint64, uint64) {
|
||||
prev, prevT := c.prevNet, c.prevNetT
|
||||
next := make(map[string][2]uint64, len(counted))
|
||||
for _, ct := range counted {
|
||||
next[ct.name] = [2]uint64{ct.rx, ct.tx}
|
||||
}
|
||||
c.prevNet, c.prevNetT = next, now
|
||||
if prev == nil {
|
||||
return 0, 0
|
||||
}
|
||||
secs := now.Sub(prevT).Seconds()
|
||||
if secs <= 0 {
|
||||
return 0, 0
|
||||
}
|
||||
var rx, tx uint64
|
||||
for _, ct := range counted {
|
||||
p, ok := prev[ct.name]
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
if ct.rx > p[0] {
|
||||
rx += ct.rx - p[0]
|
||||
}
|
||||
if ct.tx > p[1] {
|
||||
tx += ct.tx - p[1]
|
||||
}
|
||||
}
|
||||
return uint64(float64(rx) / secs), uint64(float64(tx) / secs)
|
||||
}
|
||||
|
||||
// --- addresses (scope §1.5) ---------------------------------------------------
|
||||
|
||||
// probeAddresses picks one address per family: candidates come from
|
||||
// non-skip interfaces that are up, are not link-local, prefer public over
|
||||
// private, and for v6 avoid temporary/deprecated/tentative addresses.
|
||||
func probeAddresses() (string, string) {
|
||||
transient := transientV6(readProcFile("net/if_inet6"))
|
||||
ifaces, err := net.Interfaces()
|
||||
if err != nil {
|
||||
return "", ""
|
||||
}
|
||||
type cand struct {
|
||||
ip net.IP
|
||||
transient bool
|
||||
}
|
||||
var held []cand
|
||||
for _, ifc := range ifaces {
|
||||
if ifc.Flags&net.FlagUp == 0 {
|
||||
continue
|
||||
}
|
||||
if readTrimFile(filepath.Join(sysNetRoot, ifc.Name, "operstate"), "up") != "up" {
|
||||
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 || ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast() {
|
||||
continue
|
||||
}
|
||||
if !lendsAddress(ifc.Name, ip) {
|
||||
continue
|
||||
}
|
||||
c := cand{ip: ip}
|
||||
if ip6 := ip.To16(); ip.To4() == nil && ip6 != nil {
|
||||
for _, t := range transient {
|
||||
if t.Equal(ip6) {
|
||||
c.transient = true
|
||||
}
|
||||
}
|
||||
}
|
||||
held = append(held, c)
|
||||
}
|
||||
}
|
||||
pick := func(v6 bool) string {
|
||||
var best *cand
|
||||
for i := range held {
|
||||
c := &held[i]
|
||||
if (c.ip.To4() == nil) != v6 {
|
||||
continue
|
||||
}
|
||||
if best == nil {
|
||||
best = c
|
||||
continue
|
||||
}
|
||||
bp, cp := isPublicIP(best.ip), isPublicIP(c.ip)
|
||||
if cp != bp {
|
||||
if cp {
|
||||
best = c
|
||||
}
|
||||
continue
|
||||
}
|
||||
if v6 && best.transient && !c.transient {
|
||||
best = c
|
||||
}
|
||||
}
|
||||
if best == nil {
|
||||
return ""
|
||||
}
|
||||
return best.ip.String()
|
||||
}
|
||||
return pick(false), pick(true)
|
||||
}
|
||||
|
||||
// lendsAddress mirrors the source: a skipped interface lends no address,
|
||||
// except a WireGuard interface lending a public v6 that is not in
|
||||
// Cloudflare WARP's shared 2606:4700::/32.
|
||||
func lendsAddress(name string, ip net.IP) bool {
|
||||
if !skipIfaceName(name) {
|
||||
return true
|
||||
}
|
||||
if strings.HasPrefix(name, "wg") && ip.To4() == nil && isPublicIP(ip) {
|
||||
ip16 := ip.To16()
|
||||
return !(ip16[0] == 0x26 && ip16[1] == 0x06 && ip16[2] == 0x47 && ip16[3] == 0x00)
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// transientV6 parses /proc/net/if_inet6 for addresses carrying
|
||||
// TEMPORARY|DADFAILED|DEPRECATED|TENTATIVE flags.
|
||||
func transientV6(text string) []net.IP {
|
||||
const transientFlags = 0x01 | 0x08 | 0x20 | 0x40
|
||||
var out []net.IP
|
||||
for _, line := range strings.Split(text, "\n") {
|
||||
f := strings.Fields(line)
|
||||
if len(f) < 6 {
|
||||
continue
|
||||
}
|
||||
flags, err := strconv.ParseUint(f[4], 16, 32)
|
||||
if err != nil || flags&transientFlags == 0 {
|
||||
continue
|
||||
}
|
||||
if len(f[0]) != 32 {
|
||||
continue
|
||||
}
|
||||
ip := make(net.IP, 16)
|
||||
ok := true
|
||||
for i := 0; i < 16; i++ {
|
||||
b, err := strconv.ParseUint(f[0][i*2:i*2+2], 16, 8)
|
||||
if err != nil {
|
||||
ok = false
|
||||
break
|
||||
}
|
||||
ip[i] = byte(b)
|
||||
}
|
||||
if ok {
|
||||
out = append(out, ip)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// isPublicIP mirrors monitor's is_public: v4 excludes RFC1918, CGNAT
|
||||
// 100.64/10, loopback, link-local, 0/8, >=224, 192.0.0/24, 198.18/15;
|
||||
// v6 is only 2000::/3.
|
||||
func isPublicIP(ip net.IP) bool {
|
||||
if ip4 := ip.To4(); ip4 != nil {
|
||||
a, b, c := ip4[0], ip4[1], ip4[2]
|
||||
switch {
|
||||
case ip4.IsPrivate(), ip4.IsLoopback(), ip4.IsLinkLocalUnicast():
|
||||
return false
|
||||
case a == 0 || a >= 224:
|
||||
return false
|
||||
case a == 100 && b&0xc0 == 64:
|
||||
return false
|
||||
case a == 192 && b == 0 && c == 0:
|
||||
return false
|
||||
case a == 198 && b&0xfe == 18:
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
ip16 := ip.To16()
|
||||
if ip16 == nil {
|
||||
return false
|
||||
}
|
||||
return ip16[0]&0xe0 == 0x20
|
||||
}
|
||||
|
||||
// --- sockets / processes ------------------------------------------------------
|
||||
|
||||
func connCounts(v4, v6 string) (tcp, udp uint32) {
|
||||
stat := func(text, prefix, key string) uint32 {
|
||||
for _, line := range strings.Split(text, "\n") {
|
||||
if !strings.HasPrefix(line, prefix) {
|
||||
continue
|
||||
}
|
||||
f := strings.Fields(line[len(prefix):])
|
||||
for i := 0; i+1 < len(f); i++ {
|
||||
if f[i] == key {
|
||||
if n, err := strconv.ParseUint(f[i+1], 10, 32); err == nil {
|
||||
return uint32(n)
|
||||
}
|
||||
return 0
|
||||
}
|
||||
}
|
||||
}
|
||||
return 0
|
||||
}
|
||||
tcp = stat(v4, "TCP:", "inuse") + stat(v4, "TCP:", "tw") + stat(v6, "TCP6:", "inuse")
|
||||
udp = stat(v4, "UDP:", "inuse") + stat(v6, "UDP6:", "inuse")
|
||||
return tcp, udp
|
||||
}
|
||||
|
||||
func procCount() uint32 {
|
||||
entries, err := os.ReadDir("/proc")
|
||||
if err != nil {
|
||||
return 0
|
||||
}
|
||||
var n uint32
|
||||
for _, e := range entries {
|
||||
name := e.Name()
|
||||
digits := name != ""
|
||||
for i := 0; i < len(name); i++ {
|
||||
if name[i] < '0' || name[i] > '9' {
|
||||
digits = false
|
||||
break
|
||||
}
|
||||
}
|
||||
if digits {
|
||||
n++
|
||||
}
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
// --- disk (df口径) ---------------------------------------------------------------
|
||||
|
||||
var skipFSTypes = map[string]bool{
|
||||
"tmpfs": true, "devtmpfs": true, "proc": true, "sysfs": true,
|
||||
"cgroup": true, "cgroup2": true, "devpts": true, "mqueue": true,
|
||||
"hugetlbfs": true, "debugfs": true, "tracefs": true, "securityfs": true,
|
||||
"pstore": true, "bpf": true, "configfs": true, "fusectl": true,
|
||||
"binfmt_misc": true, "autofs": true, "squashfs": true, "ramfs": true,
|
||||
"efivarfs": true, "nsfs": true, "overlay": true, "ecryptfs": true,
|
||||
"fuse": true, "rpc_pipefs": true,
|
||||
"nfs": true, "nfs4": true, "cifs": true, "smb3": true, "ceph": true,
|
||||
"glusterfs": true, "9p": true,
|
||||
}
|
||||
|
||||
func skipFSType(fstype string) bool {
|
||||
if skipFSTypes[fstype] {
|
||||
return true
|
||||
}
|
||||
// Flavours: "fuse.lxcfs" is skipped by its "fuse." prefix, like the source.
|
||||
if i := strings.IndexByte(fstype, '.'); i > 0 && skipFSTypes[fstype[:i]] {
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// realMountPoints ports parse_mounts: only the topmost mount per path,
|
||||
// deduped by source device (zfs by pool), real storage only.
|
||||
func realMountPoints(text string) []string {
|
||||
type row struct{ dev, mount, fstype string }
|
||||
var rows []row
|
||||
top := map[string]int{}
|
||||
for _, line := range strings.Split(text, "\n") {
|
||||
f := strings.Fields(line)
|
||||
if len(f) < 3 {
|
||||
continue
|
||||
}
|
||||
rows = append(rows, row{f[0], f[1], f[2]})
|
||||
top[f[1]] = len(rows) - 1
|
||||
}
|
||||
seen := map[string]bool{}
|
||||
var out []string
|
||||
for i, r := range rows {
|
||||
if top[r.mount] != i || skipFSType(r.fstype) {
|
||||
continue
|
||||
}
|
||||
if !strings.HasPrefix(r.dev, "/") && r.fstype != "zfs" && r.fstype != "btrfs" {
|
||||
continue
|
||||
}
|
||||
key := r.dev
|
||||
if r.fstype == "zfs" {
|
||||
if j := strings.IndexByte(key, '/'); j >= 0 {
|
||||
key = key[:j]
|
||||
}
|
||||
}
|
||||
if seen[key] {
|
||||
continue
|
||||
}
|
||||
seen[key] = true
|
||||
out = append(out, strings.ReplaceAll(r.mount, `\040`, " "))
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// diskUsage sums statfs over the mounts: total = f_blocks, used =
|
||||
// f_blocks - f_bfree (NOT f_bavail — df does not charge the ext reserve).
|
||||
func diskUsage(mounts []string) (total, used uint64) {
|
||||
for _, m := range mounts {
|
||||
var st syscall.Statfs_t
|
||||
if err := syscall.Statfs(m, &st); err != nil {
|
||||
continue
|
||||
}
|
||||
bs := uint64(st.Frsize)
|
||||
if bs == 0 {
|
||||
bs = uint64(st.Bsize)
|
||||
}
|
||||
total += st.Blocks * bs
|
||||
if st.Blocks > st.Bfree {
|
||||
used += (st.Blocks - st.Bfree) * bs
|
||||
}
|
||||
}
|
||||
return total, used
|
||||
}
|
||||
@@ -0,0 +1,235 @@
|
||||
package agent
|
||||
|
||||
import (
|
||||
"net"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Sample /proc/net/dev with a realistic mix: physical NIC, loopback, docker
|
||||
// bridge + veth pair, WireGuard tunnel, a VLAN child and a bridge holding
|
||||
// the host address (vmbr-style stacked name).
|
||||
const sampleNetDev = `Inter-| Receive | Transmit
|
||||
face |bytes packets errs drop fifo frame compressed multicast|bytes packets errs drop fifo colls carrier compressed
|
||||
lo: 1000 10 0 0 0 0 0 0 1000 10 0 0 0 0 0 0
|
||||
eth0: 5000000 5000 0 0 0 0 0 0 3000000 4000 0 0 0 0 0 0
|
||||
docker0: 999000 900 0 0 0 0 0 0 999000 900 0 0 0 0 0 0
|
||||
veth1234: 888000 800 0 0 0 0 0 0 888000 800 0 0 0 0 0 0
|
||||
wg0: 777000 700 0 0 0 0 0 0 777000 700 0 0 0 0 0 0
|
||||
eth0.100: 666000 600 0 0 0 0 0 0 666000 600 0 0 0 0 0 0
|
||||
vmbr0: 555000 500 0 0 0 0 0 0 555000 500 0 0 0 0 0 0
|
||||
`
|
||||
|
||||
func TestCountedInterfacesNameLayers(t *testing.T) {
|
||||
// Empty sysfs root: layers 1+2 (names) must already exclude everything
|
||||
// but eth0.
|
||||
counted := countedInterfaces(parseNetDev(sampleNetDev), t.TempDir())
|
||||
if len(counted) != 1 || counted[0].name != "eth0" {
|
||||
t.Fatalf("counted = %+v, want only eth0", counted)
|
||||
}
|
||||
if counted[0].rx != 5000000 || counted[0].tx != 3000000 {
|
||||
t.Fatalf("eth0 counters = %d/%d", counted[0].rx, counted[0].tx)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCountedElsewhereSysfsLayer(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
mk := func(name string, files map[string]string, dirs ...string) {
|
||||
dev := filepath.Join(root, name)
|
||||
if err := os.MkdirAll(dev, 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for f, content := range files {
|
||||
if err := os.WriteFile(filepath.Join(dev, f), []byte(content), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
for _, d := range dirs {
|
||||
if err := os.MkdirAll(filepath.Join(dev, d), 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
}
|
||||
// Plain NIC: no uevent devtype, hardware device link present.
|
||||
mk("eth0", map[string]string{"type": "1", "uevent": ""}, "device")
|
||||
// A user-named bridge: caught by DEVTYPE alone.
|
||||
mk("lan0", map[string]string{"type": "1", "uevent": "DEVTYPE=bridge\n"})
|
||||
// A VM tap under a made-up name: caught as a bridge port.
|
||||
mk("vnet0", map[string]string{"type": "1", "uevent": ""}, "brport")
|
||||
// A user-named tunnel (type 768 = ipip, no hardware).
|
||||
mk("he-ipv6", map[string]string{"type": "768", "uevent": ""})
|
||||
// Raw-IP LTE modem: tunnel link type (65534) but real hardware -> counts.
|
||||
mk("wwan0", map[string]string{"type": "65534", "uevent": ""}, "device")
|
||||
// macvlan child: has a lower_* link to its parent.
|
||||
mk("mac0", map[string]string{"type": "1", "uevent": ""}, "lower_eth0")
|
||||
|
||||
want := map[string]bool{
|
||||
"eth0": true, "lan0": false, "vnet0": false,
|
||||
"he-ipv6": false, "wwan0": true, "mac0": false,
|
||||
}
|
||||
for name, w := range want {
|
||||
if got := ifaceCounts(root, name); got != w {
|
||||
t.Errorf("ifaceCounts(%s) = %v, want %v", name, got, w)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestBootEpochDiscipline(t *testing.T) {
|
||||
base := bootEpoch("boot-1", []string{"eth0", "eth1"})
|
||||
// Order must not matter (the set is sorted before hashing).
|
||||
if got := bootEpoch("boot-1", []string{"eth1", "eth0"}); got != base {
|
||||
t.Fatalf("epoch not order-stable: %s vs %s", got, base)
|
||||
}
|
||||
if len(base) != len("boot-1/")+16 {
|
||||
t.Fatalf("epoch shape wrong: %q", base)
|
||||
}
|
||||
// A set change is a new epoch (re-baseline), like the source.
|
||||
if got := bootEpoch("boot-1", []string{"eth0"}); got == base {
|
||||
t.Fatal("epoch unchanged after counted set change")
|
||||
}
|
||||
// A kernel reboot is a new epoch even with the same set.
|
||||
if got := bootEpoch("boot-2", []string{"eth0", "eth1"}); got == base {
|
||||
t.Fatal("epoch unchanged after reboot")
|
||||
}
|
||||
// Known vector: FNV-1a-64 of "eth0" (offset basis, prime per hash/fnv).
|
||||
if got := bootEpoch("b", []string{"eth0"}); got != "b/"+fnvEth0 {
|
||||
t.Fatalf("epoch digest = %s, want b/%s", got, fnvEth0)
|
||||
}
|
||||
}
|
||||
|
||||
const fnvEth0 = "c41bf7607a25a404" // hash/fnv FNV-1a-64 over "eth0"
|
||||
|
||||
func TestNetRateBaselineAndSaturation(t *testing.T) {
|
||||
c := &probeCollector{}
|
||||
t0 := time.Unix(1000, 0)
|
||||
// First sample only establishes the baseline.
|
||||
if rx, tx := c.netRate([]netCounters{{"eth0", 1000, 2000}}, t0); rx != 0 || tx != 0 {
|
||||
t.Fatalf("first sample rate = %d/%d, want 0/0", rx, tx)
|
||||
}
|
||||
// Second sample 10s later: +10000/+5000 bytes -> 1000/500 B/s.
|
||||
if rx, tx := c.netRate([]netCounters{{"eth0", 11000, 7000}}, t0.Add(10*time.Second)); rx != 1000 || tx != 500 {
|
||||
t.Fatalf("rate = %d/%d, want 1000/500", rx, tx)
|
||||
}
|
||||
// A counter going backwards (restart) saturates to 0, never wraps.
|
||||
if rx, tx := c.netRate([]netCounters{{"eth0", 5, 5}}, t0.Add(20*time.Second)); rx != 0 || tx != 0 {
|
||||
t.Fatalf("regressed rate = %d/%d, want 0/0", rx, tx)
|
||||
}
|
||||
// An interface joining later contributes nothing in its first sample.
|
||||
rx, _ := c.netRate([]netCounters{{"eth0", 105, 5}, {"eth1", 999999, 0}}, t0.Add(30*time.Second))
|
||||
if rx != 10 {
|
||||
t.Fatalf("rate with joining iface = %d, want 10", rx)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMemUsedFreeSemantics(t *testing.T) {
|
||||
// free口径: used = total - available (cache not charged as used).
|
||||
total, used := memUsedFree(map[string]uint64{
|
||||
"MemTotal": 1000, "MemAvailable": 400, "MemFree": 100, "Buffers": 50, "Cached": 200,
|
||||
})
|
||||
if total != 1000 || used != 600 {
|
||||
t.Fatalf("got %d/%d, want 1000/600", total, used)
|
||||
}
|
||||
// No MemAvailable (old kernels): Free+Buffers+Cached fallback.
|
||||
_, used = memUsedFree(map[string]uint64{
|
||||
"MemTotal": 1000, "MemFree": 100, "Buffers": 50, "Cached": 200,
|
||||
})
|
||||
if used != 650 {
|
||||
t.Fatalf("fallback used = %d, want 650", used)
|
||||
}
|
||||
// LXCFS can report available > total: fall back to total - MemFree.
|
||||
_, used = memUsedFree(map[string]uint64{
|
||||
"MemTotal": 1000, "MemAvailable": 5000, "MemFree": 250,
|
||||
})
|
||||
if used != 750 {
|
||||
t.Fatalf("lxcfs used = %d, want 750", used)
|
||||
}
|
||||
// Swap: total - free, SwapCached NOT subtracted.
|
||||
st, su := swapUsedFree(map[string]uint64{"SwapTotal": 500, "SwapFree": 125})
|
||||
if st != 500 || su != 375 {
|
||||
t.Fatalf("swap = %d/%d, want 500/375", st, su)
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseMountsDfSemantics(t *testing.T) {
|
||||
mounts := `
|
||||
sysfs /sys sysfs rw 0 0
|
||||
/dev/sda1 / ext4 rw 0 0
|
||||
tmpfs /run tmpfs rw 0 0
|
||||
/dev/sda1 /mnt/bind ext4 rw 0 0
|
||||
/dev/sdb1 /data ext4 rw 0 0
|
||||
/dev/sdb1 /data/bind ext4 rw 0 0
|
||||
server:/export /mnt/nfs nfs rw 0 0
|
||||
overlay /var/lib/docker overlay rw 0 0
|
||||
tank/home /home zfs rw 0 0
|
||||
tank/data /srv zfs rw 0 0
|
||||
/dev/sdc1 /shadow ext4 rw 0 0
|
||||
tmpfs /shadow tmpfs rw 0 0
|
||||
`
|
||||
got := realMountPoints(mounts)
|
||||
// / (sda1) and /data (sdb1) dedupe their binds; zfs dedupes by pool;
|
||||
// /shadow's real fs is shadowed by a later tmpfs so it drops out.
|
||||
want := []string{"/", "/data", "/home"}
|
||||
if len(got) != len(want) {
|
||||
t.Fatalf("mounts = %v, want %v", got, want)
|
||||
}
|
||||
for i := range want {
|
||||
if got[i] != want[i] {
|
||||
t.Fatalf("mounts = %v, want %v", got, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestConnCountsSockstat(t *testing.T) {
|
||||
v4 := "sockets: used 100\nTCP: inuse 10 orphan 0 tw 5 alloc 20 mem 1\nUDP: inuse 7 mem 0\n"
|
||||
v6 := "TCP6: inuse 3\nUDP6: inuse 2\n"
|
||||
tcp, udp := connCounts(v4, v6)
|
||||
// tw counted once (v4 line serves both families): 10+5+3; udp 7+2.
|
||||
if tcp != 18 || udp != 9 {
|
||||
t.Fatalf("tcp/udp = %d/%d, want 18/9", tcp, udp)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCPUPercentDiff(t *testing.T) {
|
||||
c := &probeCollector{}
|
||||
if got := c.cpuPercent(&cpuSample{total: 1000, idle: 800}); got != 0 {
|
||||
t.Fatalf("first sample = %v, want 0", got)
|
||||
}
|
||||
// +1000 total, +600 idle -> 40% busy.
|
||||
got := c.cpuPercent(&cpuSample{total: 2000, idle: 1400})
|
||||
if got < 39.9 || got > 40.1 {
|
||||
t.Fatalf("cpu = %v, want ~40", got)
|
||||
}
|
||||
}
|
||||
|
||||
func parseIP(t *testing.T, s string) net.IP {
|
||||
t.Helper()
|
||||
ip := net.ParseIP(s)
|
||||
if ip == nil {
|
||||
t.Fatalf("bad test ip %q", s)
|
||||
}
|
||||
return ip
|
||||
}
|
||||
|
||||
func TestIsPublicIP(t *testing.T) {
|
||||
cases := map[string]bool{
|
||||
"8.8.8.8": true,
|
||||
"192.168.1.1": false,
|
||||
"100.64.0.1": false, // CGNAT
|
||||
"198.18.0.1": false, // TUN fake-IP range
|
||||
"192.0.0.9": false,
|
||||
"224.0.0.1": false,
|
||||
"2001:db8::1": true,
|
||||
"fd00::1": false, // ULA
|
||||
"fe80::1": false,
|
||||
"2606:4700::1": true, // public by range; WARP exclusion lives in lendsAddress
|
||||
"::1": false,
|
||||
}
|
||||
for s, want := range cases {
|
||||
ip := parseIP(t, s)
|
||||
if got := isPublicIP(ip); got != want {
|
||||
t.Errorf("isPublicIP(%s) = %v, want %v", s, got, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -48,6 +48,40 @@ type AgentNode struct {
|
||||
// a telemetry object.
|
||||
TelemetryAt int64 `json:"telemetryAt"`
|
||||
|
||||
// Probe snapshot (item 7, B1): the monitor-probe-style host Facts and
|
||||
// latest Metrics the agent piggybacks on register + heartbeat. Only the
|
||||
// latest values live here; history lands in B2's probe tables. Agents
|
||||
// that never reported a probe block keep ProbeAt == 0 and zero values.
|
||||
ProbeHostname string `json:"probeHostname"`
|
||||
ProbeOS string `json:"probeOs"`
|
||||
ProbeKernel string `json:"probeKernel"`
|
||||
ProbeArch string `json:"probeArch"`
|
||||
ProbeVirt string `json:"probeVirt"`
|
||||
ProbeCPUName string `json:"probeCpuName"`
|
||||
ProbeCPUCores uint32 `json:"probeCpuCores"`
|
||||
ProbeBootID string `json:"probeBootId"`
|
||||
ProbeIface string `json:"probeIface"`
|
||||
ProbeLoad1 float32 `json:"probeLoad1"`
|
||||
ProbeLoad5 float32 `json:"probeLoad5"`
|
||||
ProbeLoad15 float32 `json:"probeLoad15"`
|
||||
ProbeMemTotal uint64 `json:"probeMemTotal"`
|
||||
ProbeMemUsed uint64 `json:"probeMemUsed"`
|
||||
ProbeSwapTotal uint64 `json:"probeSwapTotal"`
|
||||
ProbeSwapUsed uint64 `json:"probeSwapUsed"`
|
||||
ProbeDiskTotal uint64 `json:"probeDiskTotal"`
|
||||
ProbeDiskUsed uint64 `json:"probeDiskUsed"`
|
||||
ProbeNetRxTot uint64 `json:"probeNetRxTotal"`
|
||||
ProbeNetTxTot uint64 `json:"probeNetTxTotal"`
|
||||
ProbeNetRx uint64 `json:"probeNetRx"`
|
||||
ProbeNetTx uint64 `json:"probeNetTx"`
|
||||
ProbeTCP uint32 `json:"probeTcp"`
|
||||
ProbeUDP uint32 `json:"probeUdp"`
|
||||
ProbeProcs uint32 `json:"probeProcs"`
|
||||
ProbeCPU float32 `json:"probeCpu"`
|
||||
// ProbeAt is the unix-milli time of the last heartbeat that carried a
|
||||
// probe metrics block.
|
||||
ProbeAt int64 `json:"probeAt"`
|
||||
|
||||
// Seen source addresses (public-address resolution): the source IP of
|
||||
// the agent's own authenticated requests, as observed by the master —
|
||||
// the CF-Connecting-IP header when the agent comes through Cloudflare,
|
||||
|
||||
@@ -186,8 +186,9 @@ func (a *AgentController) touchAgent(id uint, name string) {
|
||||
// POST /panel/api/agent/tokens; the agent only claims its token here.
|
||||
func (a *AgentController) register(c *gin.Context) {
|
||||
var req struct {
|
||||
Token string `json:"token"`
|
||||
Name string `json:"name"`
|
||||
Token string `json:"token"`
|
||||
Name string `json:"name"`
|
||||
ProbeFacts *heartbeatProbeFacts `json:"probeFacts"`
|
||||
}
|
||||
if err := c.ShouldBindJSON(&req); err != nil || req.Token == "" {
|
||||
jsonMsg(c, "Token required", errRequest)
|
||||
@@ -200,6 +201,9 @@ func (a *AgentController) register(c *gin.Context) {
|
||||
}
|
||||
a.captureSeenIP(c, agent)
|
||||
a.touchAgent(uint(agent.Id), req.Name)
|
||||
if req.ProbeFacts != nil {
|
||||
a.updateAgentProbe(uint(agent.Id), req.ProbeFacts, nil)
|
||||
}
|
||||
logger.Infof("Agent registered: %s (token %s...)", req.Name, agent.TokenPrefix)
|
||||
jsonObj(c, gin.H{"message": "registered"}, nil)
|
||||
}
|
||||
@@ -250,6 +254,145 @@ func (a *AgentController) updateAgentTelemetry(id uint, t *heartbeatTelemetry) {
|
||||
_ = database.GetDB().Model(&model.AgentNode{}).Where("id = ?", id).Updates(updates).Error
|
||||
}
|
||||
|
||||
// heartbeatProbeFacts / heartbeatProbeMetrics are the monitor-probe-style
|
||||
// host snapshot an agent piggybacks on register and heartbeat (item 7 B1).
|
||||
// Every field is a pointer for the same reason as telemetry: an absent
|
||||
// field means "not reported" and must not overwrite the stored value, so
|
||||
// old agents (no probe block at all) and partial reports both merge
|
||||
// cleanly. JSON names must match internal/agent's ProbeFacts/ProbeMetrics.
|
||||
type heartbeatProbeFacts struct {
|
||||
Hostname *string `json:"hostname"`
|
||||
OS *string `json:"os"`
|
||||
Kernel *string `json:"kernel"`
|
||||
Arch *string `json:"arch"`
|
||||
Virt *string `json:"virt"`
|
||||
CPUName *string `json:"cpuName"`
|
||||
CPUCores *uint32 `json:"cpuCores"`
|
||||
MemTotal *uint64 `json:"memTotal"`
|
||||
SwapTotal *uint64 `json:"swapTotal"`
|
||||
DiskTotal *uint64 `json:"diskTotal"`
|
||||
}
|
||||
|
||||
type heartbeatProbeMetrics struct {
|
||||
BootID *string `json:"bootId"`
|
||||
Iface *string `json:"iface"`
|
||||
Load *[3]float32 `json:"load"`
|
||||
MemTotal *uint64 `json:"memTotal"`
|
||||
MemUsed *uint64 `json:"memUsed"`
|
||||
SwapTotal *uint64 `json:"swapTotal"`
|
||||
SwapUsed *uint64 `json:"swapUsed"`
|
||||
DiskTotal *uint64 `json:"diskTotal"`
|
||||
DiskUsed *uint64 `json:"diskUsed"`
|
||||
NetRxTotal *uint64 `json:"netRxTotal"`
|
||||
NetTxTotal *uint64 `json:"netTxTotal"`
|
||||
NetRx *uint64 `json:"netRx"`
|
||||
NetTx *uint64 `json:"netTx"`
|
||||
TCP *uint32 `json:"tcp"`
|
||||
UDP *uint32 `json:"udp"`
|
||||
Procs *uint32 `json:"procs"`
|
||||
CPU *float32 `json:"cpu"`
|
||||
}
|
||||
|
||||
// updateAgentProbe persists the reported probe facts/metrics onto the agent
|
||||
// row. Only fields actually present in the report are written. Facts and
|
||||
// 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{}
|
||||
if f != nil {
|
||||
if f.Hostname != nil {
|
||||
updates["probe_hostname"] = *f.Hostname
|
||||
}
|
||||
if f.OS != nil {
|
||||
updates["probe_os"] = *f.OS
|
||||
}
|
||||
if f.Kernel != nil {
|
||||
updates["probe_kernel"] = *f.Kernel
|
||||
}
|
||||
if f.Arch != nil {
|
||||
updates["probe_arch"] = *f.Arch
|
||||
}
|
||||
if f.Virt != nil {
|
||||
updates["probe_virt"] = *f.Virt
|
||||
}
|
||||
if f.CPUName != nil {
|
||||
updates["probe_cpu_name"] = *f.CPUName
|
||||
}
|
||||
if f.CPUCores != nil {
|
||||
updates["probe_cpu_cores"] = *f.CPUCores
|
||||
}
|
||||
if f.MemTotal != nil {
|
||||
updates["probe_mem_total"] = *f.MemTotal
|
||||
}
|
||||
if f.SwapTotal != nil {
|
||||
updates["probe_swap_total"] = *f.SwapTotal
|
||||
}
|
||||
if f.DiskTotal != nil {
|
||||
updates["probe_disk_total"] = *f.DiskTotal
|
||||
}
|
||||
}
|
||||
if m != nil {
|
||||
updates["probe_at"] = time.Now().UnixMilli()
|
||||
if m.BootID != nil {
|
||||
updates["probe_boot_id"] = *m.BootID
|
||||
}
|
||||
if m.Iface != nil {
|
||||
updates["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]
|
||||
}
|
||||
if m.MemTotal != nil {
|
||||
updates["probe_mem_total"] = *m.MemTotal
|
||||
}
|
||||
if m.MemUsed != nil {
|
||||
updates["probe_mem_used"] = *m.MemUsed
|
||||
}
|
||||
if m.SwapTotal != nil {
|
||||
updates["probe_swap_total"] = *m.SwapTotal
|
||||
}
|
||||
if m.SwapUsed != nil {
|
||||
updates["probe_swap_used"] = *m.SwapUsed
|
||||
}
|
||||
if m.DiskTotal != nil {
|
||||
updates["probe_disk_total"] = *m.DiskTotal
|
||||
}
|
||||
if m.DiskUsed != nil {
|
||||
updates["probe_disk_used"] = *m.DiskUsed
|
||||
}
|
||||
if m.NetRxTotal != nil {
|
||||
updates["probe_net_rx_tot"] = *m.NetRxTotal
|
||||
}
|
||||
if m.NetTxTotal != nil {
|
||||
updates["probe_net_tx_tot"] = *m.NetTxTotal
|
||||
}
|
||||
if m.NetRx != nil {
|
||||
updates["probe_net_rx"] = *m.NetRx
|
||||
}
|
||||
if m.NetTx != nil {
|
||||
updates["probe_net_tx"] = *m.NetTx
|
||||
}
|
||||
if m.TCP != nil {
|
||||
updates["probe_tcp"] = *m.TCP
|
||||
}
|
||||
if m.UDP != nil {
|
||||
updates["probe_udp"] = *m.UDP
|
||||
}
|
||||
if m.Procs != nil {
|
||||
updates["probe_procs"] = *m.Procs
|
||||
}
|
||||
if m.CPU != nil {
|
||||
updates["probe_cpu"] = *m.CPU
|
||||
}
|
||||
}
|
||||
if len(updates) == 0 {
|
||||
return
|
||||
}
|
||||
_ = database.GetDB().Model(&model.AgentNode{}).Where("id = ?", id).Updates(updates).Error
|
||||
}
|
||||
|
||||
// heartbeatUpdateStatus is the agent's report about a one-click update
|
||||
// intent (item 20), piggybacked on the heartbeat like telemetry.
|
||||
type heartbeatUpdateStatus struct {
|
||||
@@ -329,10 +472,15 @@ func (a *AgentController) heartbeat(c *gin.Context) {
|
||||
var req struct {
|
||||
Telemetry *heartbeatTelemetry `json:"telemetry"`
|
||||
UpdateStatus *heartbeatUpdateStatus `json:"updateStatus"`
|
||||
ProbeFacts *heartbeatProbeFacts `json:"probeFacts"`
|
||||
ProbeMetrics *heartbeatProbeMetrics `json:"probeMetrics"`
|
||||
}
|
||||
_ = c.ShouldBindJSON(&req)
|
||||
a.touchAgent(uint(agent.Id), "")
|
||||
reportedVersion := agent.AgentVersion
|
||||
if req.ProbeFacts != nil || req.ProbeMetrics != nil {
|
||||
a.updateAgentProbe(uint(agent.Id), req.ProbeFacts, req.ProbeMetrics)
|
||||
}
|
||||
if req.Telemetry != nil {
|
||||
a.updateAgentTelemetry(uint(agent.Id), req.Telemetry)
|
||||
if req.Telemetry.AgentVersion != nil && *req.Telemetry.AgentVersion != "" {
|
||||
|
||||
@@ -0,0 +1,118 @@
|
||||
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"
|
||||
)
|
||||
|
||||
// Item 7 B1: the heartbeat/register optionally carry a monitor-probe-style
|
||||
// probe snapshot (facts + metrics). Reported fields land on the agent row;
|
||||
// a heartbeat without a probe block (old agents) must not clear stored
|
||||
// values, and a partial block only overwrites what it carries.
|
||||
func TestAgentHeartbeatProbe(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()
|
||||
|
||||
const token = "probe-test-token"
|
||||
agent := &model.AgentNode{
|
||||
TokenHash: hashAgentToken(token),
|
||||
TokenPrefix: "probe-",
|
||||
Name: "probe-node",
|
||||
}
|
||||
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(path, body string) int {
|
||||
req := httptest.NewRequest(http.MethodPost, path, 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
|
||||
}
|
||||
|
||||
facts := `"probeFacts":{"hostname":"host-1","os":"Debian GNU/Linux 13","kernel":"6.12.1",` +
|
||||
`"arch":"amd64","virt":"kvm","cpuName":"Test CPU","cpuCores":4,` +
|
||||
`"memTotal":4000000000,"swapTotal":1000000000,"diskTotal":8000000000,` +
|
||||
`"agentVersion":"A1.0.1","ipv4":"203.0.113.9","ipv6":"2001:db8::9"}`
|
||||
metrics := `"probeMetrics":{"bootId":"boot-x/0123456789abcdef","iface":"",` +
|
||||
`"uptime":3600,"cpu":12.5,"load":[0.5,0.25,0.125],` +
|
||||
`"memTotal":4000000000,"memUsed":1500000000,"swapTotal":1000000000,"swapUsed":5000000,` +
|
||||
`"diskTotal":8000000000,"diskUsed":3000000000,` +
|
||||
`"netRxTotal":111000,"netTxTotal":222000,"netRx":1000,"netTx":2000,` +
|
||||
`"tcp":42,"udp":7,"procs":123}`
|
||||
|
||||
// Register carries facts only.
|
||||
if code := post("/panel/api/agent/register",
|
||||
`{"token":"`+token+`","name":"probe-node",`+facts+`}`); code != http.StatusOK {
|
||||
t.Fatalf("register with facts: status %d", code)
|
||||
}
|
||||
row := load()
|
||||
if row.ProbeHostname != "host-1" || row.ProbeVirt != "kvm" || row.ProbeCPUCores != 4 ||
|
||||
row.ProbeMemTotal != 4000000000 || row.ProbeDiskTotal != 8000000000 {
|
||||
t.Fatalf("register facts not stored: %+v", row)
|
||||
}
|
||||
if row.ProbeAt != 0 {
|
||||
t.Fatalf("register must not stamp ProbeAt (no metrics yet): %d", row.ProbeAt)
|
||||
}
|
||||
|
||||
// Heartbeat with facts+metrics stores the latest snapshot.
|
||||
if code := post("/panel/api/agent/heartbeat",
|
||||
`{"token":"`+token+`","timestamp":1,`+facts+`,`+metrics+`}`); code != http.StatusOK {
|
||||
t.Fatalf("heartbeat with probe: status %d", code)
|
||||
}
|
||||
row = load()
|
||||
if row.ProbeBootID != "boot-x/0123456789abcdef" || row.ProbeMemUsed != 1500000000 ||
|
||||
row.ProbeDiskUsed != 3000000000 || row.ProbeNetRxTot != 111000 || row.ProbeNetTxTot != 222000 ||
|
||||
row.ProbeNetRx != 1000 || row.ProbeNetTx != 2000 || row.ProbeTCP != 42 || row.ProbeUDP != 7 ||
|
||||
row.ProbeProcs != 123 || row.ProbeLoad1 != 0.5 || row.ProbeLoad15 != 0.125 || row.ProbeAt == 0 {
|
||||
t.Fatalf("heartbeat probe not stored: %+v", row)
|
||||
}
|
||||
if row.ProbeCPU < 12.4 || row.ProbeCPU > 12.6 {
|
||||
t.Fatalf("probe cpu = %v, want 12.5", row.ProbeCPU)
|
||||
}
|
||||
|
||||
// Bare heartbeat (old agent): stored probe values survive untouched.
|
||||
if code := post("/panel/api/agent/heartbeat", `{"token":"`+token+`","timestamp":2}`); code != http.StatusOK {
|
||||
t.Fatalf("bare heartbeat: status %d", code)
|
||||
}
|
||||
row = load()
|
||||
if row.ProbeBootID != "boot-x/0123456789abcdef" || row.ProbeMemUsed != 1500000000 || row.ProbeHostname != "host-1" {
|
||||
t.Fatalf("bare heartbeat cleared probe: %+v", row)
|
||||
}
|
||||
|
||||
// Partial metrics: only the carried field changes.
|
||||
if code := post("/panel/api/agent/heartbeat",
|
||||
`{"token":"`+token+`","timestamp":3,"probeMetrics":{"netRx":7777}}`); code != http.StatusOK {
|
||||
t.Fatalf("partial heartbeat: status %d", code)
|
||||
}
|
||||
row = load()
|
||||
if row.ProbeNetRx != 7777 || row.ProbeNetTx != 2000 || row.ProbeMemUsed != 1500000000 {
|
||||
t.Fatalf("partial probe merged wrong: %+v", row)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user