Files
theta-agent/telemetry.go
T
wmantly 1837d18da8
release / agent-darwin-amd64 (push) Successful in 54s
release / agent-darwin-arm64 (push) Successful in 30s
release / agent-linux-amd64 (push) Successful in 29s
release / agent-linux-arm64 (push) Successful in 29s
release / agent-linux-arm7 (push) Successful in 28s
release / agent-windows-amd64 (push) Successful in 28s
release / agent-windows-arm64 (push) Successful in 28s
release / tray/helper/setup (push) Has been cancelled
release / Attach to GitHub release (push) Has been cancelled
release(v2.1.2): Linux mDNS local-discovery
Adds the CHANGELOG entry for this pass's mDNS work (local_discovery.go,
hosts_override.go -- see prior commit for the two real bugs found via
live testing), bumps AgentVersion, and documents prefer_local_directory
in agent.yml.example.

Also backfills CHANGELOG entries for v2.1.0/v2.1.1 (Windows agent,
WireGuard client, installer, CI), which were tagged and released but
never documented here.
2026-08-10 19:02:00 -04:00

669 lines
17 KiB
Go

package main
import (
"encoding/json"
"fmt"
"io"
"log"
"net"
"net/http"
"os"
"path/filepath"
"runtime"
"strings"
"time"
"github.com/gorilla/websocket"
"github.com/shirou/gopsutil/v3/cpu"
"github.com/shirou/gopsutil/v3/disk"
"github.com/shirou/gopsutil/v3/host"
"github.com/shirou/gopsutil/v3/mem"
)
type CPUDetails struct {
Model string `json:"model"`
Cores int `json:"cores"`
Threads int `json:"threads"`
MHz float64 `json:"mhz"`
}
type RAMDetails struct {
TotalBytes uint64 `json:"total_bytes"`
UsedBytes uint64 `json:"used_bytes"`
BuffersCacheBytes uint64 `json:"buffers_cache_bytes"`
FreeBytes uint64 `json:"free_bytes"`
UsedPercent float64 `json:"used_percent"`
BuffersCachePercent float64 `json:"buffers_cache_percent"`
FreePercent float64 `json:"free_percent"`
}
type DiskItem struct {
Mountpoint string `json:"mountpoint"`
Device string `json:"device"`
FSType string `json:"fstype"`
DriveType string `json:"drivetype"`
TotalBytes uint64 `json:"total_bytes"`
UsedBytes uint64 `json:"used_bytes"`
FreeBytes uint64 `json:"free_bytes"`
UsagePercent float64 `json:"usage_percent"`
}
type LoggedUser struct {
User string `json:"user"`
Terminal string `json:"terminal"`
Host string `json:"host"`
Started int64 `json:"started"`
}
type HostDetails struct {
StaticHostname string `json:"static_hostname"`
IconName string `json:"icon_name"`
Chassis string `json:"chassis"`
MachineID string `json:"machine_id"`
BootID string `json:"boot_id"`
OS string `json:"os"`
Kernel string `json:"kernel"`
Arch string `json:"arch"`
HardwareVendor string `json:"hardware_vendor"`
HardwareModel string `json:"hardware_model"`
FirmwareVersion string `json:"firmware_version"`
FirmwareDate string `json:"firmware_date"`
}
type DiscoveryData struct {
Hostname string `json:"hostname"`
IPs []string `json:"ip_addresses"`
PublicIP string `json:"public_ip"`
OS string `json:"os"`
Kernel string `json:"kernel"`
CPUModel string `json:"cpu"`
CPUDetails CPUDetails `json:"cpu_details"`
RAMTotalGB float64 `json:"ram_total_gb"`
RAMDetails RAMDetails `json:"ram_details"`
DiskTotalGB float64 `json:"disk_total_gb"`
Disks []DiskItem `json:"disks"`
LoggedUsers []LoggedUser `json:"logged_users"`
HostDetails HostDetails `json:"host_details"`
Version string `json:"version"`
Location string `json:"location"`
Capabilities map[string]interface{} `json:"capabilities"`
}
type TelemetryData struct {
CPUUsagePercent float64 `json:"cpu_usage_percent"`
CPUDetails CPUDetails `json:"cpu_details"`
RAMUsagePercent float64 `json:"ram_usage_percent"`
RAMDetails RAMDetails `json:"ram_details"`
DiskUsagePercent float64 `json:"disk_usage_percent"`
Disks []DiskItem `json:"disks"`
LoggedUsers []LoggedUser `json:"logged_users"`
HostDetails HostDetails `json:"host_details"`
Version string `json:"version"`
ZFSHealth string `json:"zfs_health,omitempty"`
GPUUsage float64 `json:"gpu_usage_percent,omitempty"`
Timestamp string `json:"timestamp"`
}
func getPublicIP() string {
client := &http.Client{Timeout: 3 * time.Second}
endpoints := []string{
"https://api.ipify.org",
"https://ifconfig.me/ip",
"https://icanhazip.com",
}
for _, ep := range endpoints {
resp, err := client.Get(ep)
if err == nil && resp.StatusCode == 200 {
body, err := io.ReadAll(resp.Body)
resp.Body.Close()
if err == nil {
ip := strings.TrimSpace(string(body))
if net.ParseIP(ip) != nil {
return ip
}
}
}
}
return ""
}
func collectCPUDetails() CPUDetails {
cpuInfo, _ := cpu.Info()
model := "Unknown"
cores := 0
mhz := 0.0
if len(cpuInfo) > 0 {
model = cpuInfo[0].ModelName
if model == "" || strings.TrimSpace(model) == "154" || len(model) < 4 {
if data, err := os.ReadFile("/proc/cpuinfo"); err == nil {
for _, line := range strings.Split(string(data), "\n") {
if strings.HasPrefix(line, "model name") {
parts := strings.Split(line, ":")
if len(parts) > 1 {
model = strings.TrimSpace(parts[1])
break
}
}
}
}
}
if model == "" {
model = cpuInfo[0].Model
}
cores = int(cpuInfo[0].Cores)
mhz = cpuInfo[0].Mhz
}
threads := runtime.NumCPU()
if t, err := cpu.Counts(true); err == nil && t > 0 {
threads = t
}
if cores <= 0 {
if c, err := cpu.Counts(false); err == nil && c > 0 {
cores = c
} else {
cores = threads
}
}
return CPUDetails{
Model: model,
Cores: cores,
Threads: threads,
MHz: mhz,
}
}
func collectRAMDetails() RAMDetails {
vm, err := mem.VirtualMemory()
if err != nil || vm == nil {
return RAMDetails{}
}
bufCache := vm.Buffers + vm.Cached
total := float64(vm.Total)
usedPct := 0.0
bufPct := 0.0
freePct := 0.0
if total > 0 {
usedPct = (float64(vm.Used) / total) * 100.0
bufPct = (float64(bufCache) / total) * 100.0
freePct = (float64(vm.Free) / total) * 100.0
}
return RAMDetails{
TotalBytes: vm.Total,
UsedBytes: vm.Used,
BuffersCacheBytes: bufCache,
FreeBytes: vm.Free,
UsedPercent: usedPct,
BuffersCachePercent: bufPct,
FreePercent: freePct,
}
}
func getDriveType(device string) string {
devName := filepath.Base(device)
devName = strings.TrimRight(devName, "0123456789p")
if strings.HasPrefix(devName, "nvme") {
return "NVMe"
}
rotPath := filepath.Join("/sys/block", devName, "queue/rotational")
data, err := os.ReadFile(rotPath)
if err == nil {
val := strings.TrimSpace(string(data))
if val == "0" {
return "SSD"
} else if val == "1" {
return "HDD"
}
}
return "SSD/HDD"
}
func collectLoggedUsers() []LoggedUser {
var list []LoggedUser
seen := make(map[string]bool)
// 1. Try loginctl list-sessions --no-legend (systemd logind)
exec := SystemExecutor{}
if out, err := exec.Execute("loginctl", "list-sessions", "--no-legend"); err == nil && len(out) > 0 {
lines := strings.Split(string(out), "\n")
for _, line := range lines {
fields := strings.Fields(line)
// Format: SESSION UID USER SEAT TTY STATE IDLE SINCE
// e.g. c2 1000 william seat0 tty7 active no -
if len(fields) >= 3 {
user := fields[2]
term := ""
if len(fields) >= 5 && fields[4] != "-" {
term = fields[4]
}
key := fmt.Sprintf("%s@%s", user, term)
if !seen[key] && user != "" {
seen[key] = true
list = append(list, LoggedUser{
User: user,
Terminal: term,
Host: "localhost",
Started: time.Now().Unix(),
})
}
}
}
}
// 2. Fallback to gopsutil / who if loginctl returned nothing
if len(list) == 0 {
users, err := host.Users()
if err == nil {
for _, u := range users {
key := fmt.Sprintf("%s@%s:%s", u.User, u.Terminal, u.Host)
if !seen[key] {
seen[key] = true
list = append(list, LoggedUser{
User: u.User,
Terminal: u.Terminal,
Host: u.Host,
Started: int64(u.Started),
})
}
}
}
if len(list) == 0 {
out, err := exec.Execute("who")
if err == nil && len(out) > 0 {
lines := strings.Split(string(out), "\n")
for _, line := range lines {
fields := strings.Fields(line)
if len(fields) >= 2 {
user := fields[0]
term := fields[1]
hostStr := ""
if len(fields) >= 5 {
hostStr = strings.Trim(fields[4], "()")
}
key := fmt.Sprintf("%s@%s:%s", user, term, hostStr)
if !seen[key] {
seen[key] = true
list = append(list, LoggedUser{
User: user,
Terminal: term,
Host: hostStr,
Started: time.Now().Unix(),
})
}
}
}
}
}
}
return list
}
func collectDiskItems() []DiskItem {
var items []DiskItem
partitions, err := disk.Partitions(true)
if err != nil || len(partitions) == 0 {
d, err2 := disk.Usage("/")
if err2 == nil {
items = append(items, DiskItem{
Mountpoint: "/",
Device: d.Path,
FSType: d.Fstype,
DriveType: getDriveType(d.Path),
TotalBytes: d.Total,
UsedBytes: d.Used,
FreeBytes: d.Free,
UsagePercent: d.UsedPercent,
})
}
return items
}
seen := make(map[string]bool)
for _, p := range partitions {
if !strings.HasPrefix(p.Device, "/dev/") || strings.HasPrefix(p.Device, "/dev/loop") {
continue
}
if seen[p.Mountpoint] {
continue
}
seen[p.Mountpoint] = true
u, err := disk.Usage(p.Mountpoint)
if err != nil || u.Total == 0 {
continue
}
fstype := p.Fstype
if fstype == "" {
fstype = u.Fstype
}
items = append(items, DiskItem{
Mountpoint: p.Mountpoint,
Device: p.Device,
FSType: fstype,
DriveType: getDriveType(p.Device),
TotalBytes: u.Total,
UsedBytes: u.Used,
FreeBytes: u.Free,
UsagePercent: u.UsedPercent,
})
}
if len(items) == 0 {
d, err2 := disk.Usage("/")
if err2 == nil {
items = append(items, DiskItem{
Mountpoint: "/",
Device: d.Path,
FSType: d.Fstype,
DriveType: getDriveType(d.Path),
TotalBytes: d.Total,
UsedBytes: d.Used,
FreeBytes: d.Free,
UsagePercent: d.UsedPercent,
})
}
}
return items
}
func collectHostDetails() HostDetails {
details := HostDetails{}
exec := SystemExecutor{}
out, err := exec.Execute("hostnamectl")
if err == nil {
for _, line := range strings.Split(string(out), "\n") {
parts := strings.SplitN(line, ":", 2)
if len(parts) == 2 {
key := strings.TrimSpace(parts[0])
val := strings.TrimSpace(parts[1])
switch key {
case "Static hostname":
details.StaticHostname = val
case "Icon name":
details.IconName = val
case "Chassis":
details.Chassis = val
case "Machine ID":
details.MachineID = val
case "Boot ID":
details.BootID = val
case "Operating System":
details.OS = val
case "Kernel":
details.Kernel = val
case "Architecture":
details.Arch = val
case "Hardware Vendor":
details.HardwareVendor = val
case "Hardware Model":
details.HardwareModel = val
case "Firmware Version":
details.FirmwareVersion = val
case "Firmware Date":
details.FirmwareDate = val
}
}
}
}
if details.HardwareVendor == "" {
if d, err := os.ReadFile("/sys/class/dmi/id/sys_vendor"); err == nil {
details.HardwareVendor = strings.TrimSpace(string(d))
}
}
if details.HardwareModel == "" {
if d, err := os.ReadFile("/sys/class/dmi/id/product_name"); err == nil {
details.HardwareModel = strings.TrimSpace(string(d))
}
}
if details.FirmwareVersion == "" {
if d, err := os.ReadFile("/sys/class/dmi/id/bios_version"); err == nil {
details.FirmwareVersion = strings.TrimSpace(string(d))
}
}
if details.FirmwareDate == "" {
if d, err := os.ReadFile("/sys/class/dmi/id/bios_date"); err == nil {
details.FirmwareDate = strings.TrimSpace(string(d))
}
}
return details
}
const AgentVersion = "v2.1.2"
// CollectDiscoveryData gathers static host information.
func CollectDiscoveryData(cfg *Config) DiscoveryData {
h, _ := host.Info()
var ips []string
addrs, _ := net.InterfaceAddrs()
for _, addr := range addrs {
if ipnet, ok := addr.(*net.IPNet); ok && !ipnet.IP.IsLoopback() {
if ipnet.IP.To4() != nil {
ips = append(ips, ipnet.IP.String())
}
}
}
vm := collectRAMDetails()
disks := collectDiskItems()
cpuDet := collectCPUDetails()
loggedUsers := collectLoggedUsers()
pubIP := ""
if cfg.DetectPublicIP() {
pubIP = getPublicIP()
}
diskTotalGB := 0.0
for _, d := range disks {
if d.Mountpoint == "/" {
diskTotalGB = float64(d.TotalBytes) / (1024 * 1024 * 1024)
break
}
}
if diskTotalGB == 0 && len(disks) > 0 {
diskTotalGB = float64(disks[0].TotalBytes) / (1024 * 1024 * 1024)
}
hostDet := collectHostDetails()
return DiscoveryData{
Hostname: h.Hostname,
IPs: ips,
PublicIP: pubIP,
OS: fmt.Sprintf("%s %s", h.OS, h.Platform),
Kernel: h.KernelVersion,
CPUModel: cpuDet.Model,
CPUDetails: cpuDet,
RAMTotalGB: float64(vm.TotalBytes) / (1024 * 1024 * 1024),
RAMDetails: vm,
DiskTotalGB: diskTotalGB,
Disks: disks,
LoggedUsers: loggedUsers,
HostDetails: hostDet,
Version: AgentVersion,
Location: cfg.Location,
Capabilities: map[string]interface{}{
"telemetry": cfg.Capabilities.Telemetry,
"configure_ldap": cfg.Capabilities.ConfigureLDAP,
"ldap_tunnel": cfg.Capabilities.LdapTunnel,
"secrets": cfg.Capabilities.Secrets,
"iam": cfg.Capabilities.IAM,
"reboot": cfg.Capabilities.Reboot,
"shutdown": true,
"desktop_controls": true,
"service_control": cfg.Capabilities.ServiceControl,
"arbitrary_bash": cfg.Capabilities.ArbitraryBash,
},
}
}
// CollectTelemetryData gathers real-time performance metrics including ZFS and GPU.
func CollectTelemetryData(exec Executor) TelemetryData {
cpuPerc, _ := cpu.Percent(time.Second, false)
vm := collectRAMDetails()
disks := collectDiskItems()
cpuDet := collectCPUDetails()
loggedUsers := collectLoggedUsers()
hostDet := collectHostDetails()
cpuVal := 0.0
if len(cpuPerc) > 0 {
cpuVal = cpuPerc[0]
}
diskVal := 0.0
for _, d := range disks {
if d.Mountpoint == "/" {
diskVal = d.UsagePercent
break
}
}
if diskVal == 0 && len(disks) > 0 {
diskVal = disks[0].UsagePercent
}
return TelemetryData{
CPUUsagePercent: cpuVal,
CPUDetails: cpuDet,
RAMUsagePercent: vm.UsedPercent,
RAMDetails: vm,
DiskUsagePercent: diskVal,
Disks: disks,
LoggedUsers: loggedUsers,
HostDetails: hostDet,
Version: AgentVersion,
ZFSHealth: collectZFSHealth(exec),
GPUUsage: collectGPUUsage(exec),
Timestamp: time.Now().Format(time.RFC3339),
}
}
func collectZFSHealth(exec Executor) string {
out, err := exec.Execute("zpool", "list", "-H", "-o", "health")
if err != nil {
return "unknown"
}
lines := strings.Split(strings.TrimSpace(string(out)), "\n")
if len(lines) > 0 {
return lines[0]
}
return "unknown"
}
func collectGPUUsage(exec Executor) float64 {
out, err := exec.Execute("nvidia-smi", "--query-gpu=utilization.gpu", "--format=csv,noheader,nounits")
if err != nil {
return -1.0
}
var usage float64
fmt.Sscanf(strings.TrimSpace(string(out)), "%f", &usage)
return usage
}
// StartTelemetryLoop manages the initial discovery push and the periodic telemetry stream.
func StartTelemetryLoop(c MessageWriter, cm *ConfigManager, exec Executor, stopCh <-chan struct{}) {
cfg := cm.Get()
// 1. Immediate Discovery Push & Initial Telemetry Frame
pushDiscovery(c, cfg)
pushTelemetry(c, exec)
// If telemetry capability is disabled in agent.yml, return early after discovery
if !cfg.Capabilities.Telemetry {
log.Println("Telemetry capability is disabled in agent.yml; skipping telemetry stream.")
return
}
// 2. Periodic Telemetry Stream
ticker := time.NewTicker(30 * time.Second)
go func() {
defer ticker.Stop()
var lastIPs []string
for {
select {
case <-stopCh:
return
case <-ticker.C:
currentCFG := cm.Get()
if !currentCFG.Capabilities.Telemetry {
continue
}
// Network Change Detection
currentIPs := collectIPs()
if !equalSlices(lastIPs, currentIPs) {
log.Println("Network change detected. Pushing discovery update...")
pushDiscovery(c, currentCFG)
lastIPs = currentIPs
}
pushTelemetry(c, exec)
}
}
}()
}
func pushTelemetry(c MessageWriter, exec Executor) {
telemetry := CollectTelemetryData(exec)
payload, _ := json.Marshal(WSMessage{
Type: "telemetry",
Payload: map[string]interface{}{
"cpu_usage_percent": telemetry.CPUUsagePercent,
"cpu_details": telemetry.CPUDetails,
"ram_usage_percent": telemetry.RAMUsagePercent,
"ram_details": telemetry.RAMDetails,
"disk_usage_percent": telemetry.DiskUsagePercent,
"disks": telemetry.Disks,
"logged_users": telemetry.LoggedUsers,
"host_details": telemetry.HostDetails,
"zfs_health": telemetry.ZFSHealth,
"gpu_usage_percent": telemetry.GPUUsage,
"timestamp": telemetry.Timestamp,
},
})
_ = c.WriteMessage(websocket.TextMessage, payload)
}
func collectIPs() []string {
var ips []string
addrs, _ := net.InterfaceAddrs()
for _, addr := range addrs {
if ipnet, ok := addr.(*net.IPNet); ok && !ipnet.IP.IsLoopback() {
if ipnet.IP.To4() != nil {
ips = append(ips, ipnet.IP.String())
}
}
}
return ips
}
func equalSlices(a, b []string) bool {
if len(a) != len(b) {
return false
}
for i := range a {
if a[i] != b[i] {
return false
}
}
return true
}
func pushDiscovery(c MessageWriter, cfg *Config) {
discovery := CollectDiscoveryData(cfg)
discoveryPayload, _ := json.Marshal(discovery)
var discoveryMap map[string]interface{}
json.Unmarshal(discoveryPayload, &discoveryMap)
msg := WSMessage{
Type: "discovery",
Payload: discoveryMap,
}
payload, _ := json.Marshal(msg)
if err := c.WriteMessage(websocket.TextMessage, payload); err != nil {
log.Printf("Failed to send discovery data: %v", err)
} else {
log.Println("Discovery data pushed to Theta Directory.")
}
}