package agent // probe.go is the platform-neutral core of the probe collector, ported // from monitor-probe's agent collector (collect.rs, MIT): host Facts // (once per process) and per-heartbeat Metrics with the source's // accounting discipline: // - memory "used" is free(1)'s column: total - available; // - disk "used" is df(1)'s: real storage only, total - free blocks // (not the unprivileged-available figure); // - network totals sum only "counted" interfaces — loopback, // container, tunnel and stacked devices are excluded by a filter so // no byte is counted twice; // - boot epoch = a stable boot identity + FNV-1a fingerprint of the // counted set, so a set change re-baselines exactly like a reboot; // - every source degrades independently: an unreadable source yields // a zero field, never a failed collection round. // // The samplers themselves are per-OS: probe_linux.go reads /proc and // /sys, probe_windows.go reads the Win32 performance/registry surface. // Both feed the shared rate/epoch machinery below and must keep the // same accounting semantics; where a source has no equivalent on the // platform, the field stays zero (documented at the sampler). import ( "hash/fnv" "net" "sort" "strings" "sync" "sync/atomic" "time" ) // 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 // counters 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 } // cpuSample is one cumulative CPU reading in whatever tick unit the // platform reports: total covers every state, idle only the idle state. // The differencing in cpuPercent is unit-agnostic. 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 } // statIfacePin (V2.0.1) names the single interface to count when // the operator pinned one on the master (its probe_meta.stat_iface, // delivered on the config poll). "" keeps the automatic counted-set // selection. The pin bypasses the filter on purpose: the operator // named this exact device, including when it is one the filter would // skip (a tunnel the node actually bills on). A pinned name absent // from the interface table counts nothing — the boot epoch changes // with the counted set, so the master re-baselines instead of booking // a phantom delta. var statIfacePin atomic.Value // string // SetStatIfacePin records the master's counted-interface pin. func SetStatIfacePin(name string) { statIfacePin.Store(name) } func statIfacePinValue() string { if v := statIfacePin.Load(); v != nil { return v.(string) } return "" } // netCounters is one interface's cumulative byte totals. type netCounters struct { name string rx, tx uint64 } func pinnedInterfaces(all []netCounters, pin string) []netCounters { for _, c := range all { if c.name == pin { return []netCounters{c} } } return nil } // cpuPercent diffs two cumulative samples into a busy percentage. A // missing or non-advancing previous sample reports 0 (the round only // re-baselines) — the source's discipline on both platforms. 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 } // bootEpoch = "{boot identity}/{FNV-1a-64 of the sorted counted names}". // The master only tests it for equality: any change re-baselines. func bootEpoch(bootID string, names []string) string { sorted := append([]string(nil), names...) sort.Strings(sorted) h := fnv.New64a() _, _ = h.Write([]byte(strings.Join(sorted, "\n"))) return bootID + "/" + 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) } // 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 }