hamlab/internal/api/status_sensors.go

package api

import (
	"context"
	"encoding/json"
	"os"
	"os/exec"
	"strconv"
	"strings"
	"sync"
	"time"

	"hamlab/internal/priv"
	"hamlab/internal/sensors"
	"hamlab/internal/storage"
)

// statusSensors is the sensors.Provider for what isn't a /proc number: failed services,
// pending updates, pool health, drive health. Each source costs an exec or several, so
// they are read in the background on their own clocks and served from memory; every
// sensor's t says how old its reading really is.
type statusSensors struct {
	s  *Server
	mu sync.RWMutex
	by map[string][]sensors.Sensor // source → its sensors

	cpuPrev map[string]cpuMark // container ID → CPU time at the previous read

	hist map[string]*sensors.Series // numeric sensors only, statusHistory points each
	acc  map[string]*bucket         // the point being built for each, when the window is wider than the read rate
}

// statusHistory is how many readings each status sensor keeps: at 15 s that is over an
// hour of container memory; at 10 min, two days of drive wear.
const statusHistory = 300

// A point being averaged, the same idea as the sampler's: these sources are read every 15 to
// 300 seconds, so a wide window keeps one point per stride rather than one per read.
type bucket struct {
	last int64 // when this sensor's last point was kept
	sum  float64
	n    float64
}

const (
	servicesEvery   = 30 * time.Second
	containersEvery = 15 * time.Second
	updatesEvery    = 5 * time.Minute
	zfsEvery        = time.Minute
	smartEvery      = 10 * time.Minute // storage.Manager caches SMART that long anyway, and never wakes a sleeping disk
)

func newStatusSensors(s *Server) *statusSensors {
	p := &statusSensors{s: s, by: map[string][]sensors.Sensor{}, hist: map[string]*sensors.Series{}, acc: map[string]*bucket{}}
	go p.loop("services", servicesEvery, p.services)
	go p.loop("containers", containersEvery, p.containers)
	go p.loop("updates", updatesEvery, p.updates)
	go p.loop("zfs", zfsEvery, p.zfs)
	go p.loop("smart", smartEvery, p.smart)
	return p
}

func (p *statusSensors) loop(name string, every time.Duration, read func(context.Context, int64) []sensors.Sensor) {
	for {
		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
		got := read(ctx, time.Now().UnixMilli())
		cancel()
		p.mu.Lock()
		p.by[name] = got
		window := p.s.metrics.Window()
		stride := window.Milliseconds() / statusHistory
		live := map[string]bool{}
		for _, x := range got {
			v, ok := x.Value.(float64)
			if !ok {
				continue
			}
			live[x.ID] = true
			h := p.hist[x.ID]
			if h == nil {
				h = &sensors.Series{ID: x.ID, Unit: x.Unit}
				p.hist[x.ID] = h
			}
			b := p.acc[x.ID]
			if b == nil {
				b = &bucket{}
				p.acc[x.ID] = b
			}
			b.sum, b.n = b.sum+v, b.n+1
			// due a stride after the last point kept, with half a read of slack for a late timer
			if b.last != 0 && x.T-b.last < stride-every.Milliseconds()/2 {
				continue // still filling this point
			}
			h.T, h.V = append(h.T, x.T), append(h.V, b.sum/b.n)
			*b = bucket{last: x.T}
			if len(h.T) > statusHistory {
				h.T, h.V = h.T[len(h.T)-statusHistory:], h.V[len(h.V)-statusHistory:]
			}
			if cut := x.T - window.Milliseconds(); h.T[0] < cut { // and nothing older than the window
				i := 0
				for i < len(h.T) && h.T[i] < cut {
					i++
				}
				h.T, h.V = h.T[i:], h.V[i:]
			}
		}
		for id := range p.hist { // a container that is gone takes its history with it
			if strings.HasPrefix(id, prefixOf[name]) && !live[id] && prefixOf[name] != "" {
				delete(p.hist, id)
				delete(p.acc, id)
			}
		}
		p.mu.Unlock()
		if len(got) == 0 { // no pools, no Docker, no SMART: look again, but rarely
			time.Sleep(10 * every)
			continue
		}
		time.Sleep(every)
	}
}

// prefixOf says which IDs a source owns, so its loop forgets only its own departed sensors.
var prefixOf = map[string]string{"containers": "container.", "smart": "smart.", "zfs": "zfs."}

// SensorHistories implements sensors.Historian.
func (p *statusSensors) SensorHistories(match func(id string) bool) []sensors.Series {
	p.mu.RLock()
	defer p.mu.RUnlock()
	var out []sensors.Series
	for id, h := range p.hist {
		if match(id) {
			out = append(out, sensors.Series{ID: h.ID, Unit: h.Unit, T: append([]int64{}, h.T...), V: append([]float64{}, h.V...)})
		}
	}
	return out
}

func (p *statusSensors) Sensors() []sensors.Sensor {
	p.mu.RLock()
	defer p.mu.RUnlock()
	var out []sensors.Sensor
	for _, v := range p.by {
		out = append(out, v...)
	}
	return out
}

func (p *statusSensors) services(ctx context.Context, t int64) []sensors.Sensor {
	units, err := p.s.svc.List(ctx)
	if err != nil {
		return nil
	}
	var failed, running float64
	var names []string
	var per []sensors.Sensor
	for _, u := range units {
		if u.Failed {
			failed++
			names = append(names, u.Name)
		}
		if u.Sub == "running" {
			running++
		}
		// One pair per service that is meant to be doing something: enabled, up, or broken.
		// (Every static and one-shot unit on the box would triple the catalog for nothing.)
		short, isService := strings.CutSuffix(u.Name, ".service")
		if !isService || !(u.Enabled == "enabled" || u.Active == "active" || u.Failed) {
			continue
		}
		id, l := "service."+sensors.Seg(short), map[string]string{"unit": u.Name, "description": u.Description}
		per = append(per,
			sensors.Sensor{ID: id + ".active", Name: short, Kind: "state", Value: u.Active == "active", T: t, Labels: l},
			sensors.Sensor{ID: id + ".state", Name: short + " state", Kind: "state", Value: u.Sub, T: t, Labels: l})
	}
	return append(per, []sensors.Sensor{
		{ID: "services.failed", Name: "Failed services", Kind: "count", Value: failed, T: t, Warn: sensors.F(1)},
		{ID: "services.failed_names", Name: "Failed services", Kind: "state", Value: strings.Join(names, ", "), T: t},
		{ID: "services.running", Name: "Running services", Kind: "count", Value: running, T: t},
	}...)
}

// containers reads Docker (or Podman) with one `ps`, then each running container's memory
// and CPU straight from its cgroup — `docker stats` takes seconds and a daemon round-trip
// per call; two small files per container take microseconds.
func (p *statusSensors) containers(ctx context.Context, t int64) []sensors.Sensor {
	var out []byte
	var err error
	tool := ""
	for _, t := range []string{"docker", "podman"} {
		if _, e := exec.LookPath(t); e != nil {
			continue
		}
		tool = t
		args := []string{tool, "ps", "-a", "--no-trunc", "--format", "{{json .}}"}
		if out, err = exec.CommandContext(ctx, args[0], args[1:]...).Output(); err != nil && priv.Available() {
			full := priv.Wrap(args...) // not in the docker group: ask as root
			out, err = exec.CommandContext(ctx, full[0], full[1:]...).Output()
		}
		break
	}
	if err != nil || len(out) == 0 {
		return nil
	}
	// "Not running" is not "down": servarr stopped by hand in June is off, not broken. What is
	// broken is a container that should be up and isn't — one inspect of the stopped ones tells
	// them apart: Docker refused to start it (State.Error: yesterday's Gitea), it is stuck
	// restarting, or it died with a non-zero code that no restart policy picked up. Under
	// `always` / `unless-stopped` a crash is restarted by Docker, so "exited" there means a
	// person stopped it. With no restart policy, `docker stop` leaves 143 (the process took the
	// SIGTERM), 137 (it didn't, and Docker killed it after the grace period) or 130 (SIGINT):
	// also a person, unless the kernel did the killing for want of memory.
	type stopped struct {
		Name  string
		State struct {
			Status    string
			ExitCode  int
			Error     string
			OOMKilled bool
		}
		HostConfig struct{ RestartPolicy struct{ Name string } }
	}
	down := map[string]bool{}
	var ids []string
	for _, line := range strings.Split(strings.TrimSpace(string(out)), "\n") {
		var c struct{ ID, State string }
		if json.Unmarshal([]byte(line), &c) == nil && c.ID != "" && c.State != "running" {
			ids = append(ids, c.ID)
		}
	}
	if len(ids) > 0 {
		args := append([]string{"inspect"}, ids...)
		insp, ierr := exec.CommandContext(ctx, tool, args...).Output()
		if ierr != nil && priv.Available() {
			full := priv.Wrap(append([]string{tool}, args...)...)
			insp, _ = exec.CommandContext(ctx, full[0], full[1:]...).Output()
		}
		var list []stopped
		if json.Unmarshal(insp, &list) == nil {
			for _, c := range list {
				pol := c.HostConfig.RestartPolicy.Name
				managed := pol == "always" || pol == "unless-stopped"
				code := c.State.ExitCode
				byHand := (code == 130 || code == 137 || code == 143) && !c.State.OOMKilled
				down[strings.TrimPrefix(c.Name, "/")] = c.State.Status == "restarting" || c.State.Status == "dead" || c.State.Error != "" ||
					(!managed && c.State.Status == "exited" && code != 0 && !byHand)
			}
		}
	}
	var res []sensors.Sensor
	var running, unhealthy, total float64
	seenCPU := map[string]cpuMark{}
	for _, line := range strings.Split(strings.TrimSpace(string(out)), "\n") {
		var c struct{ ID, Names, Image, State, Status string }
		if json.Unmarshal([]byte(line), &c) != nil || c.Names == "" {
			continue
		}
		total++
		name := strings.Split(c.Names, ",")[0]
		id, l := "container."+sensors.Seg(name), map[string]string{"container": name, "image": c.Image}
		up := c.State == "running"
		health := "none"
		for _, h := range []string{"unhealthy", "healthy", "starting"} { // "unhealthy" first: it contains "healthy"
			if strings.Contains(c.Status, "("+h+")") || strings.Contains(c.Status, "(health: "+h+")") {
				health = h
				break
			}
		}
		if up {
			running++
		}
		if health == "unhealthy" {
			unhealthy++
		}
		res = append(res,
			sensors.Sensor{ID: id + ".running", Name: name, Kind: "state", Value: up, T: t, Labels: l},
			sensors.Sensor{ID: id + ".down", Name: name + " down", Kind: "state", Value: down[name], T: t, Labels: l},
			sensors.Sensor{ID: id + ".state", Name: name + " state", Kind: "state", Value: c.State, T: t, Labels: l},
			sensors.Sensor{ID: id + ".health", Name: name + " health", Kind: "state", Value: health, T: t, Labels: l})
		if !up {
			continue
		}
		cg := ""
		for _, dir := range []string{"/sys/fs/cgroup/system.slice/docker-" + c.ID + ".scope", "/sys/fs/cgroup/docker/" + c.ID, "/sys/fs/cgroup/machine.slice/libpod-" + c.ID + ".scope"} {
			if _, e := os.Stat(dir + "/memory.current"); e == nil {
				cg = dir
				break
			}
		}
		if cg == "" {
			continue
		}
		if b, e := os.ReadFile(cg + "/memory.current"); e == nil {
			if v, e := strconv.ParseFloat(strings.TrimSpace(string(b)), 64); e == nil {
				res = append(res, sensors.Sensor{ID: id + ".mem", Name: name + " memory", Kind: "bytes", Unit: "B", Value: v, T: t, Labels: l})
			}
		}
		if b, e := os.ReadFile(cg + "/cpu.stat"); e == nil {
			for _, ln := range strings.Split(string(b), "\n") {
				if f := strings.Fields(ln); len(f) == 2 && f[0] == "usage_usec" {
					usec, _ := strconv.ParseFloat(f[1], 64)
					seenCPU[c.ID] = cpuMark{usec, t}
					if prev, ok := p.cpuPrev[c.ID]; ok && t > prev.t && usec >= prev.usec {
						// percent of one core, like the Processes tab
						pctv := (usec - prev.usec) / 1000 / float64(t-prev.t) * 100
						res = append(res, sensors.Sensor{ID: id + ".cpu", Name: name + " CPU", Kind: "usage", Unit: "%", Value: float64(int(pctv*10+0.5)) / 10, T: t, Min: sensors.F(0), Labels: l})
					}
				}
			}
		}
	}
	p.cpuPrev = seenCPU // only this loop touches it
	return append(res,
		sensors.Sensor{ID: "containers.total", Name: "Containers", Kind: "count", Value: total, T: t},
		sensors.Sensor{ID: "containers.running", Name: "Running containers", Kind: "count", Value: running, T: t},
		sensors.Sensor{ID: "containers.unhealthy", Name: "Unhealthy containers", Kind: "count", Value: unhealthy, T: t, Warn: sensors.F(1)})
}

type cpuMark struct {
	usec float64
	t    int64
}

func (p *statusSensors) updates(ctx context.Context, t int64) []sensors.Sensor {
	st := p.s.updates.Status(ctx)
	if st.Manager == "" {
		return nil
	}
	return []sensors.Sensor{
		{ID: "updates.pending", Name: "Pending updates", Kind: "count", Value: float64(len(st.Packages)), T: t, Labels: map[string]string{"manager": st.Manager}},
		{ID: "updates.security", Name: "Pending security updates", Kind: "count", Value: float64(st.SecurityCount), T: t, Warn: sensors.F(1)},
		{ID: "updates.reboot_required", Name: "Reboot required", Kind: "state", Value: st.RebootNeeded, T: t},
	}
}

func (p *statusSensors) zfs(ctx context.Context, t int64) []sensors.Sensor {
	var out []sensors.Sensor
	for _, pool := range p.s.storage.Pools(ctx) {
		id, l := "zfs."+sensors.Seg(pool.Name), map[string]string{"pool": pool.Name}
		out = append(out,
			sensors.Sensor{ID: id + ".health", Name: pool.Name + " health", Kind: "state", Value: pool.Health, T: t, Labels: l},
			sensors.Sensor{ID: id + ".healthy", Name: pool.Name + " healthy", Kind: "state", Value: pool.Health == "ONLINE", T: t, Labels: l},
			sensors.Sensor{ID: id + ".size", Name: pool.Name + " size", Kind: "bytes", Unit: "B", Value: float64(pool.Size), T: t, Labels: l},
			sensors.Sensor{ID: id + ".used", Name: pool.Name + " used", Kind: "bytes", Unit: "B", Value: float64(pool.Alloc), T: t, Min: sensors.F(0), Max: sensors.F(float64(pool.Size)), Labels: l},
			sensors.Sensor{ID: id + ".used_pct", Name: pool.Name + " used", Kind: "usage", Unit: "%", Value: float64(pool.Cap), T: t, Min: sensors.F(0), Max: sensors.F(100), Labels: l},
			sensors.Sensor{ID: id + ".frag", Name: pool.Name + " fragmentation", Kind: "fragmentation", Unit: "%", Value: float64(pool.Frag), T: t, Min: sensors.F(0), Max: sensors.F(100), Labels: l},
		)
	}
	return out
}

func (p *statusSensors) smart(ctx context.Context, t int64) []sensors.Sensor {
	devs, err := storage.Devices(ctx)
	if err != nil {
		return nil
	}
	var out []sensors.Sensor
	for _, d := range devs {
		sm := p.s.storage.Smart(ctx, d.Name, false)
		if sm.Passed == nil || sm.Standby || sm.Error != "" {
			continue // no verdict to report: no smartctl, asleep, or not a SMART device
		}
		id, l := "smart."+sensors.Seg(d.Name), map[string]string{"device": d.Name, "model": d.Model}
		out = append(out,
			sensors.Sensor{ID: id + ".passed", Name: d.Name + " health", Kind: "state", Value: *sm.Passed, T: sm.FetchedAt, Labels: l},
			sensors.Sensor{ID: id + ".warnings", Name: d.Name + " warnings", Kind: "count", Value: float64(len(sm.Warnings)), T: sm.FetchedAt, Warn: sensors.F(1), Labels: l},
		)
		if sm.PowerOnHours > 0 {
			out = append(out, sensors.Sensor{ID: id + ".power_on", Name: d.Name + " powered on", Kind: "duration", Unit: "s", Value: float64(sm.PowerOnHours) * 3600, T: sm.FetchedAt, Labels: l})
		}
		if sm.NVMe != nil {
			out = append(out, sensors.Sensor{ID: id + ".wear", Name: d.Name + " endurance used", Kind: "usage", Unit: "%", Value: float64(sm.NVMe.PercentageUsed), T: sm.FetchedAt, Min: sensors.F(0), Max: sensors.F(100), Labels: l})
		}
	}
	return out
}