feat: machine status page, speed test, live presence, forwards without ufw (0.0.5)

Status page per machine: CPU load, RAM and disk of the client, sampled
by the agent only while the page is open and visible (one reading
serves every open tab), plus a button that has the agent measure its
link to the host: ping, down and up.

Online/offline now comes from the control polls the hub holds open, so
a dropped agent shows offline at once instead of 45 s later, and the
dashboard keeps the status column live.

A forward rule can be created "without ufw": the panel opens nothing in
ufw for it and so revokes nothing when it is deleted. The checkbox is
shown only while the ufw integration is on.

Dashboard row actions are icons; a status action is added.

internal/agent/speed.go

@@ -0,0 +1,122 @@
+package agent
+
+import (
+	"context"
+	"crypto/rand"
+	"encoding/json"
+	"io"
+	"net/http"
+	"time"
+
+	"github.com/x-grk/longpolling-remote-access/internal/proto"
+)
+
+const (
+	// speedWindow is how long each direction is measured for.
+	speedWindow = 3 * time.Second
+	// speedUpCap bounds the upload, whatever the link does in speedWindow.
+	speedUpCap = 48 << 20
+)
+
+// doSpeed measures the agent's HTTP link to the server: round-trip time of an
+// empty request, then how much flows each way in speedWindow.
+func (a *Agent) doSpeed(ctx context.Context) proto.TaskResult {
+	res, err := a.measureSpeed(ctx)
+	if err != nil {
+		return proto.TaskResult{Error: err.Error()}
+	}
+	data, _ := json.Marshal(res)
+	return proto.TaskResult{OK: true, Data: data}
+}
+
+func (a *Agent) measureSpeed(ctx context.Context) (proto.SysSpeedResult, error) {
+	var res proto.SysSpeedResult
+	// Ping: the first request may pay for a new connection, so keep the best.
+	for i := 0; i < 4; i++ {
+		start := time.Now()
+		resp, err := a.speedRequest(ctx, http.MethodGet, "/api/agent/speed/down?bytes=0", nil)
+		if err != nil {
+			return res, err
+		}
+		io.Copy(io.Discard, resp.Body)
+		resp.Body.Close()
+		ms := float64(time.Since(start).Microseconds()) / 1000
+		if i == 0 || ms < res.PingMs {
+			res.PingMs = ms
+		}
+	}
+
+	// Down: read what the server streams until the window closes.
+	dctx, cancel := context.WithTimeout(ctx, speedWindow)
+	resp, err := a.speedRequest(dctx, http.MethodGet, "/api/agent/speed/down", nil)
+	if err != nil {
+		cancel()
+		return res, err
+	}
+	start := time.Now()
+	n, _ := io.Copy(io.Discard, resp.Body)
+	elapsed := time.Since(start)
+	resp.Body.Close()
+	cancel()
+	res.DownBps = bps(n, elapsed)
+
+	// Up: send until the window closes or the cap is hit.
+	body := &speedSource{deadline: time.Now().Add(speedWindow), left: speedUpCap}
+	start = time.Now()
+	resp, err = a.speedRequest(ctx, http.MethodPost, "/api/agent/speed/up", body)
+	if err != nil {
+		return res, err
+	}
+	io.Copy(io.Discard, resp.Body)
+	resp.Body.Close()
+	res.UpBps = bps(int64(speedUpCap-body.left), time.Since(start))
+	return res, nil
+}
+
+func (a *Agent) speedRequest(ctx context.Context, method, path string, body io.Reader) (*http.Response, error) {
+	req, err := http.NewRequestWithContext(ctx, method, a.cfg.Host+path, body)
+	if err != nil {
+		return nil, err
+	}
+	a.authHeaders(req)
+	resp, err := a.http.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusNoContent {
+		resp.Body.Close()
+		return nil, errStatus(resp.StatusCode)
+	}
+	return resp, nil
+}
+
+func bps(n int64, d time.Duration) float64 {
+	if d <= 0 {
+		return 0
+	}
+	return float64(n) * 8 / d.Seconds()
+}
+
+// speedSource is an upload body of incompressible bytes that ends at the
+// deadline or after `left` bytes.
+type speedSource struct {
+	deadline time.Time
+	left     int
+	block    []byte
+}
+
+func (s *speedSource) Read(p []byte) (int, error) {
+	if s.left <= 0 || !time.Now().Before(s.deadline) {
+		return 0, io.EOF
+	}
+	if s.block == nil {
+		s.block = make([]byte, 64<<10)
+		rand.Read(s.block)
+	}
+	n := copy(p, s.block)
+	if n > s.left {
+		n = s.left
+	}
+	s.left -= n
+	return n, nil
+}

internal/agent/stat.go

@@ -0,0 +1,45 @@
+package agent
+
+import (
+	"runtime"
+	"sync"
+	"syscall"
+
+	"github.com/x-grk/longpolling-remote-access/internal/proto"
+)
+
+// statMu serializes samples: the CPU figure is a delta against the last one.
+var statMu sync.Mutex
+
+// sysStat samples the machine's load. It costs nothing while nobody asks: the
+// panel requests it only for an open status page.
+func sysStat() (proto.SysStatResult, error) {
+	statMu.Lock()
+	defer statMu.Unlock()
+	res := proto.SysStatResult{NCPU: runtime.NumCPU()}
+	cpu, err := cpuPercent()
+	if err != nil {
+		return res, err
+	}
+	res.CPU = cpu
+	if res.MemTotal, res.MemUsed, err = memUsage(); err != nil {
+		return res, err
+	}
+	var fs syscall.Statfs_t
+	if err := syscall.Statfs("/", &fs); err != nil {
+		return res, err
+	}
+	res.DiskTotal = uint64(fs.Blocks) * uint64(fs.Bsize)
+	res.DiskFree = uint64(fs.Bavail) * uint64(fs.Bsize)
+	return res, nil
+}
+
+func clampPercent(v float64) float64 {
+	if v < 0 {
+		return 0
+	}
+	if v > 100 {
+		return 100
+	}
+	return v
+}

internal/agent/stat_darwin.go

@@ -0,0 +1,43 @@
+package agent
+
+import (
+	"bytes"
+	"os/exec"
+	"runtime"
+	"strconv"
+
+	"golang.org/x/sys/unix"
+)
+
+// cpuPercent sums the per-process CPU shares ps reports. macOS exposes the
+// host CPU ticks only through Mach calls, which need cgo.
+func cpuPercent() (float64, error) {
+	out, err := exec.Command("/bin/ps", "-A", "-o", "%cpu=").Output()
+	if err != nil {
+		return 0, err
+	}
+	var sum float64
+	for _, f := range bytes.Fields(out) {
+		// Some locales print a decimal comma.
+		v, _ := strconv.ParseFloat(string(bytes.Replace(f, []byte(","), []byte("."), 1)), 64)
+		sum += v
+	}
+	return clampPercent(sum / float64(runtime.NumCPU())), nil
+}
+
+// memUsage takes the kernel's own share of memory still available
+// (kern.memorystatus_level, what `memory_pressure` prints).
+func memUsage() (total, used uint64, err error) {
+	total, err = unix.SysctlUint64("hw.memsize")
+	if err != nil {
+		return 0, 0, err
+	}
+	level, err := unix.SysctlUint32("kern.memorystatus_level")
+	if err != nil {
+		return 0, 0, err
+	}
+	if level > 100 {
+		level = 100
+	}
+	return total, total / 100 * uint64(100-level), nil
+}

internal/agent/stat_linux.go

@@ -0,0 +1,121 @@
+package agent
+
+import (
+	"bytes"
+	"errors"
+	"os"
+	"strconv"
+	"time"
+)
+
+// Previous /proc/stat reading; the load is the busy share of the ticks since.
+var (
+	cpuPrevIdle, cpuPrevTotal uint64
+	cpuPrevAt                 time.Time
+)
+
+// cpuStaleAfter is how old the previous reading may be: past it the delta
+// would average over a stretch nobody was watching, so take a fresh pair.
+const cpuStaleAfter = 15 * time.Second
+
+func cpuPercent() (float64, error) {
+	if time.Since(cpuPrevAt) > cpuStaleAfter {
+		idle, total, err := readCPUTicks()
+		if err != nil {
+			return 0, err
+		}
+		cpuPrevIdle, cpuPrevTotal = idle, total
+		time.Sleep(200 * time.Millisecond)
+	}
+	idle, total, err := readCPUTicks()
+	if err != nil {
+		return 0, err
+	}
+	dIdle, dTotal := idle-cpuPrevIdle, total-cpuPrevTotal
+	cpuPrevIdle, cpuPrevTotal, cpuPrevAt = idle, total, time.Now()
+	if dTotal == 0 || dIdle > dTotal {
+		return 0, nil
+	}
+	return clampPercent(100 * float64(dTotal-dIdle) / float64(dTotal)), nil
+}
+
+// readCPUTicks parses the aggregate "cpu" line of /proc/stat:
+// user nice system idle iowait irq softirq steal ...
+func readCPUTicks() (idle, total uint64, err error) {
+	var buf [512]byte
+	f, err := os.Open("/proc/stat")
+	if err != nil {
+		return 0, 0, err
+	}
+	n, _ := f.Read(buf[:])
+	f.Close()
+	line := buf[:n]
+	if i := bytes.IndexByte(line, '\n'); i >= 0 {
+		line = line[:i]
+	}
+	fields := bytes.Fields(line)
+	if len(fields) < 5 || string(fields[0]) != "cpu" {
+		return 0, 0, errors.New("unexpected /proc/stat")
+	}
+	for i, f := range fields[1:] {
+		if i >= 8 { // guest time is already counted in user/nice
+			break
+		}
+		v, _ := strconv.ParseUint(string(f), 10, 64)
+		total += v
+		if i == 3 || i == 4 { // idle, iowait
+			idle += v
+		}
+	}
+	return idle, total, nil
+}
+
+// memUsage reads /proc/meminfo; used is everything not available to new
+// processes (so page cache does not count as used).
+func memUsage() (total, used uint64, err error) {
+	data, err := os.ReadFile("/proc/meminfo")
+	if err != nil {
+		return 0, 0, err
+	}
+	var avail, free uint64
+	var haveAvail bool
+	for len(data) > 0 {
+		line := data
+		if i := bytes.IndexByte(data, '\n'); i >= 0 {
+			line, data = data[:i], data[i+1:]
+		} else {
+			data = nil
+		}
+		switch {
+		case bytes.HasPrefix(line, []byte("MemTotal:")):
+			total = meminfoKB(line)
+		case bytes.HasPrefix(line, []byte("MemFree:")):
+			free = meminfoKB(line)
+		case bytes.HasPrefix(line, []byte("MemAvailable:")):
+			avail, haveAvail = meminfoKB(line), true
+		}
+		if haveAvail && total > 0 {
+			break
+		}
+	}
+	if total == 0 {
+		return 0, 0, errors.New("unexpected /proc/meminfo")
+	}
+	if !haveAvail {
+		avail = free
+	}
+	if avail > total {
+		avail = total
+	}
+	return total, total - avail, nil
+}
+
+// meminfoKB returns the value of a "Key:   123 kB" line in bytes.
+func meminfoKB(line []byte) uint64 {
+	f := bytes.Fields(line)
+	if len(f) < 2 {
+		return 0
+	}
+	v, _ := strconv.ParseUint(string(f[1]), 10, 64)
+	return v * 1024
+}

internal/agent/task.go

@@ -17,9 +17,12 @@ func (a *Agent) runTask(ctx context.Context, taskID string) {
 		return
 	}
 	var res proto.TaskResult
-	if env.Type == proto.TaskSysUpdate {
+	switch env.Type {
+	case proto.TaskSysUpdate:
 		res = a.doUpdate()
-	} else {
+	case proto.TaskSysSpeed:
+		res = a.doSpeed(ctx)
+	default:
 		res = a.execTask(env)
 	}
 	a.postResult(ctx, taskID, res)

internal/agent/users.go

@@ -33,6 +33,8 @@ func dispatchTask(env proto.TaskEnvelope) (any, error) {
 	switch env.Type {
 	case proto.TaskSysUsers:
 		return sysUsers(), nil
+	case proto.TaskSysStat:
+		return sysStat()
 	case proto.TaskSysUserAdd:
 		var a proto.SysUserAddArgs
 		_ = json.Unmarshal(env.Args, &a)