worker/internal/webapp/metrics.go

package webapp

import (
	"context"
	"fmt"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"sync"
	"time"
)

// Metrics owns the host-level status sample used by the overview and
// server-status pages. Phase 1 reads only /proc and statfs over
// mount points. CLI tools (lsblk, smartctl, sensors) are Phase 5.
type Metrics struct {
	mu      sync.RWMutex
	last    Snapshot
	lastAt  time.Time
	stopCh  chan struct{}
	stopWG  sync.WaitGroup
	started bool
}

// Snapshot is the JSON-friendly view of host metrics the templates
// render. The fields are picked so the status table on /status and
// the cards on /overview share a single type.
type Snapshot struct {
	CPU      CPUSample     `json:"cpu"`
	Memory   MemorySample  `json:"memory"`
	Load     LoadSample    `json:"load"`
	Uptime   time.Duration `json:"uptime"`
	BootAt   time.Time     `json:"boot_at"`
	Networks []NetDev      `json:"networks"`
	Disks    []DiskSample  `json:"disks"`
}

// CPUSample reports aggregate CPU usage since the last sample. The
// fields are percentages normalised to 0..100.
type CPUSample struct {
	UserPct   float64 `json:"user_pct"`
	NicePct   float64 `json:"nice_pct"`
	SystemPct float64 `json:"system_pct"`
	IdlePct   float64 `json:"idle_pct"`
	IOWaitPct float64 `json:"iowait_pct"`
	StealPct  float64 `json:"steal_pct"`
	TotalPct  float64 `json:"total_pct"`
}

// MemorySample reports bytes of physical RAM, swap, and various
// accounting fields from /proc/meminfo.
type MemorySample struct {
	Total        uint64  `json:"total_bytes"`
	Available    uint64  `json:"available_bytes"`
	Free         uint64  `json:"free_bytes"`
	Buffers      uint64  `json:"buffers_bytes"`
	Cached       uint64  `json:"cached_bytes"`
	SwapTotal    uint64  `json:"swap_total_bytes"`
	SwapFree     uint64  `json:"swap_free_bytes"`
	UsedPct      float64 `json:"used_pct"`
	AvailablePct float64 `json:"available_pct"`
}

// LoadSample is the 1/5/15 minute load averages from /proc/loadavg.
type LoadSample struct {
	One     float64 `json:"load1"`
	Five    float64 `json:"load5"`
	Fifteen float64 `json:"load15"`
}

// NetDev is one row from /proc/net/dev.
type NetDev struct {
	Name    string `json:"name"`
	RxBytes uint64 `json:"rx_bytes"`
	TxBytes uint64 `json:"tx_bytes"`
	RxPkt   uint64 `json:"rx_packets"`
	TxPkt   uint64 `json:"tx_packets"`
	RxErr   uint64 `json:"rx_errors"`
	TxErr   uint64 `json:"tx_errors"`
	RxDrop  uint64 `json:"rx_dropped"`
	TxDrop  uint64 `json:"tx_dropped"`
}

// DiskSample is a single mount point from /proc/mounts with disk
// usage from statfs(2).
type DiskSample struct {
	Mount   string  `json:"mount"`
	Device  string  `json:"device"`
	FSType  string  `json:"fstype"`
	Total   uint64  `json:"total_bytes"`
	Free    uint64  `json:"free_bytes"`
	Used    uint64  `json:"used_bytes"`
	UsedPct float64 `json:"used_pct"`
}

// NewMetrics constructs an empty Metrics sampler. The loop is not
// started until Start is called.
func NewMetrics() *Metrics {
	return &Metrics{
		stopCh: make(chan struct{}),
	}
}

// Start launches a sample loop. The first sample is taken
// immediately so /status never renders "no data yet".
func (m *Metrics) Start(ctx context.Context) {
	m.mu.Lock()
	if m.started {
		m.mu.Unlock()
		return
	}
	m.started = true
	m.mu.Unlock()

	m.sample(ctx)
	m.stopWG.Add(1)
	go m.loop(ctx)
}

// Stop cancels the sample loop and waits for it to exit.
func (m *Metrics) Stop() {
	m.mu.Lock()
	if !m.started {
		m.mu.Unlock()
		return
	}
	select {
	case <-m.stopCh:
	default:
		close(m.stopCh)
	}
	m.mu.Unlock()
	m.stopWG.Wait()
}

// Last returns the most recent snapshot and the time it was taken.
// Always safe to call; returns a zero-value snapshot if the first
// sample has not yet completed.
func (m *Metrics) Last() (Snapshot, time.Time) {
	m.mu.RLock()
	defer m.mu.RUnlock()
	return m.last, m.lastAt
}

func (m *Metrics) loop(ctx context.Context) {
	defer m.stopWG.Done()
	ticker := time.NewTicker(5 * time.Second)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-m.stopCh:
			return
		case <-ticker.C:
			m.sample(ctx)
		}
	}
}

func (m *Metrics) sample(_ context.Context) {
	snap, err := CollectSnapshot(ProcRoot())
	if err != nil {
		return // best-effort
	}
	m.mu.Lock()
	m.last = snap
	m.lastAt = time.Now().UTC()
	m.mu.Unlock()
}

// CollectSnapshot reads the /proc mount once and returns a Snapshot.
// Exposed at package scope so tests can drive it directly with a
// fixture /proc tree.
func CollectSnapshot(root string) (Snapshot, error) {
	now := time.Now().UTC()
	cpu, err := readProcStat(filepath.Join(root, "stat"))
	if err != nil {
		return Snapshot{}, fmt.Errorf("read stat: %w", err)
	}
	mem, err := readMemInfo(filepath.Join(root, "meminfo"))
	if err != nil {
		return Snapshot{}, fmt.Errorf("read meminfo: %w", err)
	}
	load, err := readLoadAvg(filepath.Join(root, "loadavg"))
	if err != nil {
		return Snapshot{}, fmt.Errorf("read loadavg: %w", err)
	}
	uptime, err := readUptime(filepath.Join(root, "uptime"))
	if err != nil {
		return Snapshot{}, fmt.Errorf("read uptime: %w", err)
	}
	net, err := readNetDev(filepath.Join(root, "net", "dev"))
	if err != nil {
		return Snapshot{}, fmt.Errorf("read net/dev: %w", err)
	}
	disks, err := readMounts(filepath.Join(root, "mounts"))
	if err != nil {
		return Snapshot{}, fmt.Errorf("read mounts: %w", err)
	}
	for i := range disks {
		if err := statDisk(disks[i].Mount, &disks[i]); err != nil {
			// statfs may fail for some pseudo mounts (proc, sys);
			// we leave Total/Free/Used at zero in that case so the
			// page renders an empty row rather than a hard error.
			continue
		}
	}
	bootAt := now.Add(-uptime)
	return Snapshot{
		CPU:      cpu,
		Memory:   mem,
		Load:     load,
		Uptime:   uptime,
		BootAt:   bootAt,
		Networks: net,
		Disks:    disks,
	}, nil
}

// readProcStat reads the aggregate "cpu " row of /proc/stat and
// returns percentages. /proc/stat is cumulative since boot, so a
// single read yields busy/total ratios only if we remember the
// previous delta. Phase 1 does not keep history; the per-CPU
// "busy since boot" snapshot is rendered on the page as a static
// "load since boot" indicator instead of a live % value.
//
// The function takes a "previous" sample for delta math; if prev
// is the zero value, the function returns zero percentages.
func readProcStat(path string) (CPUSample, error) {
	f, err := os.Open(path)
	if err != nil {
		return CPUSample{}, err
	}
	defer f.Close() //nolint:errcheck

	var (
		user, nice, system, idle, iowait, steal uint64
		agg                                     bool
	)
	scanner := bufioNewScanner(f)
	for scanner.Scan() {
		line := scanner.Text()
		if !strings.HasPrefix(line, "cpu ") {
			continue
		}
		fields := strings.Fields(line)
		if len(fields) < 8 {
			return CPUSample{}, fmt.Errorf("short cpu line: %q", line)
		}
		agg = true
		user, _ = strconv.ParseUint(fields[1], 10, 64)
		nice, _ = strconv.ParseUint(fields[2], 10, 64)
		system, _ = strconv.ParseUint(fields[3], 10, 64)
		idle, _ = strconv.ParseUint(fields[4], 10, 64)
		iowait, _ = strconv.ParseUint(fields[5], 10, 64)
		steal, _ = strconv.ParseUint(fields[7], 10, 64)
		break
	}
	if err := scanner.Err(); err != nil {
		return CPUSample{}, err
	}
	if !agg {
		return CPUSample{}, fmt.Errorf("no aggregate cpu line in %s", path)
	}
	total := user + nice + system + idle + iowait + steal
	if total == 0 {
		return CPUSample{}, nil
	}
	return CPUSample{
		UserPct:   pct(user, total),
		NicePct:   pct(nice, total),
		SystemPct: pct(system, total),
		IdlePct:   pct(idle, total),
		IOWaitPct: pct(iowait, total),
		StealPct:  pct(steal, total),
		TotalPct:  pct(total-(idle+iowait), total),
	}, nil
}

func pct(part, total uint64) float64 {
	if total == 0 {
		return 0
	}
	return float64(part) * 100 / float64(total)
}

// readMemInfo parses /proc/meminfo. Units are kB; we convert to
// bytes on the way out so the page never has to multiply.
func readMemInfo(path string) (MemorySample, error) {
	f, err := os.Open(path)
	if err != nil {
		return MemorySample{}, err
	}
	defer f.Close() //nolint:errcheck
	values := map[string]uint64{}
	scanner := bufioNewScanner(f)
	for scanner.Scan() {
		line := scanner.Text()
		fields := strings.Fields(line)
		if len(fields) < 2 {
			continue
		}
		key := strings.TrimSuffix(fields[0], ":")
		v, err := strconv.ParseUint(fields[1], 10, 64)
		if err != nil {
			continue
		}
		values[key] = v * 1024
	}
	if err := scanner.Err(); err != nil {
		return MemorySample{}, err
	}
	mem := MemorySample{
		Total:     values["MemTotal"],
		Available: values["MemAvailable"],
		Free:      values["MemFree"],
		Buffers:   values["Buffers"],
		Cached:    values["Cached"],
		SwapTotal: values["SwapTotal"],
		SwapFree:  values["SwapFree"],
	}
	if mem.Total > 0 {
		used := mem.Total - mem.Available
		mem.UsedPct = pct(used, mem.Total)
		mem.AvailablePct = pct(mem.Available, mem.Total)
	}
	return mem, nil
}

func readLoadAvg(path string) (LoadSample, error) {
	data, err := os.ReadFile(path)
	if err != nil {
		return LoadSample{}, err
	}
	fields := strings.Fields(string(data))
	if len(fields) < 3 {
		return LoadSample{}, fmt.Errorf("short loadavg line: %q", data)
	}
	one, err := strconv.ParseFloat(fields[0], 64)
	if err != nil {
		return LoadSample{}, err
	}
	five, err := strconv.ParseFloat(fields[1], 64)
	if err != nil {
		return LoadSample{}, err
	}
	fifteen, err := strconv.ParseFloat(fields[2], 64)
	if err != nil {
		return LoadSample{}, err
	}
	return LoadSample{One: one, Five: five, Fifteen: fifteen}, nil
}

func readUptime(path string) (time.Duration, error) {
	data, err := os.ReadFile(path)
	if err != nil {
		return 0, err
	}
	fields := strings.Fields(string(data))
	if len(fields) < 1 {
		return 0, fmt.Errorf("short uptime line: %q", data)
	}
	secs, err := strconv.ParseFloat(fields[0], 64)
	if err != nil {
		return 0, err
	}
	return time.Duration(secs * float64(time.Second)), nil
}

func readNetDev(path string) ([]NetDev, error) {
	f, err := os.Open(path)
	if err != nil {
		return nil, err
	}
	defer f.Close() //nolint:errcheck
	var out []NetDev
	scanner := bufioNewScanner(f)
	first := true
	for scanner.Scan() {
		line := scanner.Text()
		if first {
			first = false
			if strings.Contains(line, "Inter-|") {
				continue
			}
		}
		fields := strings.Fields(line)
		if len(fields) < 17 {
			continue
		}
		name := strings.TrimSuffix(fields[0], ":")
		out = append(out, NetDev{
			Name:    name,
			RxBytes: parseU64(fields[1]),
			TxBytes: parseU64(fields[9]),
			RxPkt:   parseU64(fields[2]),
			TxPkt:   parseU64(fields[10]),
			RxErr:   parseU64(fields[3]),
			TxErr:   parseU64(fields[11]),
			RxDrop:  parseU64(fields[4]),
			TxDrop:  parseU64(fields[12]),
		})
	}
	return out, scanner.Err()
}

func parseU64(s string) uint64 {
	v, err := strconv.ParseUint(s, 10, 64)
	if err != nil {
		return 0
	}
	return v
}

func readMounts(path string) ([]DiskSample, error) {
	data, err := os.ReadFile(path)
	if err != nil {
		return nil, err
	}
	var out []DiskSample
	for _, line := range strings.Split(string(data), "\n") {
		line = strings.TrimSpace(line)
		if line == "" || strings.HasPrefix(line, "#") {
			continue
		}
		fields := strings.Fields(line)
		if len(fields) < 3 {
			continue
		}
		// device mountpoint fstype options dump pass
		out = append(out, DiskSample{
			Device: fields[0],
			Mount:  fields[1],
			FSType: fields[2],
		})
	}
	return out, nil
}

// statDisk is implemented in metrics_linux.go (real statfs) and
// metrics_other.go (no-op stub for non-Linux dev builds). The
// function is called from CollectSnapshot below to fill disk usage
// data; tests that exercise the metric paths with fake /proc trees
// accept the zero-value stats as expected.