v0.342.0 (unreleased): the upload and CPU watch — an app that sends or burns far more than its own numbers explain is one operator-only alarm; the line learnt from the off-site push; qBittorrent's default upload limit (never over the household's)

internal/upwatch: exact per-container counters from the Docker Engine API, per-app judging (upload >= max(5 Mbit/s, 25% of the line) for 15 min; CPU >= max(150%, 50% x cores) for 30 min), explainers qBittorrent / Transmission / Plex, app_unexplained_upload / app_unexplained_cpu + _cleared (hub 0.169.0). The off-site push is sampled once a second for the line (a lower bound). qBittorrent limit = 50% of the line, the household's limit always wins. 20 red-proofs. Spike + design: felhom.eu audits/SPIKE-upload-watch-2026-10-23.md, architecture/08 section 9. Not released.
2026-10-23 21:26:09 +02:00
parent 277e99839d
commit a9c1997fbc
35 changed files with 2438 additions and 0 deletions
@@ -0,0 +1,423 @@
+package upwatch
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"runtime"
+	"sort"
+	"time"
+
+	"gitea.dooplex.hu/admin/felhom-controller/internal/logx"
+)
+
+// Rules are the watch's numbers. Every default comes from the spike (`audits/SPIKE-upload-watch-2026-10-23.md`); the
+// field comment says which measurement.
+type Rules struct {
+	// UploadFloorMbps / UploadLineShare: an app is "uploading a lot" at max(floor, share × the measured line). The floor
+	// stands alone while the line is unknown. 5 Mbit/s: on three boxes over 19–31 days no app's own p95 reached 6 Mbit/s
+	// and the longest run at ≥ 5 Mbit/s was 3 minutes; the DooPlex flood averaged ~44 Mbit/s for a day.
+	UploadFloorMbps float64
+	UploadLineShare float64
+	// UploadSustainMin: how long. 15 = five times the longest normal run at ≥ 5 Mbit/s (3 min, qBittorrent on Tester 1).
+	UploadSustainMin int
+	// ExplainedRatio / ExplainedSlackMbps: the app explains its bytes when container_bytes ≤ ratio × app_bytes + slack.
+	// Measured overhead: qBittorrent 3–7 %, Transmission 0.1–2 %; 25 % + 1 Mbit/s also covers DHT, trackers and the web UI.
+	ExplainedRatio     float64
+	ExplainedSlackMbps float64
+	// NoExplainerFactor / NoExplainerSustainMin: an app the box cannot ask alarms only when it sends ≥ factor × its own
+	// p95 (and ≥ the trigger) for this long — worded „no app number to compare".
+	NoExplainerFactor     float64
+	NoExplainerSustainMin int
+	// CPU: an app burns CPU at max(CPUFloorPct, CPUCoreShare × cores × 100) percent-of-one-core for CPUSustainMin with
+	// no explaining work. Measured: the longest normal run at ≥ 150 % was 7 minutes (Immich machine learning) and at
+	// ≥ 80 % 13 minutes; DooPlex's hacked Plex used ~10 cores.
+	CPUFloorPct           float64
+	CPUCoreShare          float64
+	CPUSustainMin         int
+	NoExplainerCPUFloor   float64
+	NoExplainerCPUShare   float64
+	NoExplainerCPUSustain int
+	// ClearAfterMin: an alarm clears after this many minutes in a row below the trigger (or explained).
+	ClearAfterMin int
+	// Cores: the guest's CPU count (0 → runtime.NumCPU, which honours the LXC's cpuset).
+	Cores int
+}
+
+// DefaultRules are the spike's numbers.
+func DefaultRules() Rules {
+	return Rules{
+		UploadFloorMbps: 5, UploadLineShare: 0.25, UploadSustainMin: 15,
+		ExplainedRatio: 1.25, ExplainedSlackMbps: 1,
+		NoExplainerFactor: 10, NoExplainerSustainMin: 60,
+		CPUFloorPct: 150, CPUCoreShare: 0.5, CPUSustainMin: 30,
+		NoExplainerCPUFloor: 200, NoExplainerCPUShare: 0.75, NoExplainerCPUSustain: 60,
+		ClearAfterMin: 15,
+	}
+}
+
+// Infrastructure stacks are never judged: the controller's own off-site push is the line measure (line.go), and
+// traefik/cloudflared relay the APPS' traffic (the app's own container sent it first, and is judged there).
+var infrastructure = map[string]bool{"felhom-controller": true, "traefik": true, "cloudflared": true, "felhom-samba": true, "samba": true}
+
+// Details is the event's payload (operator-only; the hub mails the operator a line built from it).
+type Details struct {
+	App        string  `json:"app"`
+	StackName  string  `json:"stack_name"`
+	Kind       string  `json:"kind"`                  // "upload" | "cpu"
+	Minutes    int     `json:"minutes"`               // how long the condition held when judged
+	RateMbps   float64 `json:"rate_mbps,omitempty"`   // the app's containers, average over the window
+	AppMbps    float64 `json:"app_mbps"`              // what the app itself counted; -1 = it has no number
+	CPUPercent float64 `json:"cpu_percent,omitempty"` // percent of ONE core, average over the window
+	Explainer  string  `json:"explainer"`             // what was asked: "qbittorrent", "transmission", "plex", "none"
+	Viewers    int     `json:"viewers"`
+	Transcodes int     `json:"transcodes"`
+	LineMbps   float64 `json:"line_mbps,omitempty"` // the measured line, when known
+	Cleared    bool    `json:"cleared,omitempty"`
+}
+
+// Watcher holds the per-app state. One goroutine (Run) drives it; Tick is the testable step.
+type Watcher struct {
+	Source     Source
+	Explainers map[string]Explainer // by stack
+	Rules      Rules
+	// Line returns the measured upload line in Mbit/s (line.go); ok=false while unknown.
+	Line func() (float64, bool)
+	// Baseline returns the app's own p95 upload in Mbit/s over the stored history (its containers summed per minute);
+	// ok=false until ≥ 24 h exist.
+	Baseline func(stack string, containers []string) (float64, bool)
+	// Busy reports work the BOX is doing on the app (backup, update, restore) — that minute is never judged.
+	Busy func(stack string) bool
+	Emit func(eventType string, severity string, d Details)
+	Log  *log.Logger
+	Now  func() time.Time
+
+	prevAt  time.Time
+	prev    map[string]Sample // by container
+	apps    map[string]*appState
+	asked   map[string]Report // last report per stack (for counter deltas)
+	askedOK map[string]bool
+}
+
+type appState struct {
+	upRun, upClear       int
+	winTx, winApp        uint64
+	winSec               float64
+	winViewers, winTrans int
+	appCounterKnown      bool
+	upAlarm              bool
+
+	cpuRun, cpuClear int
+	cpuSum           float64
+	cpuAlarm         bool
+
+	lastRate float64 // the last judged minute's Mbit/s (tests read it)
+}
+
+func (w *Watcher) now() time.Time {
+	if w.Now != nil {
+		return w.Now()
+	}
+	return time.Now()
+}
+
+func (w *Watcher) cores() int {
+	if w.Rules.Cores > 0 {
+		return w.Rules.Cores
+	}
+	return runtime.NumCPU()
+}
+
+// UploadTrigger is the Mbit/s at which an app is judged.
+func (w *Watcher) UploadTrigger() float64 {
+	t := w.Rules.UploadFloorMbps
+	if w.Line != nil {
+		if l, ok := w.Line(); ok && l*w.Rules.UploadLineShare > t {
+			t = l * w.Rules.UploadLineShare
+		}
+	}
+	return t
+}
+
+func maxf(a, b float64) float64 {
+	if a > b {
+		return a
+	}
+	return b
+}
+
+// Run ticks once a minute until ctx ends.
+func (w *Watcher) Run(ctx context.Context, every time.Duration) {
+	t := time.NewTicker(every)
+	defer t.Stop()
+	w.Tick(ctx)
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-t.C:
+			w.Tick(ctx)
+		}
+	}
+}
+
+// Tick samples every container, adds the deltas up per app, and judges each app.
+func (w *Watcher) Tick(ctx context.Context) {
+	now := w.now()
+	samples, err := w.Source.Samples(ctx)
+	if err != nil {
+		logx.Warnf(w.Log, "[upwatch] reading container counters: %v — this minute is skipped", err)
+		return
+	}
+	if w.prev == nil {
+		w.prev, w.apps, w.asked, w.askedOK = map[string]Sample{}, map[string]*appState{}, map[string]Report{}, map[string]bool{}
+	}
+	first := w.prevAt.IsZero()
+	dt := now.Sub(w.prevAt).Seconds()
+	type agg struct {
+		tx, cpu    uint64
+		containers []string
+	}
+	per := map[string]*agg{}
+	cur := map[string]Sample{}
+	for _, s := range samples {
+		cur[s.Container] = s
+		if s.Stack == "" || infrastructure[s.Stack] {
+			continue
+		}
+		a := per[s.Stack]
+		if a == nil {
+			a = &agg{}
+			per[s.Stack] = a
+		}
+		a.containers = append(a.containers, s.Container)
+		p, ok := w.prev[s.Container]
+		if !ok || p.Started != s.Started || s.TxBytes < p.TxBytes || s.CPUNanos < p.CPUNanos {
+			continue // new or restarted container: its counters start now; no delta this minute
+		}
+		a.tx += s.TxBytes - p.TxBytes
+		a.cpu += s.CPUNanos - p.CPUNanos
+	}
+	w.prev, w.prevAt = cur, now
+	if first || dt <= 0 || dt > 600 {
+		// The first tick only sets the baseline; a gap over 10 minutes (a suspended guest, a stalled tick) is not a
+		// minute anyone can judge. The explainers are asked so their counters have a baseline too.
+		for stack := range per {
+			w.ask(ctx, stack)
+		}
+		return
+	}
+	stacks := make([]string, 0, len(per))
+	for s := range per {
+		stacks = append(stacks, s)
+	}
+	sort.Strings(stacks)
+	for _, stack := range stacks {
+		a := per[stack]
+		w.judge(ctx, stack, a.containers, a.tx, a.cpu, dt)
+	}
+	for stack, st := range w.apps { // an app that vanished (stopped, removed) clears what it had raised
+		if _, ok := per[stack]; !ok && (st.upAlarm || st.cpuAlarm) {
+			w.clear(stack, st, "upload", st.upAlarm, 0, -1)
+			w.clear(stack, st, "cpu", st.cpuAlarm, 0, -1)
+			delete(w.apps, stack)
+		}
+	}
+}
+
+// ask returns the app's delta of its own upload counter since the last ask (ok=false when it has none or when this
+// is the first reading or the app restarted), plus the report.
+func (w *Watcher) ask(ctx context.Context, stack string) (rep Report, upDelta uint64, deltaOK bool, has bool) {
+	ex, ok := w.Explainers[stack]
+	if !ok {
+		return Report{}, 0, false, false
+	}
+	r, err := ex.Ask(ctx)
+	if err != nil {
+		logx.Debugf(w.Log, "[upwatch] %s: the app did not answer (%v) — treated as no number this minute", stack, err)
+		delete(w.askedOK, stack)
+		return Report{}, 0, false, true
+	}
+	prev, had := w.asked[stack], w.askedOK[stack]
+	w.asked[stack], w.askedOK[stack] = r, true
+	if r.HasUploadCounter && had && prev.HasUploadCounter && r.UploadCounter >= prev.UploadCounter {
+		return r, r.UploadCounter - prev.UploadCounter, true, true
+	}
+	return r, 0, false, true
+}
+
+func (w *Watcher) judge(ctx context.Context, stack string, containers []string, tx, cpu uint64, dt float64) {
+	st := w.apps[stack]
+	if st == nil {
+		st = &appState{}
+		w.apps[stack] = st
+	}
+	rate := float64(tx) * 8 / dt / 1e6
+	cpuPct := float64(cpu) / (dt * 1e9) * 100
+	st.lastRate = rate
+	rep, upDelta, deltaOK, hasEx := w.ask(ctx, stack)
+	busy := w.Busy != nil && w.Busy(stack)
+
+	// ── upload ──
+	trig := w.UploadTrigger()
+	sustain := w.Rules.UploadSustainMin
+	if !hasEx {
+		p95, ok := baseline(w.Baseline, stack, containers)
+		if !ok {
+			trig = -1 // learning: an app the box cannot ask is not judged before it has a history
+		} else {
+			trig = maxf(trig, w.Rules.NoExplainerFactor*p95)
+			sustain = w.Rules.NoExplainerSustainMin
+		}
+	}
+	over := trig > 0 && rate >= trig && !busy
+	slack := w.Rules.ExplainedSlackMbps * 1e6 / 8 * dt
+	minuteExplained := (deltaOK && float64(tx) <= w.Rules.ExplainedRatio*float64(upDelta)+slack) || (rep.HasViewerCount && rep.Viewers > 0)
+	if !st.upAlarm {
+		if over {
+			st.upRun++
+			st.winTx += tx
+			st.winSec += dt
+			if deltaOK {
+				st.winApp += upDelta
+				st.appCounterKnown = true
+			}
+			if rep.Viewers > st.winViewers {
+				st.winViewers = rep.Viewers
+			}
+			if st.upRun >= sustain {
+				d := Details{App: stack, StackName: stack, Kind: "upload", Minutes: st.upRun, RateMbps: round2(float64(st.winTx) * 8 / st.winSec / 1e6),
+					AppMbps: -1, Explainer: explainerName(stack, hasEx), Viewers: st.winViewers}
+				if l, ok := lineOf(w.Line); ok {
+					d.LineMbps = round2(l)
+				}
+				explained := false
+				switch {
+				case st.appCounterKnown:
+					d.AppMbps = round2(float64(st.winApp) * 8 / st.winSec / 1e6)
+					explained = float64(st.winTx) <= w.Rules.ExplainedRatio*float64(st.winApp)+w.Rules.ExplainedSlackMbps*1e6/8*st.winSec
+				case st.winViewers > 0:
+					explained = true
+				}
+				if explained {
+					logx.Infof(w.Log, "[upwatch] %s sent %.1f Mbit/s for %d min — explained by the app (%s: %.1f Mbit/s, %d viewer(s)); normal use",
+						stack, d.RateMbps, d.Minutes, d.Explainer, d.AppMbps, d.Viewers)
+				} else {
+					st.upAlarm, st.upClear = true, 0
+					logx.Warnf(w.Log, "[upwatch] %s sent %.1f Mbit/s for %d min — NOT explained by the app (%s: %s, %d viewer(s)) — operator alarm",
+						stack, d.RateMbps, d.Minutes, d.Explainer, appNumber(d.AppMbps), d.Viewers)
+					w.emit("app_unexplained_upload", "warning", d)
+				}
+				st.upRun, st.winTx, st.winApp, st.winSec, st.winViewers, st.appCounterKnown = 0, 0, 0, 0, 0, false
+			}
+		} else {
+			st.upRun, st.winTx, st.winApp, st.winSec, st.winViewers, st.appCounterKnown = 0, 0, 0, 0, 0, false
+		}
+	} else {
+		if !over || minuteExplained {
+			st.upClear++
+		} else {
+			st.upClear = 0
+		}
+		if st.upClear >= w.Rules.ClearAfterMin {
+			w.clear(stack, st, "upload", true, rate, -1)
+		}
+	}
+
+	// ── CPU ──
+	cores := float64(w.cores())
+	ctrig, csus := maxf(w.Rules.CPUFloorPct, w.Rules.CPUCoreShare*cores*100), w.Rules.CPUSustainMin
+	if !(rep.HasTranscodeCount) {
+		ctrig, csus = maxf(w.Rules.NoExplainerCPUFloor, w.Rules.NoExplainerCPUShare*cores*100), w.Rules.NoExplainerCPUSustain
+	}
+	cpuExplained := busy || (rep.HasTranscodeCount && rep.Transcodes > 0)
+	cpuOver := cpuPct >= ctrig && !cpuExplained
+	if !st.cpuAlarm {
+		if cpuOver {
+			st.cpuRun++
+			st.cpuSum += cpuPct
+			if st.cpuRun >= csus {
+				d := Details{App: stack, StackName: stack, Kind: "cpu", Minutes: st.cpuRun, CPUPercent: round2(st.cpuSum / float64(st.cpuRun)),
+					AppMbps: -1, Explainer: cpuExplainerName(stack, rep.HasTranscodeCount), Transcodes: rep.Transcodes}
+				st.cpuAlarm, st.cpuClear, st.cpuRun, st.cpuSum = true, 0, 0, 0
+				logx.Warnf(w.Log, "[upwatch] %s used %.0f%% CPU (of one core; %d cores) for %d min with no work that explains it (%s) — operator alarm",
+					stack, d.CPUPercent, int(cores), d.Minutes, d.Explainer)
+				w.emit("app_unexplained_cpu", "warning", d)
+			}
+		} else {
+			st.cpuRun, st.cpuSum = 0, 0
+		}
+	} else {
+		if !cpuOver {
+			st.cpuClear++
+		} else {
+			st.cpuClear = 0
+		}
+		if st.cpuClear >= w.Rules.ClearAfterMin {
+			w.clear(stack, st, "cpu", true, 0, cpuPct)
+		}
+	}
+}
+
+func (w *Watcher) clear(stack string, st *appState, kind string, raised bool, rate, cpuPct float64) {
+	if !raised {
+		return
+	}
+	d := Details{App: stack, StackName: stack, Kind: kind, Minutes: w.Rules.ClearAfterMin, AppMbps: -1, Cleared: true}
+	if kind == "upload" {
+		st.upAlarm, st.upClear, st.upRun = false, 0, 0
+		d.RateMbps = round2(rate)
+		logx.Infof(w.Log, "[upwatch] %s: the unexplained upload stopped (%d min below the trigger or explained) — cleared", stack, d.Minutes)
+		w.emit("app_unexplained_upload_cleared", "info", d)
+		return
+	}
+	st.cpuAlarm, st.cpuClear, st.cpuRun, st.cpuSum = false, 0, 0, 0
+	if cpuPct >= 0 {
+		d.CPUPercent = round2(cpuPct)
+	}
+	logx.Infof(w.Log, "[upwatch] %s: the unexplained CPU stopped (%d min below the trigger or explained) — cleared", stack, d.Minutes)
+	w.emit("app_unexplained_cpu_cleared", "info", d)
+}
+
+func (w *Watcher) emit(t, sev string, d Details) {
+	if w.Emit != nil {
+		w.Emit(t, sev, d)
+	}
+}
+
+func baseline(fn func(string, []string) (float64, bool), stack string, containers []string) (float64, bool) {
+	if fn == nil {
+		return 0, false
+	}
+	return fn(stack, containers)
+}
+
+func lineOf(fn func() (float64, bool)) (float64, bool) {
+	if fn == nil {
+		return 0, false
+	}
+	return fn()
+}
+
+func explainerName(stack string, has bool) string {
+	if has {
+		return stack
+	}
+	return "none"
+}
+
+func cpuExplainerName(stack string, has bool) string {
+	if has {
+		return stack + " transcodes"
+	}
+	return "none"
+}
+
+func appNumber(v float64) string {
+	if v < 0 {
+		return "no app number to compare"
+	}
+	return fmt.Sprintf("%.1f Mbit/s", v)
+}
+
+func round2(v float64) float64 { return float64(int64(v*100+0.5)) / 100 }