From a9c1997fbc9b81fc2d2075713d5219d43bbf7588 Mon Sep 17 00:00:00 2001 From: kisfenyo Date: Fri, 23 Oct 2026 21:26:09 +0200 Subject: [PATCH] =?UTF-8?q?v0.342.0=20(unreleased):=20the=20upload=20and?= =?UTF-8?q?=20CPU=20watch=20=E2=80=94=20an=20app=20that=20sends=20or=20bur?= =?UTF-8?q?ns=20far=20more=20than=20its=20own=20numbers=20explain=20is=20o?= =?UTF-8?q?ne=20operator-only=20alarm;=20the=20line=20learnt=20from=20the?= =?UTF-8?q?=20off-site=20push;=20qBittorrent's=20default=20upload=20limit?= =?UTF-8?q?=20(never=20over=20the=20household's)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- CHANGELOG.md | 40 ++ REUSE.md | 3 + controller/README.md | 12 + controller/cmd/controller/main.go | 11 + controller/cmd/controller/upwatch_wiring.go | 105 +++++ .../cmd/controller/upwatch_wiring_test.go | 52 +++ controller/internal/backup/backup.go | 2 + controller/internal/backup/offbox.go | 14 + controller/internal/i18n/locales/en.json | 10 + controller/internal/i18n/locales/hu.json | 10 + controller/internal/metrics/stack_p95_test.go | 36 ++ controller/internal/metrics/store.go | 55 +++ controller/internal/notify/notifier.go | 28 ++ .../internal/notify/upwatch_events_test.go | 47 ++ controller/internal/settings/settings.go | 27 ++ controller/internal/upwatch/explain.go | 305 +++++++++++++ controller/internal/upwatch/explain_test.go | 140 ++++++ controller/internal/upwatch/line.go | 194 ++++++++ controller/internal/upwatch/line_test.go | 147 ++++++ controller/internal/upwatch/qblimit.go | 77 ++++ controller/internal/upwatch/service.go | 155 +++++++ controller/internal/upwatch/source.go | 127 ++++++ controller/internal/upwatch/watch.go | 423 ++++++++++++++++++ controller/internal/upwatch/watch_test.go | 312 +++++++++++++ controller/internal/web/handlers.go | 14 + controller/internal/web/i18n_cases_a_test.go | 1 + controller/internal/web/server.go | 7 + .../internal/web/templates/app_shell.html | 3 + .../web/templates/settings_system.html | 4 + .../i18n_parity/settings_system_failed.html | 4 + .../i18n_parity/settings_system_full.html | 4 + .../i18n_parity/settings_system_nomemory.html | 4 + .../i18n_parity/settings_system_pending.html | 4 + controller/internal/web/upwatch_lines_test.go | 55 +++ controller/scripts/i18n_go_keys.json | 6 + 35 files changed, 2438 insertions(+) create mode 100644 controller/cmd/controller/upwatch_wiring.go create mode 100644 controller/cmd/controller/upwatch_wiring_test.go create mode 100644 controller/internal/metrics/stack_p95_test.go create mode 100644 controller/internal/notify/upwatch_events_test.go create mode 100644 controller/internal/upwatch/explain.go create mode 100644 controller/internal/upwatch/explain_test.go create mode 100644 controller/internal/upwatch/line.go create mode 100644 controller/internal/upwatch/line_test.go create mode 100644 controller/internal/upwatch/qblimit.go create mode 100644 controller/internal/upwatch/service.go create mode 100644 controller/internal/upwatch/source.go create mode 100644 controller/internal/upwatch/watch.go create mode 100644 controller/internal/upwatch/watch_test.go create mode 100644 controller/internal/web/upwatch_lines_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 33ab9cb..5f6c0f0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,43 @@ +## v0.342.0 (unreleased — committed 2026-10-23 night, NOT released; a test image `0.342.0-rc.1` for scratch 9202 only) — the upload and CPU watch (operator ruling 2026-10-23 19:22) + +**MinAgent: 0.158.1** (unchanged). **Hub dependency:** the four new event types need hub 0.169.0 (committed the same +night, not built); an older hub answers 400 and the event is dropped (logged), nothing else breaks. Task: the weekend- +night brief, Part 2 (spike → design → build, commit only). Design: felhom.eu `architecture/08-alarm-ladder.md` §9 +[DESIGN — proposed]; spike numbers: felhom.eu `audits/SPIKE-upload-watch-2026-10-23.md`. + +- **`internal/upwatch` — the watch.** Once a minute it reads every app container's EXACT sent bytes and CPU time from the + Docker Engine API on the socket (`DockerAPI`; `docker stats`' text has three significant digits — ~1.3 Mbit/s steps at a + one-minute sample), adds them up per app (the compose project) and judges each app: **upload** at ≥ max(5 Mbit/s, 25 % of + the measured line) for 15 minutes; **CPU** at ≥ max(150 %, 50 % × cores) of one core for 30 minutes. Then it ASKS THE APP + (`Explainer`): qBittorrent's `up_info_data`, Transmission's `current-stats.uploadedBytes` (bytes it counted — explained + when the container sent ≤ 1.25 × that + 1 Mbit/s; measured overhead 3–7 % and 0.1–2 %), Plex's `/status/sessions` (any + viewer explains upload; no viewer and a big upload is the Plex2Shell signature) and `/transcode/sessions` (any + transcode explains CPU). Not explained → **`app_unexplained_upload` / `app_unexplained_cpu`** (warning, operator-only on + the hub), ONE per episode, and a `_cleared` (info) after 15 quiet minutes. An app the box cannot ask is judged only once + it has ~20 hours of history, at ≥ 10 × its own p95 for an hour, worded „no app number to compare". The box's own work + on the app (a backup run, an update, a deploy) is never judged. Infrastructure (the controller, traefik, cloudflared, + Samba) is never judged. Passwords and tokens are read at use (the mediawire seams, `09` §3 decision 248), never stored. +- **The upload line, learnt from the off-site push** (`line.go`, `backup.SetOffboxPushObserver`): during the push the + controller samples its OWN sent bytes once a second; the best 10-second rate is kept per night (14 nights, + settings.json `upwatch.line`) and the line is the highest night whose best 10 seconds moved ≥ 10 MB — a lower bound, + shown as „legalább N Mbit/s" on the system settings page („még mérjük" until then). +- **qBittorrent's default upload limit** (`qblimit.go`): 50 % of the measured line (CC's pick; decide row in felhom.eu + STATUS), checked every 15 minutes, set through its Web API, re-set when the line moves > 20 %, and NEVER over a limit the + household set — a limit that was there first, or any change from the value qBittorrent stored for the box (it keeps + KiB/s; the stored value is compared), makes the limit the household's for good. No line yet → nothing is set. While the + box owns the limit, qBittorrent's page says one line: „A qBittorrent feltöltése legfeljebb N Mbit/s, hogy a családi + internet gyors maradjon. A qBittorrentben átállíthatod." +- `metrics.StackTxP95Mbps` (the per-app baseline), `settings.Upwatch` (one opaque block), `notify.NotifyUnexplained`. +- Tests, each red-proved (20 mutations, each seen failing): `TestUpwatch_SeedingQBittorrentIsExplained`, + `…_HackedQBittorrentAlarmsOnceAndClears`, `…_PlexWithNoViewerAlarmsWithAViewerDoesNot`, + `…_NoExplainerLearnsThenNeedsTenTimesItsP95ForAnHour`, `…_CPUBurnerAlarmsTranscodeAndBoxWorkDoNot`, + `…_RestartAndInfrastructureAreNotJudged`, `…_AnAppThatDoesNotAnswerIsNotAnExplanation`, `…_TriggerFollowsTheLine`, + `TestLine_SummarizePeakAndCounts`, `TestQBLimit_TheHouseholdAlwaysWins`, `TestQBittorrent_AskKeepsTheSessionAndReLogsInOn403`, + `TestTransmission_AskFollowsTheSessionID`, `TestPlex_AskReadsSessionsWithTheTokenOnlyWhenClaimed`, + `TestStackTxP95Mbps_SumsContainersPerMinute`, `TestNotifyUnexplained_TypesSentencesAndDetails`, + `TestUpwatch_QBLimitLineOnQBittorrentsPageOnly`, `TestUpwatchIsWiredAtStartup`, `TestNewUpwatch_ExplainersAndEmit`; + the system page's four frozen Hungarian renders gained the one new row (diffed: nothing else moved). + ## v0.341.0 (2026-10-23) — the night update takes a security-fix step (D39 = B, `09` §3 decision 255, R-1239) **Released 2026-10-23: global floor 11:24:47 UTC (the operator's yes in chat), MinAgent 0.158.1; image `sha256:8aa4ce60…c0bc` diff --git a/REUSE.md b/REUSE.md index 82c71fe..89b2b1f 100644 --- a/REUSE.md +++ b/REUSE.md @@ -386,6 +386,8 @@ | `report.SetPendingControllerLog` / `SetControllerLogSource` | controller/internal/report/selftail.go | ACK-armed consume-once self-log pull (the logtail.go shape) | selftail_test.go; source = `logBuffer.Lines`, wired once in main.go | | `util.ParseVersion` / `util.Version.Compare` | controller/internal/util/version.go | THE one semver comparator (house rule: never a second) — selfupdate aliases it; agentapi's MinAgent comparison uses it | rejects pre-release/dev/latest (callers fall back, never trust); numeric compare (0.100 > 0.81) | | `util.ParseVersionPre` | controller/internal/util/version.go | the same comparator, reading `X.Y.Z-rc.N` as BELOW `X.Y.Z` (v0.314.0, R-1053) — ONLY for „is this box below the floor?" | never use it to pick a tag to offer: the button must not see a test build | +| `upwatch.Source` / `upwatch.Explainer` / `Watcher.Now` (v0.342.0, the upload watch) | controller/internal/upwatch/ | `DockerAPI` (Engine API on the socket, exact counters) / the qBittorrent, Transmission and Plex clients | `fakeSource`, `fakeApp` and a minute clock in watch_test.go drive the WHOLE judge path; the app clients are tested against httptest servers that behave as measured on 9202. **Exact counters come from the Engine API, never from `docker stats` text** (three significant digits) | +| `upwatch.QBittorrent.SetUploadLimit` + `ApplyQBLimit` | controller/internal/upwatch/qblimit.go | the one writer of qBittorrent's global upload limit | compare against the value qBittorrent STORED (KiB/s rounding), never the requested one; a limit the household touched is theirs for good | | `mediawire.PlexURL` | controller/internal/mediawire/plexaddr.go | Plex's base URL BY ADDRESS, resolved at each call (v0.314.0, R-1060) — every controller call to Plex | Plex refuses the Host name `plex` as non-local (401); never cache the address | | `stacks.GateCapture` / `stacks.ReadGateAtBackup` | controller/internal/stacks/restore_gate.go | the setup gate a per-app backup records, and the restore's read of it (v0.314.0, R-1057) | the unit `app.yaml` key is `setup_gate:` — the live file's keys, so either reads the same | | `agentapi.AgentVersionReporter` + `featureMinAgent` | controller/internal/agentapi/features.go | version-first Supports (v0.82.0 header channel); probe = fallback for header-less agents | a coupled feature adds BOTH a featureProbes row AND a featureMinAgent row; v0.116.0: `SupportsWithSource` also reports HOW the verdict was reached (version/probe-cache/probe) for the gate log line | @@ -447,4 +449,5 @@ Cross-repo edges: | JSON writers ×5, 3 envelope shapes | api `writeJSON`; web `writeDiskJSON`, `jsonResponse`/`jsonError`, `writeDebugJSON` | | Safe-name validators ×4 | controller/internal/web/validate.go `validStackName`; controller/internal/api/router.go `validStackParam` (same body — api↔web import cycle); controller/internal/backup/offbox.go `isSafeStackName`; controller/internal/appexport/validate.go `ValidateSegment` (strictest) | | DB wait/import ×2 | controller/internal/appbackup/dbdump.go `waitDBReady`/`ImportDump` vs controller/internal/appexport/restore.go `waitForDB`/`importDBDump` | +| qBittorrent Web API login ×2 | controller/internal/mediawire/mediawire.go `qbLogin` (writes the drive wiring) vs controller/internal/upwatch/explain.go `QBittorrent.login` (reads every minute; keeps its session cookie, which mediawire's per-run login does not need) | | compose exec ×2 | controller/internal/stacks/manager.go `composeExecCustomEnv` vs controller/internal/appexport/restore.go `composeExecEnv` (the latter has ctx+timeout; the former has the userdata belt) | diff --git a/controller/README.md b/controller/README.md index 6fe334b..16884a4 100644 --- a/controller/README.md +++ b/controller/README.md @@ -3524,6 +3524,18 @@ no key into the household's account). Code: `internal/stacks/after_unclean_boot. - **Plex:** the claim code is an `after_start` field — asked on the Plex page once Plex runs (`POST /apps/plex/plex-claim`); the gate opens only when Plex says claimed, and „Kész" is refused before that. +### 8e. The upload and CPU watch (v0.342.0, unreleased; felhom.eu `08` §9 [DESIGN — proposed]) + +Once a minute `internal/upwatch` reads every app container's exact sent bytes and CPU time (Docker Engine API), adds them +up per app and, when an app sends a lot or burns CPU for a long time, asks the app itself whether that is its work: +qBittorrent's and Transmission's own upload counters, Plex's playing and transcoding sessions. Explained → normal use. +Not explained → one operator-only event (`app_unexplained_upload` / `app_unexplained_cpu`) and a clear when it stops. +Numbers: upload ≥ max(5 Mbit/s, 25 % of the line) for 15 min; CPU ≥ max(150 %, 50 % × cores) for 30 min; an app with no +explainer only at ≥ 10 × its own p95 for an hour, after ~20 h of history. The upload line is learnt from the nightly +off-site push (one-second samples of the controller's own counter; best 10 s; a lower bound) and shown on the system +settings page. qBittorrent gets a default upload limit of 50 % of that line (never over a household's own limit), and its +page says so in one line. State: settings.json `upwatch`. Spike: felhom.eu `audits/SPIKE-upload-watch-2026-10-23.md`. + ### 8d. Storage that knows its physical disks — the internal place, partitions (v0.317.0 build A, v0.318.0 build B) Design: felhom.eu `architecture/16-storage-physical-identity.md`. The host agent decides which physical disk a place sits on; diff --git a/controller/cmd/controller/main.go b/controller/cmd/controller/main.go index 59eff6c..b1d8456 100644 --- a/controller/cmd/controller/main.go +++ b/controller/cmd/controller/main.go @@ -685,6 +685,15 @@ func main() { // R-550: an interrupted restore is raised once, now that the notifier exists. reportInterruptedRestore(notifier, interruptedRestore) + // The upload and CPU watch (internal/upwatch; felhom.eu `08` §9, operator ruling 2026-10-23). Built here, where the + // notifier first exists, and its push measurement handed to the off-site backup BEFORE any scheduler can start a + // push. Wiring pinned by TestUpwatchIsWiredAtStartup. + upwatchSvc := newUpwatch(sett, stackMgr, backupMgr, metricsStore, notifier, logger) + if backupMgr != nil { + backupMgr.SetOffboxPushObserver(upwatchSvc.PushObserver) + } + startUpwatch(ctx, upwatchSvc, logger) + // R-97a: wire the quiesce loop's hub-event seam. It MUST happen here and not in // startQuiesceLoop, because the notifier is constructed after it — and it must happen at all, // or the seam is inert (the "built but never wired" trap, four recorded instances). Reachability @@ -2044,6 +2053,8 @@ func main() { // URL is derived from it, so two builds of one version can never share a cache entry. Init-only, // before the HTTP server starts. webServer.SetBuildIdentity(GitCommit, BuildTime) + // The upload watch's two page lines: the measured line (settings → system) and qBittorrent's limit (its app page). + webServer.SetUpwatchView(upwatchSvc.View) // R-871 (decision 110): the missed-backup banner reads the night ledger and the box's own on/off record. webServer.SetMissedBackupBanner(webNightLedger, metricsStore) // Migration done-hook: a decommission-initiated migration finalizes the source decommission on diff --git a/controller/cmd/controller/upwatch_wiring.go b/controller/cmd/controller/upwatch_wiring.go new file mode 100644 index 0000000..0435edf --- /dev/null +++ b/controller/cmd/controller/upwatch_wiring.go @@ -0,0 +1,105 @@ +package main + +import ( + "context" + "log" + "sync" + "time" + + "gitea.dooplex.hu/admin/felhom-controller/internal/backup" + "gitea.dooplex.hu/admin/felhom-controller/internal/mediawire" + "gitea.dooplex.hu/admin/felhom-controller/internal/metrics" + "gitea.dooplex.hu/admin/felhom-controller/internal/notify" + "gitea.dooplex.hu/admin/felhom-controller/internal/settings" + "gitea.dooplex.hu/admin/felhom-controller/internal/stacks" + "gitea.dooplex.hu/admin/felhom-controller/internal/upwatch" +) + +// qbLimitShare is qBittorrent's default share of the measured upload line — CC's pick, open for the operator (decide +// row in felhom.eu STATUS): 50 %. +const qbLimitShare = 0.5 + +// baselineMinMinutes: an app the box cannot ask is judged only once its history holds ~20 hours of minutes. +const baselineMinMinutes = 1200 + +// newUpwatch builds the upload and CPU watch (internal/upwatch; felhom.eu `08` §9). It does not start it. +func newUpwatch(sett *settings.Settings, stackMgr *stacks.Manager, backupMgr *backup.Manager, metricsStore *metrics.MetricsStore, + notifier *notify.Notifier, logger *log.Logger) *upwatch.Service { + box := mediawire.DockerBox{} + qb := &upwatch.QBittorrent{Box: box} + svc := &upwatch.Service{Store: sett, QB: qb, QBRunning: func() bool { return box.Running("qbittorrent") }, QBShare: qbLimitShare, Log: logger} + w := &upwatch.Watcher{ + Source: &upwatch.DockerAPI{}, + Explainers: map[string]upwatch.Explainer{ + "qbittorrent": qb, + "transmission": &upwatch.Transmission{Box: box}, + "plex": &upwatch.Plex{Token: func() string { return mediawire.PlexToken(box) }, URL: mediawire.PlexURL}, + }, + Rules: upwatch.DefaultRules(), + Line: svc.Line, + Busy: func(stack string) bool { + if backupMgr != nil && backupMgr.IsRunning() { + return true + } + if st, ok := stackMgr.GetStack(stack); ok && st != nil && (st.Updating || st.Deploying) { + return true + } + return false + }, + Emit: notifier.NotifyUnexplained, + Log: logger, + } + if metricsStore != nil { + w.Baseline = dailyBaseline(metricsStore) + } + svc.Watcher = w + return svc +} + +// dailyBaseline caches each app's p95 for a day (a 7-day query is a few hundred thousand rows). +func dailyBaseline(ms *metrics.MetricsStore) func(string, []string) (float64, bool) { + type entry struct { + at time.Time + p95 float64 + ok bool + } + var mu sync.Mutex + cache := map[string]entry{} + return func(stack string, containers []string) (float64, bool) { + mu.Lock() + defer mu.Unlock() + if e, ok := cache[stack]; ok && time.Since(e.at) < 24*time.Hour { + return e.p95, e.ok + } + p95, minutes, err := ms.StackTxP95Mbps(containers, time.Now().Add(-7*24*time.Hour)) + e := entry{at: time.Now(), p95: p95, ok: err == nil && minutes >= baselineMinMinutes} + cache[stack] = e + return e.p95, e.ok + } +} + +// startUpwatch runs the minute watcher and the 15-minute qBittorrent limit rule until ctx ends. +func startUpwatch(ctx context.Context, svc *upwatch.Service, logger *log.Logger) { + go svc.Watcher.Run(ctx, time.Minute) + go func() { + t := time.NewTicker(15 * time.Minute) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + svc.QBLimitNow(ctx) + } + } + }() + l, ok := svc.Line() + logger.Printf("[INFO] [upwatch] upload and CPU watch started (line %s, qBittorrent share %.0f %%)", lineText(l, ok), qbLimitShare*100) +} + +func lineText(l float64, ok bool) string { + if !ok { + return "unknown — learning from the off-site push" + } + return "at least " + upwatch.MbpsText(int64(l*1e6/8)) + " Mbit/s" +} diff --git a/controller/cmd/controller/upwatch_wiring_test.go b/controller/cmd/controller/upwatch_wiring_test.go new file mode 100644 index 0000000..80f435b --- /dev/null +++ b/controller/cmd/controller/upwatch_wiring_test.go @@ -0,0 +1,52 @@ +package main + +import ( + "io" + "log" + "path/filepath" + "testing" + + "gitea.dooplex.hu/admin/felhom-controller/internal/notify" + "gitea.dooplex.hu/admin/felhom-controller/internal/settings" + "gitea.dooplex.hu/admin/felhom-controller/internal/upwatch" +) + +// The upload watch is built, started, handed to the off-site push BEFORE the scheduler can start one, and its view +// handed to the pages — a seam built and never wired is this project's commonest defect. +// RED-PROOF: delete the SetOffboxPushObserver line → fails; move it below the scheduler's start → fails. +func TestUpwatchIsWiredAtStartup(t *testing.T) { + lines, _, _ := slice4CallLines(t) + for _, fn := range []string{"newUpwatch", "SetOffboxPushObserver", "startUpwatch", "SetUpwatchView"} { + if len(lines[fn]) == 0 { + t.Fatalf("main.go never calls %s", fn) + } + } + if lines["SetOffboxPushObserver"][0] > lines["Daily"][0] { + t.Fatalf("SetOffboxPushObserver (line %d) comes after the first scheduled job (line %d)", lines["SetOffboxPushObserver"][0], lines["Daily"][0]) + } +} + +// newUpwatch asks the three apps the spike proved, reports through the notifier, and reads exact counters. +// RED-PROOF: drop the transmission entry → fails. +func TestNewUpwatch_ExplainersAndEmit(t *testing.T) { + lg := log.New(io.Discard, "", 0) + sett, err := settings.Load(filepath.Join(t.TempDir(), "settings.json"), lg) + if err != nil { + t.Fatal(err) + } + svc := newUpwatch(sett, nil, nil, nil, notify.New("", "", "", sett, lg, false), lg) + for _, a := range []string{"qbittorrent", "transmission", "plex"} { + if _, ok := svc.Watcher.Explainers[a]; !ok { + t.Errorf("no explainer for %s", a) + } + } + if svc.Watcher.Emit == nil || svc.Watcher.Line == nil || svc.QB == nil || svc.QBShare != 0.5 { + t.Fatalf("service not wired: %+v", svc) + } + if _, ok := svc.Watcher.Source.(*upwatch.DockerAPI); !ok { + t.Fatal("the watch must read the Engine API's exact counters, not docker stats' text") + } + if svc.Watcher.Baseline != nil { + t.Fatal("no metrics store, yet a baseline: an app with no explainer would be judged without history") + } +} diff --git a/controller/internal/backup/backup.go b/controller/internal/backup/backup.go index fd7add6..b64b5c6 100644 --- a/controller/internal/backup/backup.go +++ b/controller/internal/backup/backup.go @@ -106,6 +106,8 @@ type Manager struct { // offbox (Part B): the restic-SFTP exec seam (nil → real restic) + the failure→operator-alert hook. offboxRunner offboxRunner offboxNotify func(dur time.Duration, snapshots int, err error) + // offboxPushObserver (the upload watch) measures the push; nil = none. + offboxPushObserver func() func() // offboxStreamRunner + offboxProgress: the MANUAL run's live progress (v0.147.0, 4c). The stream // seam scans restic's `--json` stdout line-by-line; the state is what the page polls. Both are // inert on the nightly path — the sink is installed only for the duration of a manual run. diff --git a/controller/internal/backup/offbox.go b/controller/internal/backup/offbox.go index 57dada5..de34bcb 100644 --- a/controller/internal/backup/offbox.go +++ b/controller/internal/backup/offbox.go @@ -52,6 +52,11 @@ func defaultOffboxRunner(ctx context.Context, env []string, args ...string) ([]b // SetOffboxRunner overrides the restic exec (tests). SetOffboxNotify wires the failure→operator alert. func (m *Manager) SetOffboxRunner(r offboxRunner) { m.offboxRunner = r } + +// SetOffboxPushObserver hands the off-site push a measurement: fn is called as the push starts and returns the +// function that ends it (the upload watch's line sampler). Nil = none. +func (m *Manager) SetOffboxPushObserver(fn func() func()) { m.offboxPushObserver = fn } + func (m *Manager) SetOffboxNotify(fn func(dur time.Duration, snapshots int, err error)) { m.offboxNotify = fn } @@ -1127,7 +1132,16 @@ func (m *Manager) runOffboxBackup(ctx context.Context, withProgress bool) error } else { m.logger.Printf("[INFO] [offbox] pre-push dump leg completed in %s — snapshot pair is coherent", time.Since(dumpStart).Round(time.Millisecond)) } + // The upload watch learns the household's upload line from this push (internal/upwatch/line.go): a + // one-second sample of this container's own sent bytes, from here to the end of the push only. + var endPush func() + if m.offboxPushObserver != nil { + endPush = m.offboxPushObserver() + } runResult, runErr = m.runOffboxInternal(ctx, apps, base, env, t) + if endPush != nil { + endPush() + } backedUp = runResult.backedUp missing = runResult.missing endStamp() diff --git a/controller/internal/i18n/locales/en.json b/controller/internal/i18n/locales/en.json index 941d645..4368d7c 100644 --- a/controller/internal/i18n/locales/en.json +++ b/controller/internal/i18n/locales/en.json @@ -2782,6 +2782,16 @@ "poweroff.refused.other_guests": "Another system runs on this machine, so it cannot be turned off from here.", "poweroff.refused.generic": "The machine cannot turn itself off safely right now. Try again in a few minutes.", "event.app_repaired_after_crash": "%s: after the unexpected stop, %d damaged preview file(s), %d repaired, %d still broken; lost originals: %d", + "event.app_unexplained_upload": "%s: uploaded %.1f Mbit/s for %d minutes, but the app itself counted only %.1f Mbit/s. This can be the sign of a hacked app.", + "event.app_unexplained_upload_noapp": "%s: uploaded %.1f Mbit/s for %d minutes; the app gives no number to compare, and this is far more than usual. This can be the sign of a hacked app.", + "event.app_unexplained_upload_cleared": "%s: the unexplained upload stopped.", + "event.app_unexplained_cpu": "%s: %.0f%% CPU (counted on one core) for %d minutes, with no work that explains it. This can be the sign of a hacked app.", + "event.app_unexplained_cpu_cleared": "%s: the unexplained CPU use stopped.", + "upwatch.qb_limit_note": "qBittorrent uploads at most %s Mbit/s, so the family internet stays fast. You can change this in qBittorrent.", + "upwatch.line_at_least": "at least", + "upwatch.line_measured": "Mbit/s (measured from the nightly off-site backup)", + "upwatch.line_label": "Upload speed", + "upwatch.line_learning": "still being measured (from the nightly off-site backup)", "event.app_repair_after_crash_skipped": "%s: the repair after the unexpected stop was skipped: %s", "app_info.lost_files_title": "Photos lost in an unexpected stop", "app_info.lost_files_text": "The machine stopped unexpectedly (for example in a power cut), and these files were lost. The machine started again at:", diff --git a/controller/internal/i18n/locales/hu.json b/controller/internal/i18n/locales/hu.json index 7531451..828d35b 100644 --- a/controller/internal/i18n/locales/hu.json +++ b/controller/internal/i18n/locales/hu.json @@ -2770,6 +2770,16 @@ "poweroff.refused.other_guests": "Ezen a gépen más rendszer is fut, ezért innen nem lehet kikapcsolni.", "poweroff.refused.generic": "A gép most nem tudja biztonságosan kikapcsolni magát. Próbáld újra néhány perc múlva.", "event.app_repaired_after_crash": "%s: a váratlan leállás után %d sérült előnézeti fájl, ebből %d javítva, %d maradt hibás; elveszett eredeti fájl: %d", + "event.app_unexplained_upload": "%s: %.1f Mbit/s feltöltés %d percen át, de az alkalmazás saját adata szerint csak %.1f Mbit/s. Ez feltört alkalmazás jele lehet.", + "event.app_unexplained_upload_noapp": "%s: %.1f Mbit/s feltöltés %d percen át; az alkalmazástól nincs összevethető adat, és ez sokkal több a szokásosnál. Ez feltört alkalmazás jele lehet.", + "event.app_unexplained_upload_cleared": "%s: a megmagyarázatlan feltöltés megszűnt.", + "event.app_unexplained_cpu": "%s: %.0f%% CPU (egy magra számolva) %d percen át, és nincs munka, ami ezt magyarázná. Ez feltört alkalmazás jele lehet.", + "event.app_unexplained_cpu_cleared": "%s: a megmagyarázatlan CPU-használat megszűnt.", + "upwatch.qb_limit_note": "A qBittorrent feltöltése legfeljebb %s Mbit/s, hogy a családi internet gyors maradjon. A qBittorrentben átállíthatod.", + "upwatch.line_at_least": "legalább", + "upwatch.line_measured": "Mbit/s (az éjszakai távoli mentésből mérve)", + "upwatch.line_label": "Feltöltési sebesség", + "upwatch.line_learning": "még mérjük (az éjszakai távoli mentésből)", "event.app_repair_after_crash_skipped": "%s: a váratlan leállás utáni javítás elmaradt: %s", "app_info.lost_files_title": "Elveszett fotók egy váratlan leállás miatt", "app_info.lost_files_text": "A gép váratlanul leállt (például áramszünet miatt), és ezek a fájlok elvesztek. A gép újraindulásának ideje:", diff --git a/controller/internal/metrics/stack_p95_test.go b/controller/internal/metrics/stack_p95_test.go new file mode 100644 index 0000000..76f66f3 --- /dev/null +++ b/controller/internal/metrics/stack_p95_test.go @@ -0,0 +1,36 @@ +package metrics + +import ( + "io" + "log" + "path/filepath" + "testing" + "time" +) + +// The upload watch's baseline: per-minute sums across the app's containers, p95, only the named containers. +// RED-PROOF: take the p95 per container instead of the per-minute SUM → 2 + 1 = 3 reads as 2 → fails. +func TestStackTxP95Mbps_SumsContainersPerMinute(t *testing.T) { + st, err := NewMetricsStore(filepath.Join(t.TempDir(), "m.db"), log.New(io.Discard, "", 0)) + if err != nil { + t.Fatal(err) + } + defer st.Close() + t0 := time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC).Unix() + var a, b, other int64 + for i := int64(0); i < 200; i++ { + ts := t0 + i*60 + st.InsertContainerMetrics([]ContainerSample{ + {Timestamp: ts, ContainerName: "nextcloud", NetTxBytes: a}, + {Timestamp: ts, ContainerName: "nextcloud-db", NetTxBytes: b}, + {Timestamp: ts, ContainerName: "qbittorrent", NetTxBytes: other}, + }) + a += 2e6 / 8 * 60 // 2 Mbit/s + b += 1e6 / 8 * 60 // 1 Mbit/s + other += 500e6 / 8 * 60 + } + p95, minutes, err := st.StackTxP95Mbps([]string{"nextcloud", "nextcloud-db"}, time.Unix(t0, 0)) + if err != nil || minutes != 199 || p95 < 2.99 || p95 > 3.01 { + t.Fatalf("p95 = %v over %d minutes, err %v — want 3 Mbit/s over 199", p95, minutes, err) + } +} diff --git a/controller/internal/metrics/store.go b/controller/internal/metrics/store.go index 43be00e..f4f72d1 100644 --- a/controller/internal/metrics/store.go +++ b/controller/internal/metrics/store.go @@ -4,6 +4,8 @@ import ( "database/sql" "fmt" "log" + "sort" + "strings" "time" _ "modernc.org/sqlite" @@ -357,3 +359,56 @@ func (s *MetricsStore) HourSampleCounts(from, to time.Time) (map[int64]int, erro } return out, rows.Err() } + +// StackTxP95Mbps is the 95th percentile of an app's upload, its containers summed per minute, over the rows since +// `since` (the upload watch's „far above its own p95" rule for an app the box cannot ask; internal/upwatch). minutes +// is how many one-minute buckets the history holds. The counters are `docker stats`' text values (three significant +// digits), coarse but enough for a p95 the rule multiplies by ten. +func (s *MetricsStore) StackTxP95Mbps(containers []string, since time.Time) (p95 float64, minutes int, err error) { + if len(containers) == 0 { + return 0, 0, nil + } + q := "SELECT container_name, ts, net_tx_bytes FROM container_metrics WHERE ts >= ? AND container_name IN (?" + + strings.Repeat(",?", len(containers)-1) + ") ORDER BY container_name, ts" + args := []interface{}{since.Unix()} + for _, c := range containers { + args = append(args, c) + } + rows, err := s.db.Query(q, args...) + if err != nil { + return 0, 0, err + } + defer rows.Close() + buckets := map[int64]float64{} + var prevName string + var prevTS, prevTx int64 + havePrev := false + for rows.Next() { + var name string + var ts int64 + var tx sql.NullInt64 + if err := rows.Scan(&name, &ts, &tx); err != nil { + return 0, 0, err + } + if !tx.Valid { + havePrev = false + continue + } + if havePrev && name == prevName && ts > prevTS && ts-prevTS <= 180 && tx.Int64 >= prevTx { + buckets[ts/60] += float64(tx.Int64-prevTx) * 8 / float64(ts-prevTS) / 1e6 + } + prevName, prevTS, prevTx, havePrev = name, ts, tx.Int64, true + } + if err := rows.Err(); err != nil { + return 0, 0, err + } + if len(buckets) == 0 { + return 0, 0, nil + } + vals := make([]float64, 0, len(buckets)) + for _, v := range buckets { + vals = append(vals, v) + } + sort.Float64s(vals) + return vals[int(float64(len(vals)-1)*0.95)], len(vals), nil +} diff --git a/controller/internal/notify/notifier.go b/controller/internal/notify/notifier.go index d7a5bc6..10efc0e 100644 --- a/controller/internal/notify/notifier.go +++ b/controller/internal/notify/notifier.go @@ -14,6 +14,7 @@ import ( "time" "gitea.dooplex.hu/admin/felhom-controller/internal/settings" + "gitea.dooplex.hu/admin/felhom-controller/internal/upwatch" "gitea.dooplex.hu/admin/felhom-controller/internal/i18n" ) @@ -882,6 +883,33 @@ func (n *Notifier) NotifyAppRepairedAfterCrash(d AppRepairDetails) { } } +// NotifyUnexplained sends the upload and CPU watch's events (internal/upwatch; felhom.eu `08` §9). OPERATOR-ONLY on the +// hub (notify.operatorOnlyEvents there, registered in the same commit as the types): the household can do nothing +// about a hacked app but wait for support (the R-388 rule). Edge-triggered by the watch: one event when the condition +// is judged, one `_cleared` (info, recorded) when it stops. +func (n *Notifier) NotifyUnexplained(eventType, _ string, d upwatch.Details) { + // The severity is fixed per type (alarm = warning, clear = info) — never the caller's word, so the hub's vocabulary + // contract (TestR329_EveryEmittedSeverityIsInTheHubVocabulary) can read every value from this function. + switch eventType { + case "app_unexplained_upload": + if d.AppMbps >= 0 { + n.pushEventMsg(eventType, "warning", "event.app_unexplained_upload", d, d.App, d.RateMbps, d.Minutes, d.AppMbps) + } else { + n.pushEventMsg(eventType, "warning", "event.app_unexplained_upload_noapp", d, d.App, d.RateMbps, d.Minutes) + } + case "app_unexplained_upload_cleared": + n.pushEventMsg(eventType, "info", "event.app_unexplained_upload_cleared", d, d.App) + case "app_unexplained_cpu": + n.pushEventMsg(eventType, "warning", "event.app_unexplained_cpu", d, d.App, d.CPUPercent, d.Minutes) + case "app_unexplained_cpu_cleared": + n.pushEventMsg(eventType, "info", "event.app_unexplained_cpu_cleared", d, d.App) + default: + if n.logger != nil { + n.logger.Printf("[ERROR] NotifyUnexplained: unknown event type %q — not sent", eventType) + } + } +} + type OOMStormDetails struct { App string `json:"app"` StackName string `json:"stack_name"` diff --git a/controller/internal/notify/upwatch_events_test.go b/controller/internal/notify/upwatch_events_test.go new file mode 100644 index 0000000..3492c2d --- /dev/null +++ b/controller/internal/notify/upwatch_events_test.go @@ -0,0 +1,47 @@ +package notify + +import ( + "strings" + "testing" + + "gitea.dooplex.hu/admin/felhom-controller/internal/upwatch" +) + +// The upload and CPU watch's four events reach the wire with the type, the severity, the numbers in the sentence and +// the details object (the hub mails the operator from it). An unknown type is not sent. +// RED-PROOF: swap the AppMbps branch (use the no-number sentence always) → the first case lacks „csak 0.0" → fails. +func TestNotifyUnexplained_TypesSentencesAndDetails(t *testing.T) { + type got struct { + typ, sev, msg string + d interface{} + } + var gs []got + n := &Notifier{} + n.pushFn = func(typ, sev, msg, _ string, d interface{}) { gs = append(gs, got{typ, sev, msg, d}) } + d := upwatch.Details{App: "qbittorrent", StackName: "qbittorrent", Kind: "upload", Minutes: 15, RateMbps: 10136.77, AppMbps: 0, Explainer: "qbittorrent"} + n.NotifyUnexplained("app_unexplained_upload", "warning", d) + d.AppMbps = -1 + n.NotifyUnexplained("app_unexplained_upload", "warning", d) + n.NotifyUnexplained("app_unexplained_upload_cleared", "info", upwatch.Details{App: "qbittorrent", Cleared: true}) + n.NotifyUnexplained("app_unexplained_cpu", "warning", upwatch.Details{App: "plex", Kind: "cpu", Minutes: 30, CPUPercent: 380}) + n.NotifyUnexplained("app_unexplained_cpu_cleared", "info", upwatch.Details{App: "plex", Cleared: true}) + n.NotifyUnexplained("app_unexplained_typo", "warning", d) + if len(gs) != 5 { + t.Fatalf("pushed %d events, want 5 (the unknown type must not be sent): %+v", len(gs), gs) + } + want := []struct{ typ, sev, frag string }{ + {"app_unexplained_upload", "warning", "qbittorrent: 10136.8 Mbit/s feltöltés 15 percen át, de az alkalmazás saját adata szerint csak 0.0 Mbit/s"}, + {"app_unexplained_upload", "warning", "az alkalmazástól nincs összevethető adat"}, + {"app_unexplained_upload_cleared", "info", "qbittorrent: a megmagyarázatlan feltöltés megszűnt."}, + {"app_unexplained_cpu", "warning", "plex: 380% CPU (egy magra számolva) 30 percen át"}, + {"app_unexplained_cpu_cleared", "info", "plex: a megmagyarázatlan CPU-használat megszűnt."}, + } + for i, w := range want { + if gs[i].typ != w.typ || gs[i].sev != w.sev || !strings.Contains(gs[i].msg, w.frag) { + t.Errorf("event %d = %s/%s %q, want %s/%s containing %q", i, gs[i].typ, gs[i].sev, gs[i].msg, w.typ, w.sev, w.frag) + } + if _, ok := gs[i].d.(upwatch.Details); !ok { + t.Errorf("event %d carries no details object", i) + } + } +} diff --git a/controller/internal/settings/settings.go b/controller/internal/settings/settings.go index 72c4e42..5b77d53 100644 --- a/controller/internal/settings/settings.go +++ b/controller/internal/settings/settings.go @@ -78,6 +78,11 @@ type Settings struct { // AppUpdateEventsSeeded (v0.264.0) guards the ONE-TIME add of app_update_undone / app_update_held // into an existing household's stored prefs (seedAppUpdateEvents). AppUpdateEventsSeeded bool `json:"app_update_events_seeded,omitempty"` + + // Upwatch (the upload and CPU watch, internal/upwatch) keeps its own state here as one opaque JSON block: the + // measured upload line (the last 14 off-site pushes) and qBittorrent's default-limit record (who owns the limit). + // Opaque on purpose — this package does not know the watch's types, and the watch does not reach into settings. + Upwatch json.RawMessage `json:"upwatch,omitempty"` // UnhealthyStopEventSeeded (v0.269.0) guards the ONE-TIME add of app_stopped_unhealthy (the // seedAppUpdateEvents shape). UnhealthyStopEventSeeded bool `json:"unhealthy_stop_event_seeded,omitempty"` @@ -900,6 +905,28 @@ func (s *Settings) migrateResticToRsync() { // Save writes settings to disk atomically (write to .tmp, rename). // Caller must hold the write lock or call this from a method that does. +// GetUpwatch returns a copy of the watch's state block (nil when none was ever written). +func (s *Settings) GetUpwatch() json.RawMessage { + s.mu.RLock() + defer s.mu.RUnlock() + if s.Upwatch == nil { + return nil + } + return append(json.RawMessage(nil), s.Upwatch...) +} + +// UpdateUpwatch replaces the watch's state block through fn (read-modify-write under the lock) and saves. +func (s *Settings) UpdateUpwatch(fn func(old json.RawMessage) (json.RawMessage, error)) error { + s.mu.Lock() + defer s.mu.Unlock() + nv, err := fn(append(json.RawMessage(nil), s.Upwatch...)) + if err != nil { + return err + } + s.Upwatch = nv + return s.save() +} + func (s *Settings) save() error { data, err := json.MarshalIndent(s, "", " ") if err != nil { diff --git a/controller/internal/upwatch/explain.go b/controller/internal/upwatch/explain.go new file mode 100644 index 0000000..f10dd99 --- /dev/null +++ b/controller/internal/upwatch/explain.go @@ -0,0 +1,305 @@ +package upwatch + +import ( + "bytes" + "context" + "encoding/base64" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "regexp" + "strings" + "time" +) + +// Report is what one app says about its own work right now. +type Report struct { + // UploadCounter is the app's own cumulative count of bytes it uploaded (qBittorrent `up_info_data`, Transmission + // `current-stats.uploadedBytes`). It restarts at zero with the app; the watcher treats a drop as a restart. + UploadCounter uint64 + HasUploadCounter bool + // Viewers is the number of playing sessions (Plex `/status/sessions` size). A viewer EXPLAINS upload of any size: + // the per-session bandwidth was not measured in the spike (no claimed server with a remote player tonight), so + // the rule is deliberately the conservative half — no viewer and a big upload is the Plex2Shell signature. + Viewers int + HasViewerCount bool + // Transcodes is CPU work the app reports (Plex `/transcode/sessions` size). Any transcode explains the CPU. + Transcodes int + HasTranscodeCount bool + // Limit is the app's current global upload limit in bytes/s (0 = none); HasLimit when the app has one to read. + Limit int64 + HasLimit bool +} + +// Explainer asks one app about its own work. +type Explainer interface { + Stack() string + Ask(ctx context.Context) (Report, error) +} + +// Box is what an explainer needs outside HTTP — the mediawire seams (`docker exec printenv`, the Plex token read at +// use). Production: mediawire.DockerBox. +type Box interface { + Env(container, name string) (string, error) + Running(container string) bool +} + +// httpDo is one request to an app's own container. Small and local on purpose: mediawire's helpers are methods of +// its Wirer, which also writes; this package only reads (and sets one limit, qblimit.go). +func httpDo(ctx context.Context, c *http.Client, method, u string, hdr map[string]string, body []byte, ctype string) (int, []byte, http.Header, error) { + req, err := http.NewRequestWithContext(ctx, method, u, bytes.NewReader(body)) + if err != nil { + return 0, nil, nil, err + } + for k, v := range hdr { + if strings.EqualFold(k, "Host") { + req.Host = v + continue + } + req.Header.Set(k, v) + } + if ctype != "" { + req.Header.Set("Content-Type", ctype) + } + resp, err := c.Do(req) + if err != nil { + return 0, nil, nil, err + } + defer resp.Body.Close() + b, _ := io.ReadAll(io.LimitReader(resp.Body, 4<<20)) + return resp.StatusCode, b, resp.Header, nil +} + +// ── qBittorrent ────────────────────────────────────────────────────────────────────────────────────────────────── + +// QBittorrent asks qBittorrent's Web API. Measured on 9202 (qBittorrent 5.2.4, WebAPI 2.15.1): while seeding, the +// container sent 1.03–1.07 × what `up_info_data` counted (protocol overhead); `transfer/uploadLimit` and +// `app/preferences.up_limit` are the same value, persisted across a restart (`Session\GlobalUPSpeedLimit`, KiB/s). +type QBittorrent struct { + Box Box + URL string // default http://qbittorrent:8080 (the container's name on the docker network, as mediawire calls it) + HTTP *http.Client + + sess map[string]string // the session cookie, kept between minutes (a login a minute would fill qBittorrent's log) +} + +func (q *QBittorrent) Stack() string { return "qbittorrent" } + +func (q *QBittorrent) base() string { + if q.URL != "" { + return q.URL + } + return "http://qbittorrent:8080" +} + +func (q *QBittorrent) client() *http.Client { + if q.HTTP != nil { + return q.HTTP + } + return &http.Client{Timeout: 20 * time.Second} +} + +var qbCookieRE = regexp.MustCompile(`((?:QBT_)?SID[A-Za-z0-9_]*)=([^;]+)`) + +// login returns the headers of an authenticated session. The password is the catalog's QBT_ADMIN_PASSWORD, read at +// use; Referer must equal the base (qBittorrent's CSRF check — a mismatch is a 401, measured). +func (q *QBittorrent) login(ctx context.Context) (map[string]string, error) { + pw, err := q.Box.Env("qbittorrent", "QBT_ADMIN_PASSWORD") + if err != nil { + return nil, err + } + form := url.Values{"username": {"admin"}, "password": {pw}}.Encode() + code, _, hd, err := httpDo(ctx, q.client(), "POST", q.base()+"/api/v2/auth/login", map[string]string{"Referer": q.base()}, []byte(form), "application/x-www-form-urlencoded") + if err != nil { + return nil, err + } + m := qbCookieRE.FindStringSubmatch(strings.Join(hd.Values("Set-Cookie"), "; ")) + if (code != 200 && code != 204) || m == nil { // 5.x answers a good login 204 + return nil, fmt.Errorf("qBittorrent login: HTTP %d", code) + } + return map[string]string{"Referer": q.base(), "Cookie": m[1] + "=" + m[2]}, nil +} + +// session returns the kept session, logging in when there is none or when fresh is set. +func (q *QBittorrent) session(ctx context.Context, fresh bool) (map[string]string, error) { + if q.sess != nil && !fresh { + return q.sess, nil + } + h, err := q.login(ctx) + if err != nil { + q.sess = nil + return nil, err + } + q.sess = h + return h, nil +} + +func (q *QBittorrent) Ask(ctx context.Context) (Report, error) { + h, err := q.session(ctx, false) + if err != nil { + return Report{}, err + } + code, b, _, err := httpDo(ctx, q.client(), "GET", q.base()+"/api/v2/transfer/info", h, nil, "") + if err == nil && code == 403 { // the session expired (qBittorrent restarted, or its 1-hour timeout): once more + if h, err = q.session(ctx, true); err != nil { + return Report{}, err + } + code, b, _, err = httpDo(ctx, q.client(), "GET", q.base()+"/api/v2/transfer/info", h, nil, "") + } + if err != nil || code != 200 { + return Report{}, fmt.Errorf("qBittorrent transfer/info: HTTP %d %v", code, err) + } + var ti struct { + UpInfoData uint64 `json:"up_info_data"` + } + if err := json.Unmarshal(b, &ti); err != nil { + return Report{}, err + } + r := Report{UploadCounter: ti.UpInfoData, HasUploadCounter: true} + if lim, err := q.UploadLimit(ctx, h); err == nil { + r.Limit, r.HasLimit = lim, true + } + return r, nil +} + +// UploadLimit reads the global upload limit (bytes/s, 0 = none). h nil → logs in. +func (q *QBittorrent) UploadLimit(ctx context.Context, h map[string]string) (int64, error) { + if h == nil { + var err error + if h, err = q.session(ctx, true); err != nil { + return 0, err + } + } + code, b, _, err := httpDo(ctx, q.client(), "GET", q.base()+"/api/v2/transfer/uploadLimit", h, nil, "") + if err != nil || code != 200 { + return 0, fmt.Errorf("qBittorrent uploadLimit: HTTP %d %v", code, err) + } + var v int64 + if _, err := fmt.Sscan(strings.TrimSpace(string(b)), &v); err != nil { + return 0, fmt.Errorf("qBittorrent uploadLimit: %q", strings.TrimSpace(string(b))) + } + return v, nil +} + +// SetUploadLimit sets the global upload limit (bytes/s) and returns the value qBittorrent stored (it keeps KiB/s, so +// 2 500 000 reads back as 2 499 584 — measured). +func (q *QBittorrent) SetUploadLimit(ctx context.Context, bytesPerSec int64) (int64, error) { + h, err := q.session(ctx, true) + if err != nil { + return 0, err + } + form := url.Values{"limit": {fmt.Sprint(bytesPerSec)}}.Encode() + code, _, _, err := httpDo(ctx, q.client(), "POST", q.base()+"/api/v2/transfer/setUploadLimit", h, []byte(form), "application/x-www-form-urlencoded") + if err != nil || code != 200 { + return 0, fmt.Errorf("qBittorrent setUploadLimit: HTTP %d %v", code, err) + } + return q.UploadLimit(ctx, h) +} + +// ── Transmission ───────────────────────────────────────────────────────────────────────────────────────────────── + +// Transmission asks Transmission's RPC (4.1.3 on 9202): `session-stats.current-stats.uploadedBytes` matched the +// container's sent bytes within 0.1–2 % while seeding (measured). +type Transmission struct { + Box Box + URL string // default http://transmission:9091 + HTTP *http.Client +} + +func (t *Transmission) Stack() string { return "transmission" } + +func (t *Transmission) Ask(ctx context.Context) (Report, error) { + pw, err := t.Box.Env("transmission", "PASS") + if err != nil { + return Report{}, err + } + base := t.URL + if base == "" { + base = "http://transmission:9091" + } + c := t.HTTP + if c == nil { + c = &http.Client{Timeout: 20 * time.Second} + } + h := map[string]string{"Authorization": "Basic " + base64.StdEncoding.EncodeToString([]byte("admin:"+pw)), "Host": "transmission:9091"} + body := []byte(`{"method":"session-stats"}`) + for i := 0; i < 2; i++ { + code, b, hd, err := httpDo(ctx, c, "POST", base+"/transmission/rpc", h, body, "application/json") + if err != nil { + return Report{}, err + } + if code == 409 { + h["X-Transmission-Session-Id"] = hd.Get("X-Transmission-Session-Id") + continue + } + if code != 200 { + return Report{}, fmt.Errorf("transmission session-stats: HTTP %d", code) + } + var st struct { + Arguments struct { + Current struct { + UploadedBytes uint64 `json:"uploadedBytes"` + } `json:"current-stats"` + } `json:"arguments"` + } + if err := json.Unmarshal(b, &st); err != nil { + return Report{}, err + } + return Report{UploadCounter: st.Arguments.Current.UploadedBytes, HasUploadCounter: true}, nil + } + return Report{}, fmt.Errorf("transmission session-stats: no session id") +} + +// ── Plex ───────────────────────────────────────────────────────────────────────────────────────────────────────── + +// Plex asks Plex Media Server for its playing and transcoding sessions. Measured on 9202 (1.43.5, unclaimed): both +// answer `{"MediaContainer":{"size":0}}` without a token from the docker network; a CLAIMED server needs its own +// PlexOnlineToken, read from Preferences.xml at use exactly as mediawire does (`09` §3 decision 248). +type Plex struct { + Token func() string // mediawire.PlexToken(box); "" for an unclaimed server + URL func(port int) (string, error) // mediawire.PlexURL (Plex refuses a call by NAME as non-local, R-1060) + HTTP *http.Client +} + +func (p *Plex) Stack() string { return "plex" } + +func (p *Plex) Ask(ctx context.Context) (Report, error) { + base, err := p.URL(32400) + if err != nil { + return Report{}, err + } + c := p.HTTP + if c == nil { + c = &http.Client{Timeout: 20 * time.Second} + } + h := map[string]string{"Accept": "application/json"} + if tok := p.Token(); tok != "" { + h["X-Plex-Token"] = tok + } + size := func(path string) (int, error) { + code, b, _, err := httpDo(ctx, c, "GET", base+path, h, nil, "") + if err != nil || code != 200 { + return 0, fmt.Errorf("plex %s: HTTP %d %v", path, code, err) + } + var mc struct { + MediaContainer struct { + Size int `json:"size"` + } `json:"MediaContainer"` + } + if err := json.Unmarshal(b, &mc); err != nil { + return 0, err + } + return mc.MediaContainer.Size, nil + } + v, err := size("/status/sessions") + if err != nil { + return Report{}, err + } + tr, err := size("/transcode/sessions") + if err != nil { + return Report{}, err + } + return Report{Viewers: v, HasViewerCount: true, Transcodes: tr, HasTranscodeCount: true}, nil +} diff --git a/controller/internal/upwatch/explain_test.go b/controller/internal/upwatch/explain_test.go new file mode 100644 index 0000000..21fe0a1 --- /dev/null +++ b/controller/internal/upwatch/explain_test.go @@ -0,0 +1,140 @@ +package upwatch + +import ( + "context" + "encoding/base64" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +type envBox map[string]string + +func (b envBox) Env(c, n string) (string, error) { + if v, ok := b[c+"/"+n]; ok { + return v, nil + } + return "", fmt.Errorf("no %s in %s", n, c) +} +func (b envBox) Running(string) bool { return true } + +// qBittorrent as measured on 9202: a login is 204 + a SID cookie, and only when Referer equals the base; transfer/info +// carries up_info_data; a stale session is 403. +func fakeQBServer(t *testing.T, pw string, logins *int, expireNext *bool, limit *int64) *httptest.Server { + var srv *httptest.Server + srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/api/v2/auth/login": + r.ParseForm() + if r.Header.Get("Referer") != srv.URL || r.Form.Get("username") != "admin" || r.Form.Get("password") != pw { + w.WriteHeader(401) + return + } + *logins++ + http.SetCookie(w, &http.Cookie{Name: "QBT_SID_8080", Value: fmt.Sprint("s", *logins)}) + w.WriteHeader(204) + case "/api/v2/transfer/info": + if *expireNext { + *expireNext = false + w.WriteHeader(403) + return + } + if !strings.HasPrefix(r.Header.Get("Cookie"), "QBT_SID_8080=") { + w.WriteHeader(403) + return + } + fmt.Fprint(w, `{"up_info_data":123456789,"up_info_speed":16384000,"up_rate_limit":0}`) + case "/api/v2/transfer/uploadLimit": + fmt.Fprint(w, *limit) + case "/api/v2/transfer/setUploadLimit": + r.ParseForm() + var v int64 + fmt.Sscan(r.Form.Get("limit"), &v) + *limit = v / 1024 * 1024 + w.WriteHeader(200) + default: + w.WriteHeader(404) + } + })) + return srv +} + +// RED-PROOF: send no Referer → 401 → fails; log in every minute instead of keeping the session → logins == 3 → fails. +func TestQBittorrent_AskKeepsTheSessionAndReLogsInOn403(t *testing.T) { + logins, expire, limit := 0, false, int64(0) + srv := fakeQBServer(t, "pw-canary", &logins, &expire, &limit) + defer srv.Close() + q := &QBittorrent{Box: envBox{"qbittorrent/QBT_ADMIN_PASSWORD": "pw-canary"}, URL: srv.URL} + for i := 0; i < 3; i++ { + r, err := q.Ask(context.Background()) + if err != nil || !r.HasUploadCounter || r.UploadCounter != 123456789 || !r.HasLimit { + t.Fatalf("ask %d: %+v %v", i, r, err) + } + } + if logins != 1 { + t.Fatalf("logins = %d for three minutes — the session must be kept", logins) + } + expire = true + if _, err := q.Ask(context.Background()); err != nil || logins != 2 { + t.Fatalf("after a 403: err=%v logins=%d — want one fresh login", err, logins) + } + got, err := q.SetUploadLimit(context.Background(), 2500000) + if err != nil || got != 2499584 { + t.Fatalf("SetUploadLimit = %d %v, want the stored 2499584", got, err) + } +} + +// Transmission's RPC: a 409 hands out the session id; Basic auth with the catalog's PASS. +func TestTransmission_AskFollowsTheSessionID(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Authorization") != "Basic "+base64.StdEncoding.EncodeToString([]byte("admin:tr-pw")) { + w.WriteHeader(401) + return + } + if r.Header.Get("X-Transmission-Session-Id") != "sid-1" { + w.Header().Set("X-Transmission-Session-Id", "sid-1") + w.WriteHeader(409) + return + } + fmt.Fprint(w, `{"arguments":{"current-stats":{"uploadedBytes":987654321},"uploadSpeed":0},"result":"success"}`) + })) + defer srv.Close() + tr := &Transmission{Box: envBox{"transmission/PASS": "tr-pw"}, URL: srv.URL} + r, err := tr.Ask(context.Background()) + if err != nil || r.UploadCounter != 987654321 || !r.HasUploadCounter { + t.Fatalf("Ask = %+v %v", r, err) + } +} + +// Plex: the token goes along only when the server is claimed; both session lists are read. +// RED-PROOF: always send the header → the unclaimed case sends "X-Plex-Token: " → fails. +func TestPlex_AskReadsSessionsWithTheTokenOnlyWhenClaimed(t *testing.T) { + var seen []string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, has := r.Header["X-Plex-Token"] + seen = append(seen, fmt.Sprintf("%s token=%v", r.URL.Path, has)) + switch r.URL.Path { + case "/status/sessions": + fmt.Fprint(w, `{"MediaContainer":{"size":2}}`) + case "/transcode/sessions": + fmt.Fprint(w, `{"MediaContainer":{"size":1}}`) + } + })) + defer srv.Close() + for _, tok := range []string{"", "plex-token-canary"} { + seen = nil + p := &Plex{Token: func() string { return tok }, URL: func(int) (string, error) { return srv.URL, nil }} + r, err := p.Ask(context.Background()) + if err != nil || r.Viewers != 2 || r.Transcodes != 1 || !r.HasViewerCount || !r.HasTranscodeCount { + t.Fatalf("Ask = %+v %v", r, err) + } + want := fmt.Sprintf("token=%v", tok != "") + for _, s := range seen { + if !strings.HasSuffix(s, want) { + t.Fatalf("claimed=%v: request %q", tok != "", s) + } + } + } +} diff --git a/controller/internal/upwatch/line.go b/controller/internal/upwatch/line.go new file mode 100644 index 0000000..e919ee1 --- /dev/null +++ b/controller/internal/upwatch/line.go @@ -0,0 +1,194 @@ +package upwatch + +import ( + "bufio" + "fmt" + "os" + "strconv" + "strings" + "sync" + "time" +) + +// ── The upload line, learnt from what the box already does ──────────────────────────────────────────────────────── +// +// The box's off-site push (restic, inside this controller's own container) is the only big upload the box makes on +// its own. While it runs, the controller samples its OWN network counter once a second and keeps the best 10-second +// rate. Over nights, the highest such rate is a LOWER BOUND of the household's upload line — the push may not fill +// the line (the family uploads too, the Storage Box may be slow), so the page says „at least". +// +// Measured (spike §2): at a one-minute sample the push looked like 0.3–37 Mbit/s, because most nights it moves a few +// megabytes in under a minute — the one-minute counter shows the data size, not the line. Hence a 1-second sampler, +// and a night only COUNTS when its best 10 seconds moved ≥ MinWindowBytes: a short push says nothing about the line. + +// MinWindowBytes is the least a push must move in its best 10 seconds to say anything about the line (≥ 8 Mbit/s). +const MinWindowBytes = 10 << 20 + +// KeepNights is how many nights' records the line is the maximum of. +const KeepNights = 14 + +// LineRecord is one push's measurement. +type LineRecord struct { + At string `json:"at"` // RFC3339, the push's end + Peak10Mbps float64 `json:"peak10_mbps"` // the best 10-second rate + Bytes uint64 `json:"bytes"` // everything the container sent during the push + Seconds int `json:"seconds"` + Counts bool `json:"counts"` // the best 10 seconds moved ≥ MinWindowBytes +} + +// LineFromRecords is the line estimate: the highest counting peak over the kept nights. ok=false while none counts. +func LineFromRecords(recs []LineRecord) (float64, bool) { + best, ok := 0.0, false + for _, r := range recs { + if r.Counts && r.Peak10Mbps > best { + best, ok = r.Peak10Mbps, true + } + } + return best, ok +} + +// AddRecord appends one push and keeps the newest KeepNights. +func AddRecord(recs []LineRecord, r LineRecord) []LineRecord { + recs = append(recs, r) + if len(recs) > KeepNights { + recs = recs[len(recs)-KeepNights:] + } + return recs +} + +// OwnTxBytes sums the bytes sent on every interface but loopback in THIS process's network namespace +// (/proc/self/net/dev; the 9th field of an interface line is transmit bytes). +func OwnTxBytes() (uint64, error) { + return txFromNetDev("/proc/self/net/dev") +} + +func txFromNetDev(path string) (uint64, error) { + f, err := os.Open(path) + if err != nil { + return 0, err + } + defer f.Close() + var sum uint64 + sc := bufio.NewScanner(f) + for sc.Scan() { + line := sc.Text() + i := strings.Index(line, ":") + if i < 0 { + continue + } + if strings.TrimSpace(line[:i]) == "lo" { + continue + } + fs := strings.Fields(line[i+1:]) + if len(fs) < 9 { + continue + } + v, err := strconv.ParseUint(fs[8], 10, 64) + if err != nil { + return 0, fmt.Errorf("%s: %q: %w", path, line, err) + } + sum += v + } + return sum, sc.Err() +} + +// Point is one reading of the counter. +type Point struct { + At time.Time + V uint64 +} + +// Summarize turns a push's readings into its record: total bytes, seconds, and the best rate over any window of at +// least 10 seconds (for each start, the first reading ≥ 10 s later). A counter that went down is no measurement. +func Summarize(pts []Point, end time.Time) LineRecord { + r := LineRecord{At: end.UTC().Format(time.RFC3339)} + if len(pts) < 2 { + return r + } + first, last := pts[0], pts[len(pts)-1] + if last.V >= first.V { + r.Bytes = last.V - first.V + } + r.Seconds = int(last.At.Sub(first.At).Seconds() + 0.5) + var bestBytes uint64 + var bestRate float64 + j := 0 + for i := range pts { + if j < i { + j = i + } + for j < len(pts) && pts[j].At.Sub(pts[i].At) < 10*time.Second { + j++ + } + if j >= len(pts) { + break + } + if pts[j].V < pts[i].V { + continue + } + b := pts[j].V - pts[i].V + rate := float64(b) * 8 / pts[j].At.Sub(pts[i].At).Seconds() / 1e6 + if rate > bestRate { + bestRate, bestBytes = rate, b + } + } + r.Peak10Mbps = round2(bestRate) + r.Counts = bestBytes >= MinWindowBytes + return r +} + +// PushSampler samples a byte counter once a second between Start and the stop function it returns. +type PushSampler struct { + Read func() (uint64, error) // default OwnTxBytes + Every time.Duration // default 1 s + Now func() time.Time +} + +// Start begins sampling; the returned stop ends it and returns the push's record. +func (p *PushSampler) Start() (stop func() LineRecord) { + read := p.Read + if read == nil { + read = OwnTxBytes + } + every := p.Every + if every <= 0 { + every = time.Second + } + now := p.Now + if now == nil { + now = time.Now + } + var mu sync.Mutex + var pts []Point + take := func() { + if v, err := read(); err == nil { + mu.Lock() + pts = append(pts, Point{now(), v}) + mu.Unlock() + } + } + take() + done := make(chan struct{}) + finished := make(chan struct{}) + go func() { + defer close(finished) + t := time.NewTicker(every) + defer t.Stop() + for { + select { + case <-done: + return + case <-t.C: + take() + } + } + }() + return func() LineRecord { + close(done) + <-finished + take() + mu.Lock() + defer mu.Unlock() + return Summarize(pts, now()) + } +} diff --git a/controller/internal/upwatch/line_test.go b/controller/internal/upwatch/line_test.go new file mode 100644 index 0000000..c012767 --- /dev/null +++ b/controller/internal/upwatch/line_test.go @@ -0,0 +1,147 @@ +package upwatch + +import ( + "context" + "os" + "path/filepath" + "sync" + "testing" + "time" +) + +// The best ≥ 10-second rate, and a push only COUNTS when that window moved ≥ 10 MB. +// RED-PROOF: compute the peak over 1-second windows → the 4 MB burst reads as 33 Mbit/s → fails. +func TestLine_SummarizePeakAndCounts(t *testing.T) { + series := func(perSecond []uint64) []Point { + t0 := time.Date(2026, 10, 24, 2, 16, 0, 0, time.UTC) + pts := []Point{{t0, 0}} + for i, b := range perSecond { + pts = append(pts, Point{t0.Add(time.Duration(i+1) * time.Second), pts[len(pts)-1].V + b}) + } + return pts + } + steady := make([]uint64, 60) + for i := range steady { + steady[i] = 50e6 / 8 + } + if r := Summarize(series(steady), time.Now()); r.Peak10Mbps < 49.9 || r.Peak10Mbps > 50.1 || !r.Counts || r.Seconds != 60 { + t.Fatalf("steady 50 Mbit/s push: %+v", r) + } + burst := make([]uint64, 30) + burst[3] = 4 << 20 + if r := Summarize(series(burst), time.Now()); r.Counts || r.Peak10Mbps > 4 { + t.Fatalf("a 4 MB one-second push must not set the line: %+v", r) + } +} + +// The sampler itself: readings from a live (fast) ticker reach the summary. +func TestLine_PushSamplerCollects(t *testing.T) { + var mu sync.Mutex + var v uint64 + p := &PushSampler{Every: time.Millisecond, Read: func() (uint64, error) { mu.Lock(); defer mu.Unlock(); v += 1000; return v, nil }} + stop := p.Start() + time.Sleep(30 * time.Millisecond) + r := stop() + if r.Bytes < 2000 { + t.Fatalf("sampler collected %d bytes", r.Bytes) + } +} + +func TestLine_FromRecordsUsesOnlyCountingNightsAndKeepsFourteen(t *testing.T) { + if _, ok := LineFromRecords(nil); ok { + t.Fatal("no record: the line must be unknown") + } + recs := []LineRecord{{Peak10Mbps: 90, Counts: false}, {Peak10Mbps: 40, Counts: true}, {Peak10Mbps: 55, Counts: true}} + if l, ok := LineFromRecords(recs); !ok || l != 55 { + t.Fatalf("line = %v %v, want 55 (the 90 did not count)", l, ok) + } + var keep []LineRecord + for i := 0; i < 20; i++ { + keep = AddRecord(keep, LineRecord{Peak10Mbps: float64(i), Counts: true}) + } + if len(keep) != KeepNights || keep[0].Peak10Mbps != 6 { + t.Fatalf("kept %d, first %v", len(keep), keep[0].Peak10Mbps) + } +} + +func TestLine_TxFromNetDevSkipsLoopback(t *testing.T) { + p := filepath.Join(t.TempDir(), "dev") + os.WriteFile(p, []byte(`Inter-| Receive | Transmit + face |bytes packets errs drop fifo frame compressed multicast|bytes packets errs drop fifo colls carrier compressed + lo: 9000 10 0 0 0 0 0 0 9000 10 0 0 0 0 0 0 + eth0: 100 1 0 0 0 0 0 0 774708 5 0 0 0 0 0 0 + eth1: 1 1 0 0 0 0 0 0 1000 1 0 0 0 0 0 0 +`), 0o644) + v, err := txFromNetDev(p) + if err != nil || v != 775708 { + t.Fatalf("tx = %d %v, want 775708 (eth0+eth1, not lo)", v, err) + } +} + +// ── qBittorrent's default limit ── + +type fakeQB struct { + cur int64 + sets []int64 +} + +func (f *fakeQB) UploadLimit(context.Context, map[string]string) (int64, error) { return f.cur, nil } +func (f *fakeQB) SetUploadLimit(_ context.Context, b int64) (int64, error) { + f.sets = append(f.sets, b) + f.cur = b / 1024 * 1024 // qBittorrent keeps KiB/s + return f.cur, nil +} + +// RED-PROOF per row: drop the household check → the „household" rows set a limit → fail; compare against the +// requested value instead of the stored one → the box's own rounding reads as a household change → the re-set row fails. +func TestQBLimit_TheHouseholdAlwaysWins(t *testing.T) { + now := time.Date(2026, 10, 24, 3, 0, 0, 0, time.UTC) + ctx := context.Background() + // 1. Unlimited, never set, line 100 Mbit/s, share 50 % → 6 250 000 B/s requested, stored rounded. + q := &fakeQB{} + st, act, err := ApplyQBLimit(ctx, q, QBLimitState{}, 100, true, 0.5, now) + if err != nil || act != "set" || len(q.sets) != 1 || q.sets[0] != 6250000 || st.BoxSetBytes != 6249472 { + t.Fatalf("first set: %s %v %+v sets=%v", act, err, st, q.sets) + } + // 2. Same line, qBittorrent still holds the box's (rounded) value → kept, no write. + st2, act, _ := ApplyQBLimit(ctx, q, st, 105, true, 0.5, now) + if act != "kept" || len(q.sets) != 1 { + t.Fatalf("unchanged: %s sets=%v", act, q.sets) + } + // 3. The line moved > 20 % → re-set. + st3, act, _ := ApplyQBLimit(ctx, q, st2, 200, true, 0.5, now) + if act != "set" || len(q.sets) != 2 || st3.LineMbps != 200 { + t.Fatalf("line moved: %s %+v", act, st3) + } + // 4. The household changed it in qBittorrent → theirs, never touched again, even when the line moves. + q.cur = 1000000 + st4, act, _ := ApplyQBLimit(ctx, q, st3, 200, true, 0.5, now) + if act != "household" || !st4.Household { + t.Fatalf("household change: %s %+v", act, st4) + } + if _, act, _ := ApplyQBLimit(ctx, q, st4, 50, true, 0.5, now); act != "household" || len(q.sets) != 2 { + t.Fatalf("after the household took it: %s sets=%v", act, q.sets) + } + // 5. The household REMOVED the box's limit (0) → theirs too. + q2 := &fakeQB{} + s, _, _ := ApplyQBLimit(ctx, q2, QBLimitState{}, 100, true, 0.5, now) + q2.cur = 0 + if s2, act, _ := ApplyQBLimit(ctx, q2, s, 100, true, 0.5, now); act != "household" || !s2.Household { + t.Fatalf("household removed the limit: %s", act) + } + // 6. A limit already there before the box ever looked → the household's. + q3 := &fakeQB{cur: 500000} + if s3, act, _ := ApplyQBLimit(ctx, q3, QBLimitState{}, 100, true, 0.5, now); act != "household" || !s3.Household || len(q3.sets) != 0 { + t.Fatalf("pre-existing limit: %s", act) + } + // 6b. That household later REMOVES its own limit (unlimited again) → still theirs: the box does not step in. + q3.cur = 0 + if _, act, _ := ApplyQBLimit(ctx, q3, QBLimitState{Household: true}, 100, true, 0.5, now); act != "household" || len(q3.sets) != 0 { + t.Fatalf("household set unlimited after its own limit: %s sets=%v — the box overrode their choice", act, q3.sets) + } + // 7. No line measured → nothing set. + q4 := &fakeQB{} + if _, act, _ := ApplyQBLimit(ctx, q4, QBLimitState{}, 0, false, 0.5, now); act != "no-line" || len(q4.sets) != 0 { + t.Fatalf("no line: %s", act) + } +} diff --git a/controller/internal/upwatch/qblimit.go b/controller/internal/upwatch/qblimit.go new file mode 100644 index 0000000..a6545f3 --- /dev/null +++ b/controller/internal/upwatch/qblimit.go @@ -0,0 +1,77 @@ +package upwatch + +import ( + "context" + "math" + "time" +) + +// ── qBittorrent's default upload limit (operator ruling 2026-10-23 19:22; the size is decide row R-1245) ───────── +// +// qBittorrent seeds at whatever the line gives (measured on 9202: 131 Mbit/s until the receiver capped it), so a +// family's uploads — a video call, a photo sync — wait behind it. The box sets qBittorrent's GLOBAL upload limit to a +// share of the measured line, and leaves it alone the moment the household touches it: +// +// - never set by the box, qBittorrent unlimited (0) → the box sets it +// - never set by the box, qBittorrent already has a limit → the household's: never touched +// - set by the box, qBittorrent still holds exactly that value → the box's: re-set when the line moves > 20 % +// - set by the box, qBittorrent holds anything else (0 included) → the household changed it: never touched again +// +// The value compared is the one qBittorrent STORED (it keeps KiB/s: 2 500 000 reads back as 2 499 584, measured), so +// the box's own rounding is never mistaken for a household's change. The limit survives a qBittorrent restart +// (measured: `Session\GlobalUPSpeedLimit` in its config). No line measured yet → nothing is set. +// +// Edge: an app reinstalled from scratch starts unlimited while the record still says „set by the box"; that reads as +// the household removing the limit and is left alone. Recorded in the design (`08` §9.7), not guarded. + +// QBLimitState is what the box remembers (settings.json `upwatch.qb_limit`). +type QBLimitState struct { + BoxSetBytes int64 `json:"box_set_bytes,omitempty"` // the value qBittorrent stored when the box last set it + LineMbps float64 `json:"line_mbps,omitempty"` // the line the box set it from + SetAt string `json:"set_at,omitempty"` + Household bool `json:"household,omitempty"` // the household owns the limit: never touched again + HouseholdAt string `json:"household_at,omitempty"` +} + +// QBLimiter is the qBittorrent side the rule needs (the QBittorrent explainer satisfies it). +type QBLimiter interface { + UploadLimit(ctx context.Context, h map[string]string) (int64, error) + SetUploadLimit(ctx context.Context, bytesPerSec int64) (int64, error) +} + +// QBLimitTarget is share × line, in bytes/s. +func QBLimitTarget(lineMbps, share float64) int64 { + return int64(math.Round(lineMbps * share * 1e6 / 8)) +} + +// ApplyQBLimit runs the rule once. It returns the new state and what happened ("set", "kept", "household", "no-line", +// "error"). It never lowers or raises a limit the household owns. +func ApplyQBLimit(ctx context.Context, q QBLimiter, st QBLimitState, lineMbps float64, lineKnown bool, share float64, now time.Time) (QBLimitState, string, error) { + if st.Household { + return st, "household", nil + } + cur, err := q.UploadLimit(ctx, nil) + if err != nil { + return st, "error", err + } + if st.BoxSetBytes == 0 { + if cur != 0 { + st.Household, st.HouseholdAt = true, now.UTC().Format(time.RFC3339) + return st, "household", nil + } + } else if cur != st.BoxSetBytes { + st.Household, st.HouseholdAt = true, now.UTC().Format(time.RFC3339) + return st, "household", nil + } else if lineKnown && st.LineMbps > 0 && math.Abs(lineMbps-st.LineMbps)/st.LineMbps <= 0.2 { + return st, "kept", nil + } + if !lineKnown { + return st, "no-line", nil + } + stored, err := q.SetUploadLimit(ctx, QBLimitTarget(lineMbps, share)) + if err != nil { + return st, "error", err + } + st.BoxSetBytes, st.LineMbps, st.SetAt = stored, lineMbps, now.UTC().Format(time.RFC3339) + return st, "set", nil +} diff --git a/controller/internal/upwatch/service.go b/controller/internal/upwatch/service.go new file mode 100644 index 0000000..8c58736 --- /dev/null +++ b/controller/internal/upwatch/service.go @@ -0,0 +1,155 @@ +package upwatch + +import ( + "context" + "encoding/json" + "log" + "time" + + "gitea.dooplex.hu/admin/felhom-controller/internal/logx" +) + +// State is what the watch keeps in settings.json (`upwatch`). +type State struct { + Line []LineRecord `json:"line,omitempty"` + QBLimit *QBLimitState `json:"qb_limit,omitempty"` +} + +// Store is the settings half the service needs (settings.Settings satisfies it). +type Store interface { + GetUpwatch() json.RawMessage + UpdateUpwatch(fn func(old json.RawMessage) (json.RawMessage, error)) error +} + +func decode(raw json.RawMessage) State { + var st State + if len(raw) > 0 { + _ = json.Unmarshal(raw, &st) // a damaged block reads as empty: the line is re-learnt, the limit record too + } + return st +} + +// Service ties the watch together for main: the minute watcher, the push sampler, the qBittorrent limit rule and +// what the pages show. +type Service struct { + Store Store + Watcher *Watcher + QB *QBittorrent // nil when the box has no qBittorrent wiring + // QBRunning reports whether qBittorrent is installed and running now. + QBRunning func() bool + // QBShare is the share of the measured line qBittorrent may upload at (operator decide row; CC's pick 0.5). + QBShare float64 + Log *log.Logger +} + +// Line is the measured upload line in Mbit/s (a lower bound); ok=false while no push counted. +func (s *Service) Line() (float64, bool) { + return LineFromRecords(decode(s.Store.GetUpwatch()).Line) +} + +// RecordPush stores one off-site push's measurement. +func (s *Service) RecordPush(r LineRecord) { + err := s.Store.UpdateUpwatch(func(old json.RawMessage) (json.RawMessage, error) { + st := decode(old) + st.Line = AddRecord(st.Line, r) + return json.Marshal(st) + }) + if err != nil { + logx.Warnf(s.Log, "[upwatch] saving the push measurement: %v", err) + return + } + l, ok := s.Line() + logx.Infof(s.Log, "[upwatch] off-site push measured: %d bytes in %d s, best 10 s %.1f Mbit/s (counts=%v); line now %s", + r.Bytes, r.Seconds, r.Peak10Mbps, r.Counts, lineWords(l, ok)) +} + +func lineWords(l float64, ok bool) string { + if !ok { + return "unknown (learning)" + } + return "at least " + formatMbps(l) + " Mbit/s" +} + +// PushObserver is handed to the off-site backup: it is called at the push's start and returns the function that +// ends the measurement. +func (s *Service) PushObserver() func() { + stop := (&PushSampler{}).Start() + return func() { s.RecordPush(stop()) } +} + +// QBLimitNow runs qBittorrent's limit rule once (main calls it every 15 minutes; a no-op without qBittorrent). +func (s *Service) QBLimitNow(ctx context.Context) { + if s.QB == nil || (s.QBRunning != nil && !s.QBRunning()) { + return + } + line, known := s.Line() + cur := decode(s.Store.GetUpwatch()) + var st QBLimitState + if cur.QBLimit != nil { + st = *cur.QBLimit + } + ns, act, err := ApplyQBLimit(ctx, s.QB, st, line, known, s.QBShare, time.Now()) + if err != nil { + logx.Debugf(s.Log, "[upwatch] qBittorrent limit: %v — tried again in 15 minutes", err) + return + } + if ns != st { + if err := s.Store.UpdateUpwatch(func(old json.RawMessage) (json.RawMessage, error) { + o := decode(old) + o.QBLimit = &ns + return json.Marshal(o) + }); err != nil { + logx.Warnf(s.Log, "[upwatch] saving the qBittorrent limit record: %v", err) + return + } + } + switch act { + case "set": + logx.Infof(s.Log, "[upwatch] qBittorrent upload limit set to %d B/s (%.0f %% of the measured line, %.1f Mbit/s)", ns.BoxSetBytes, s.QBShare*100, line) + case "household": + if !st.Household { + logx.Infof(s.Log, "[upwatch] qBittorrent's upload limit is the household's — the box leaves it alone from now on") + } + } +} + +// View is what the pages show. +type View struct { + LineMbps float64 + LineKnown bool + QBLimitBps int64 // the limit the box set and still owns; 0 = none shown +} + +// View reads the state for the pages. +func (s *Service) View() View { + st := decode(s.Store.GetUpwatch()) + v := View{} + v.LineMbps, v.LineKnown = LineFromRecords(st.Line) + if st.QBLimit != nil && !st.QBLimit.Household { + v.QBLimitBps = st.QBLimit.BoxSetBytes + } + return v +} + +// MbpsText renders bytes/s as Mbit/s for a page line (one decimal under 10). +func MbpsText(bps int64) string { return formatMbps(float64(bps) * 8 / 1e6) } + +func formatMbps(v float64) string { + if v < 10 { + return trimZero(v) + } + return itoa(int64(v + 0.5)) +} + +func trimZero(v float64) string { + x := int64(v*10 + 0.5) + if x%10 == 0 { + return itoa(x / 10) + } + return itoa(x/10) + "." + itoa(x%10) +} + +func itoa(v int64) string { + b, _ := json.Marshal(v) + return string(b) +} diff --git a/controller/internal/upwatch/source.go b/controller/internal/upwatch/source.go new file mode 100644 index 0000000..93a6f5c --- /dev/null +++ b/controller/internal/upwatch/source.go @@ -0,0 +1,127 @@ +// Package upwatch is the upload and CPU watch (felhom.eu `architecture/08-alarm-ladder.md` §9 [DESIGN — proposed], +// spike `audits/SPIKE-upload-watch-2026-10-23.md`). Once a minute it reads every app container's EXACT sent bytes and +// CPU time, adds them up per app, and when an app sends a lot (or burns CPU) for a long time it ASKS THE APP whether +// that is its normal work — qBittorrent's and Transmission's own upload counters, Plex's playing and transcoding +// sessions. Work the app explains is normal use. Work it does not explain is the sign of a hacked app (DooPlex's Plex +// sent 474 GB in one day with no viewer, 2026-10-23) and becomes ONE operator-only event, cleared when it stops. +// +// It also learns the box's upload line from what the box already does (its own off-site push, line.go) and gives +// qBittorrent a default upload limit from it, never over a limit the household set (qblimit.go). +// +// SECRETS: app passwords and tokens are read at use (the mediawire seams, `09` §3 decision 248), sent only to the +// app's own container, never stored, never logged. +package upwatch + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net" + "net/http" + "strings" + "time" +) + +// Sample is one container's cumulative counters at one moment. +type Sample struct { + Container string + Stack string // the compose project; "" for a container outside any stack + TxBytes uint64 // bytes sent on every network of the container, since it started + CPUNanos uint64 // CPU time used, since it started + Started string // the container's start time: a change means the counters restarted from zero +} + +// Source returns one Sample per running container. +type Source interface { + Samples(ctx context.Context) ([]Sample, error) +} + +// DockerAPI reads the counters from the Docker Engine API on the guest's socket. +// +// WHY NOT `docker stats` (what internal/metrics uses): its NetIO is text with three significant digits ("1.23GB"), so +// at a large total one step is ~10 MB — 1.3 Mbit/s at a one-minute sample. The Engine API gives integers; measured on +// 9202 it equals the kernel's own counter in the container's network namespace to the byte +// (`audits/SPIKE-upload-watch-2026-10-23/s4-qb-seed-nolimit.jsonl`: proc_tx == engine_tx). +type DockerAPI struct { + Socket string // default /var/run/docker.sock + client *http.Client +} + +func (d *DockerAPI) http() *http.Client { + if d.client == nil { + sock := d.Socket + if sock == "" { + sock = "/var/run/docker.sock" + } + d.client = &http.Client{ + Timeout: 20 * time.Second, + Transport: &http.Transport{DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) { + var dl net.Dialer + return dl.DialContext(ctx, "unix", sock) // dialled by path every time: sockheal may re-create the file + }}, + } + } + return d.client +} + +func (d *DockerAPI) get(ctx context.Context, path string, out interface{}) error { + req, err := http.NewRequestWithContext(ctx, "GET", "http://docker"+path, nil) + if err != nil { + return err + } + resp, err := d.http().Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode != 200 { + b, _ := io.ReadAll(io.LimitReader(resp.Body, 512)) + return fmt.Errorf("docker API %s: HTTP %d %s", path, resp.StatusCode, strings.TrimSpace(string(b))) + } + return json.NewDecoder(io.LimitReader(resp.Body, 8<<20)).Decode(out) +} + +// Samples lists the running containers and reads each one's counters (one-shot stats: no one-second wait). +func (d *DockerAPI) Samples(ctx context.Context) ([]Sample, error) { + var list []struct { + ID string `json:"Id"` + Names []string `json:"Names"` + Labels map[string]string `json:"Labels"` + } + if err := d.get(ctx, "/containers/json", &list); err != nil { + return nil, err + } + out := make([]Sample, 0, len(list)) + for _, c := range list { + var st struct { + Read string `json:"read"` + Networks map[string]struct { + TxBytes uint64 `json:"tx_bytes"` + } `json:"networks"` + CPUStats struct { + CPUUsage struct { + TotalUsage uint64 `json:"total_usage"` + } `json:"cpu_usage"` + } `json:"cpu_stats"` + } + if err := d.get(ctx, "/containers/"+c.ID+"/stats?stream=false&one-shot=true", &st); err != nil { + continue // the container stopped between the list and the read: it is simply not sampled this minute + } + var started struct { + State struct { + StartedAt string `json:"StartedAt"` + } `json:"State"` + } + _ = d.get(ctx, "/containers/"+c.ID+"/json", &started) + s := Sample{Stack: c.Labels["com.docker.compose.project"], CPUNanos: st.CPUStats.CPUUsage.TotalUsage, Started: started.State.StartedAt} + if len(c.Names) > 0 { + s.Container = strings.TrimPrefix(c.Names[0], "/") + } + for _, n := range st.Networks { + s.TxBytes += n.TxBytes + } + out = append(out, s) + } + return out, nil +} diff --git a/controller/internal/upwatch/watch.go b/controller/internal/upwatch/watch.go new file mode 100644 index 0000000..c64d911 --- /dev/null +++ b/controller/internal/upwatch/watch.go @@ -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 } diff --git a/controller/internal/upwatch/watch_test.go b/controller/internal/upwatch/watch_test.go new file mode 100644 index 0000000..a0a8126 --- /dev/null +++ b/controller/internal/upwatch/watch_test.go @@ -0,0 +1,312 @@ +package upwatch + +import ( + "context" + "errors" + "io" + "log" + "testing" + "time" +) + +// The fake box: containers whose counters the test drives minute by minute, and apps that answer what the test says. + +type fakeSource struct{ s map[string]*Sample } + +func (f *fakeSource) Samples(context.Context) ([]Sample, error) { + var out []Sample + for _, s := range f.s { + out = append(out, *s) + } + return out, nil +} + +func (f *fakeSource) add(container, stack string) *Sample { + if f.s == nil { + f.s = map[string]*Sample{} + } + s := &Sample{Container: container, Stack: stack, Started: "t0"} + f.s[container] = s + return s +} + +type fakeApp struct { + stack string + rep Report + err error +} + +func (a *fakeApp) Stack() string { return a.stack } +func (a *fakeApp) Ask(context.Context) (Report, error) { return a.rep, a.err } + +type event struct { + t string + d Details +} + +type rig struct { + w *Watcher + src *fakeSource + now time.Time + ev []event +} + +func newRig(ex ...*fakeApp) *rig { + r := &rig{src: &fakeSource{}, now: time.Date(2026, 10, 23, 22, 0, 0, 0, time.UTC)} + m := map[string]Explainer{} + for _, e := range ex { + m[e.stack] = e + } + rules := DefaultRules() + rules.Cores = 4 + r.w = &Watcher{Source: r.src, Explainers: m, Rules: rules, Log: log.New(io.Discard, "", 0), + Now: func() time.Time { return r.now }, + Emit: func(t, _ string, d Details) { r.ev = append(r.ev, event{t, d}) }} + return r +} + +const mbit = 1e6 / 8 // bytes per second at 1 Mbit/s + +// minute advances the clock one minute; each container sends txMbps and burns cpuPct (of one core) during it. +func (r *rig) minute(t *testing.T, step func()) { + t.Helper() + if step != nil { + step() + } + r.now = r.now.Add(time.Minute) + r.w.Tick(context.Background()) +} + +func send(s *Sample, mbps, cpuPct float64) { + s.TxBytes += uint64(mbps * mbit * 60) + s.CPUNanos += uint64(cpuPct / 100 * 60e9) +} + +func (r *rig) count(t string) int { + n := 0 + for _, e := range r.ev { + if e.t == t { + n++ + } + } + return n +} + +// A real torrent seeding at full speed: qBittorrent's own counter explains the bytes (the measured 3–7 % overhead). +// RED-PROOF: make the explained test always false → an alarm fires → fails. +func TestUpwatch_SeedingQBittorrentIsExplained(t *testing.T) { + qb := &fakeApp{stack: "qbittorrent", rep: Report{HasUploadCounter: true}} + r := newRig(qb) + c := r.src.add("qbittorrent", "qbittorrent") + r.w.Tick(context.Background()) + for i := 0; i < 90; i++ { + r.minute(t, func() { + send(c, 131*1.05, 20) // the container: the app's bytes + 5 % protocol overhead + qb.rep.UploadCounter += uint64(131 * mbit * 60) // what qBittorrent counted + }) + } + if len(r.ev) != 0 { + t.Fatalf("a seeding qBittorrent raised %v — its own counter explains the upload", r.ev) + } +} + +// The fake hacked app: code inside qBittorrent's container floods out while qBittorrent counts nothing. ONE alarm after +// 15 minutes, none more while it goes on, ONE clear 15 minutes after it stops. +// RED-PROOF: drop `st.upAlarm = true` (no edge) → an alarm every 15 minutes → count != 1 → fails; drop the clear → +// no cleared event → fails. +func TestUpwatch_HackedQBittorrentAlarmsOnceAndClears(t *testing.T) { + qb := &fakeApp{stack: "qbittorrent", rep: Report{HasUploadCounter: true}} + r := newRig(qb) + c := r.src.add("qbittorrent", "qbittorrent") + r.w.Tick(context.Background()) + for i := 1; i <= 14; i++ { + r.minute(t, func() { send(c, 900, 30) }) + } + if len(r.ev) != 0 { + t.Fatalf("alarm before 15 minutes: %v", r.ev) + } + r.minute(t, func() { send(c, 900, 30) }) + if r.count("app_unexplained_upload") != 1 { + t.Fatalf("after 15 minutes of 900 Mbit/s with qBittorrent counting 0: events %v, want one app_unexplained_upload", r.ev) + } + d := r.ev[0].d + if d.App != "qbittorrent" || d.Kind != "upload" || d.AppMbps != 0 || d.RateMbps < 899 || d.Minutes != 15 || d.Explainer != "qbittorrent" { + t.Fatalf("details = %+v", d) + } + for i := 0; i < 120; i++ { + r.minute(t, func() { send(c, 900, 30) }) + } + if r.count("app_unexplained_upload") != 1 { + t.Fatalf("a flood that goes on alarmed %d times — must be ONE mail", r.count("app_unexplained_upload")) + } + for i := 0; i < 14; i++ { + r.minute(t, func() { send(c, 0.01, 1) }) + } + if r.count("app_unexplained_upload_cleared") != 0 { + t.Fatal("cleared before 15 quiet minutes") + } + r.minute(t, func() { send(c, 0.01, 1) }) + if r.count("app_unexplained_upload_cleared") != 1 { + t.Fatalf("15 quiet minutes after the flood: events %v, want one clear", r.ev) + } +} + +// Plex2Shell's signature: Plex sends a lot with NO viewer. With a viewer, the same bytes are normal use. +// RED-PROOF: drop the `winViewers > 0` explanation → the viewer case alarms → fails. +func TestUpwatch_PlexWithNoViewerAlarmsWithAViewerDoesNot(t *testing.T) { + for _, viewers := range []int{0, 2} { + px := &fakeApp{stack: "plex", rep: Report{HasViewerCount: true, Viewers: viewers, HasTranscodeCount: true}} + r := newRig(px) + c := r.src.add("plex", "plex") + r.w.Tick(context.Background()) + for i := 0; i < 20; i++ { + r.minute(t, func() { send(c, 44, 10) }) + } + got := r.count("app_unexplained_upload") + if viewers == 0 && got != 1 { + t.Fatalf("Plex sending 44 Mbit/s with no viewer for 20 min: %d alarm(s), want 1", got) + } + if viewers > 0 && got != 0 { + t.Fatalf("Plex sending to %d viewers alarmed: %v", viewers, r.ev) + } + } +} + +// An app the box cannot ask: no alarm while it has no history (learning), then only far above its own p95 for an hour. +// RED-PROOF: let a no-explainer app use the 15-minute rule → it alarms at 15 → fails. +func TestUpwatch_NoExplainerLearnsThenNeedsTenTimesItsP95ForAnHour(t *testing.T) { + r := newRig() + c := r.src.add("nextcloud", "nextcloud") + r.w.Tick(context.Background()) + for i := 0; i < 90; i++ { + r.minute(t, func() { send(c, 50, 5) }) + } + if len(r.ev) != 0 { + t.Fatalf("an app with no history and no explainer alarmed: %v", r.ev) + } + r = newRig() + c = r.src.add("nextcloud", "nextcloud") + r.w.Baseline = func(stack string, cs []string) (float64, bool) { return 2, true } // p95 = 2 Mbit/s → trigger 20 + r.w.Tick(context.Background()) + for i := 0; i < 59; i++ { + r.minute(t, func() { send(c, 50, 5) }) + } + if len(r.ev) != 0 { + t.Fatalf("alarmed before an hour: %v", r.ev) + } + r.minute(t, func() { send(c, 50, 5) }) + if r.count("app_unexplained_upload") != 1 || r.ev[0].d.Explainer != "none" || r.ev[0].d.AppMbps != -1 { + t.Fatalf("an hour at 25× its p95: %v", r.ev) + } + // Below 10× its p95 it is never judged. + r = newRig() + c = r.src.add("nextcloud", "nextcloud") + r.w.Baseline = func(string, []string) (float64, bool) { return 2, true } + r.w.Tick(context.Background()) + for i := 0; i < 120; i++ { + r.minute(t, func() { send(c, 15, 5) }) + } + if len(r.ev) != 0 { + t.Fatalf("15 Mbit/s (7.5× its p95) alarmed: %v", r.ev) + } +} + +// The CPU rule: a burner in Plex with no transcode → ONE alarm after 30 minutes; with a transcode, or while the box +// itself works on the app (backup, update, restore), none. +// RED-PROOF: ignore Transcodes → the transcoding case alarms → fails; ignore Busy → the busy case alarms → fails. +func TestUpwatch_CPUBurnerAlarmsTranscodeAndBoxWorkDoNot(t *testing.T) { + for _, tc := range []struct { + name string + transcodes int + busy bool + want int + }{{"burner", 0, false, 1}, {"transcoding", 1, false, 0}, {"box busy", 0, true, 0}} { + px := &fakeApp{stack: "plex", rep: Report{HasViewerCount: true, HasTranscodeCount: true, Transcodes: tc.transcodes}} + r := newRig(px) + r.w.Busy = func(string) bool { return tc.busy } + c := r.src.add("plex", "plex") + r.w.Tick(context.Background()) + for i := 0; i < 29; i++ { + r.minute(t, func() { send(c, 0.1, 380) }) // 4 cores: the trigger is max(150, 200) = 200 % + } + if len(r.ev) != 0 { + t.Fatalf("%s: alarm before 30 minutes: %v", tc.name, r.ev) + } + r.minute(t, func() { send(c, 0.1, 380) }) + if got := r.count("app_unexplained_cpu"); got != tc.want { + t.Fatalf("%s: %d app_unexplained_cpu, want %d (%v)", tc.name, got, tc.want, r.ev) + } + if tc.want == 1 { + for i := 0; i < 15; i++ { + r.minute(t, func() { send(c, 0.1, 5) }) + } + if r.count("app_unexplained_cpu_cleared") != 1 { + t.Fatalf("burner stopped 15 min: %v", r.ev) + } + } + } +} + +// A container that restarts is a new counter, even when the new run's first reading is LARGER than the old run's last +// (a busy first minute after a restart) — that minute has no delta. Infrastructure is never judged. +// RED-PROOF: drop the Started comparison → the restart minute computes 1 TiB − old → a huge rate → fails. +func TestUpwatch_RestartAndInfrastructureAreNotJudged(t *testing.T) { + qb := &fakeApp{stack: "qbittorrent", rep: Report{HasUploadCounter: true}} + r := newRig(qb) + c := r.src.add("qbittorrent", "qbittorrent") + ctl := r.src.add("felhom-controller", "felhom-controller") + cf := r.src.add("cloudflared", "cloudflared") + r.w.Tick(context.Background()) + r.minute(t, func() { send(c, 0.1, 1) }) + r.minute(t, func() { c.TxBytes, c.Started = 1<<40, "t1" }) // restarted, and its new counter is already 1 TiB + if got := r.w.apps["qbittorrent"].lastRate; got > 1 { + t.Fatalf("the restart minute was judged at %.0f Mbit/s — a delta across two container runs", got) + } + r.minute(t, func() { send(c, 0.1, 1) }) + if got := r.w.apps["qbittorrent"].lastRate; got < 0.05 || got > 0.2 { + t.Fatalf("the minute after a restart: %.3f Mbit/s, want 0.1 (the new run's own delta)", got) + } + for i := 0; i < 40; i++ { + r.minute(t, func() { + send(ctl, 900, 400) // the off-site push and the tunnel: never an app alarm + send(cf, 900, 10) + }) + } + if len(r.ev) != 0 { + t.Fatalf("infrastructure raised %v", r.ev) + } + if _, judged := r.w.apps["felhom-controller"]; judged { + t.Fatal("the controller's own stack was judged as an app") + } +} + +// An app that does not answer is „no number": its flood still alarms (the hacked app may refuse its own API). +// RED-PROOF: treat an error as "explained" → no alarm → fails. +func TestUpwatch_AnAppThatDoesNotAnswerIsNotAnExplanation(t *testing.T) { + qb := &fakeApp{stack: "qbittorrent", err: errors.New("connection refused")} + r := newRig(qb) + c := r.src.add("qbittorrent", "qbittorrent") + r.w.Tick(context.Background()) + for i := 0; i < 15; i++ { + r.minute(t, func() { send(c, 100, 5) }) + } + if r.count("app_unexplained_upload") != 1 || r.ev[0].d.AppMbps != -1 { + t.Fatalf("a silent qBittorrent flooding: %v", r.ev) + } +} + +// The trigger follows the measured line: 25 % of it, never under the 5 Mbit/s floor. +func TestUpwatch_TriggerFollowsTheLine(t *testing.T) { + r := newRig() + if got := r.w.UploadTrigger(); got != 5 { + t.Fatalf("no line: trigger %v, want the 5 Mbit/s floor", got) + } + r.w.Line = func() (float64, bool) { return 100, true } + if got := r.w.UploadTrigger(); got != 25 { + t.Fatalf("100 Mbit/s line: trigger %v, want 25", got) + } + r.w.Line = func() (float64, bool) { return 8, true } + if got := r.w.UploadTrigger(); got != 5 { + t.Fatalf("8 Mbit/s line: trigger %v, want the floor", got) + } +} diff --git a/controller/internal/web/handlers.go b/controller/internal/web/handlers.go index 0b1c352..e0cb1a6 100644 --- a/controller/internal/web/handlers.go +++ b/controller/internal/web/handlers.go @@ -27,6 +27,7 @@ import ( "gitea.dooplex.hu/admin/felhom-controller/internal/settings" "gitea.dooplex.hu/admin/felhom-controller/internal/stacks" "gitea.dooplex.hu/admin/felhom-controller/internal/system" + "gitea.dooplex.hu/admin/felhom-controller/internal/upwatch" "golang.org/x/crypto/bcrypt" ) @@ -804,6 +805,13 @@ func (s *Server) appCommonData(w http.ResponseWriter, r *http.Request, slug stri if lost, at := lostFilesNote(found.Deployed, found.AppConfig); len(lost) > 0 { data["LostFiles"], data["LostBootAt"] = lost, at } + // The upload watch (internal/upwatch): qBittorrent's limit, while the box owns it — one line the household can act on + // in qBittorrent itself. A limit the household set is theirs and is not described here. + if found.Meta.Slug == "qbittorrent" && found.Deployed && s.upwatchView != nil { + if v := s.upwatchView(); v.QBLimitBps > 0 { + data["QBLimitNote"] = s.msgLang(s.langFor(r), "upwatch.qb_limit_note", upwatch.MbpsText(v.QBLimitBps)) + } + } // v0.280.0 (decision 46): the setup gate's card, while the gate stands. data["SetupGateClosed"] = found.Deployed && found.AppConfig != nil && found.AppConfig.SetupGate.Closed() data["SetupGateHasProbe"] = found.Meta.SetupDoneProbe != nil && found.Meta.SetupDoneProbe.URL != "" @@ -2349,6 +2357,12 @@ func (s *Server) settingsBaseData(page, title string) map[string]interface{} { func (s *Server) systemPageData() map[string]interface{} { data := s.settingsBaseData("settings", "Beállítások") data["TitleKey"] = "page.title.settings" // i18n: the Hungarian title above is what hu renders + // The upload watch's measured line (a lower bound, learnt from the nightly off-site push); "" = still learning. + if s.upwatchView != nil { + if v := s.upwatchView(); v.LineKnown { + data["UplinkLine"] = upwatch.MbpsText(int64(v.LineMbps * 1e6 / 8)) + } + } data["GitRepoURL"] = s.cfg.Git.RepoURL data["GitSyncInterval"] = s.cfg.Git.SyncInterval data["BackupEnabled"] = s.cfg.Backup.Enabled diff --git a/controller/internal/web/i18n_cases_a_test.go b/controller/internal/web/i18n_cases_a_test.go index 1478482..4b0b1da 100644 --- a/controller/internal/web/i18n_cases_a_test.go +++ b/controller/internal/web/i18n_cases_a_test.go @@ -239,6 +239,7 @@ func i18nCasesA() []i18nCase { d["MemorySupported"], d["MemoryReachable"] = true, true d["MemoryAllocatedMB"], d["MemoryUsageMB"], d["MemoryMinMB"], d["MemoryMaxMB"] = 4096, 2048, 2048, 8192 d["PowerOffSupported"] = true // R-998: the power-off card with its button + d["UplinkLine"] = "47" // the upload watch's measured line (the other three cases: still learning) return d }}, {"settings_system_failed", "settings_system", func() map[string]interface{} { diff --git a/controller/internal/web/server.go b/controller/internal/web/server.go index 9fe4702..92e9f2d 100644 --- a/controller/internal/web/server.go +++ b/controller/internal/web/server.go @@ -36,9 +36,12 @@ import ( "gitea.dooplex.hu/admin/felhom-controller/internal/settings" "gitea.dooplex.hu/admin/felhom-controller/internal/stacks" "gitea.dooplex.hu/admin/felhom-controller/internal/system" + "gitea.dooplex.hu/admin/felhom-controller/internal/upwatch" ) type Server struct { + // upwatchView (SetUpwatchView) feeds the two upload-watch lines; nil = not shown. + upwatchView func() upwatch.View // R-871: the missed-backup banner's inputs (SetMissedBackupBanner); nil = no banner. nightLedger *nightchain.Ledger metricsRecord *metrics.MetricsStore @@ -515,6 +518,10 @@ func (s *Server) addShellAssetData(data map[string]interface{}) { // so a cache that already held the first build's stylesheet kept serving it — measured through // Cloudflare (`audits/app-page-2026-10-15b/P0/VERDICT.md`). build.sh now refuses a second push of a // version, and this makes the asset URL change even when the version does not. +// SetUpwatchView hands the pages the upload watch's state (internal/upwatch): the measured line and the qBittorrent +// limit the box set. Nil = neither line is shown. +func (s *Server) SetUpwatchView(fn func() upwatch.View) { s.upwatchView = fn } + func (s *Server) SetBuildIdentity(gitCommit, buildTime string) { s.buildID = gitCommit + "/" + buildTime } diff --git a/controller/internal/web/templates/app_shell.html b/controller/internal/web/templates/app_shell.html index 4d3b81a..e32e0ea 100644 --- a/controller/internal/web/templates/app_shell.html +++ b/controller/internal/web/templates/app_shell.html @@ -79,6 +79,9 @@ {{- end}} +{{- if .QBLimitNote}} +
{{.QBLimitNote}}
+{{- end}} {{if .Stack.Updating}}
{{updatePhaseText .Stack}}
{{else if and .Stack.HoldReason (eq .Stack.HoldKind "unhealthy_stop")}} diff --git a/controller/internal/web/templates/settings_system.html b/controller/internal/web/templates/settings_system.html index bb15d3f..3f4aaed 100644 --- a/controller/internal/web/templates/settings_system.html +++ b/controller/internal/web/templates/settings_system.html @@ -183,6 +183,10 @@ {{T "settings_system.atjaro"}} {{if .NetGateway}}{{.NetGateway}}{{else}}— {{T "settings_system.nem_allapithato_meg"}}{{end}} +
+ {{T "upwatch.line_label"}} + {{if .UplinkLine}}{{T "upwatch.line_at_least"}} {{.UplinkLine}} {{T "upwatch.line_measured"}}{{else}}{{T "upwatch.line_learning"}}{{end}} +

{{T "settings_system.ezek_az_ertekek_a_helyi"}}

diff --git a/controller/internal/web/testdata/i18n_parity/settings_system_failed.html b/controller/internal/web/testdata/i18n_parity/settings_system_failed.html index effcd98..3274ba4 100644 --- a/controller/internal/web/testdata/i18n_parity/settings_system_failed.html +++ b/controller/internal/web/testdata/i18n_parity/settings_system_failed.html @@ -303,6 +303,10 @@ Átjáró — nem állapítható meg +
+ Feltöltési sebesség + még mérjük (az éjszakai távoli mentésből) +

Ezek az értékek a helyi hálózattól függenek és változhatnak. Távoli hibaelhárításnál erről az oldalról olvasd fel őket.

diff --git a/controller/internal/web/testdata/i18n_parity/settings_system_full.html b/controller/internal/web/testdata/i18n_parity/settings_system_full.html index d31d7fe..968b98e 100644 --- a/controller/internal/web/testdata/i18n_parity/settings_system_full.html +++ b/controller/internal/web/testdata/i18n_parity/settings_system_full.html @@ -348,6 +348,10 @@ Átjáró 192.168.1.1 +
+ Feltöltési sebesség + legalább 47 Mbit/s (az éjszakai távoli mentésből mérve) +

Ezek az értékek a helyi hálózattól függenek és változhatnak. Távoli hibaelhárításnál erről az oldalról olvasd fel őket.

diff --git a/controller/internal/web/testdata/i18n_parity/settings_system_nomemory.html b/controller/internal/web/testdata/i18n_parity/settings_system_nomemory.html index 99a0d44..064778f 100644 --- a/controller/internal/web/testdata/i18n_parity/settings_system_nomemory.html +++ b/controller/internal/web/testdata/i18n_parity/settings_system_nomemory.html @@ -260,6 +260,10 @@ Átjáró — nem állapítható meg +
+ Feltöltési sebesség + még mérjük (az éjszakai távoli mentésből) +

Ezek az értékek a helyi hálózattól függenek és változhatnak. Távoli hibaelhárításnál erről az oldalról olvasd fel őket.

diff --git a/controller/internal/web/testdata/i18n_parity/settings_system_pending.html b/controller/internal/web/testdata/i18n_parity/settings_system_pending.html index 2a89997..b22a4bd 100644 --- a/controller/internal/web/testdata/i18n_parity/settings_system_pending.html +++ b/controller/internal/web/testdata/i18n_parity/settings_system_pending.html @@ -293,6 +293,10 @@ Átjáró — nem állapítható meg +
+ Feltöltési sebesség + még mérjük (az éjszakai távoli mentésből) +

Ezek az értékek a helyi hálózattól függenek és változhatnak. Távoli hibaelhárításnál erről az oldalról olvasd fel őket.

diff --git a/controller/internal/web/upwatch_lines_test.go b/controller/internal/web/upwatch_lines_test.go new file mode 100644 index 0000000..bdec806 --- /dev/null +++ b/controller/internal/web/upwatch_lines_test.go @@ -0,0 +1,55 @@ +package web + +import ( + "html" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "testing" + + "gitea.dooplex.hu/admin/felhom-controller/internal/upwatch" +) + +// The upload watch's household line: qBittorrent's page says the limit while the BOX owns it — in the reader's +// language — and says nothing when the household owns it (the view then carries 0), when no view is wired, or on +// another app's page. +// RED-PROOF: drop the `.QBLimitNote` card from app_shell.html → the first case fails; drop the slug check in the +// handler → crafty's page carries the line → fails. +func TestUpwatch_QBLimitLineOnQBittorrentsPageOnly(t *testing.T) { + s, _ := credsHarness(t) + sd := filepath.Join(s.cfg.Paths.StacksDir, "qbittorrent") + if err := os.MkdirAll(sd, 0o755); err != nil { + t.Fatal(err) + } + os.WriteFile(filepath.Join(sd, ".felhom.yml"), []byte("display_name: qBittorrent\nslug: qbittorrent\n"), 0o644) + os.WriteFile(filepath.Join(sd, "docker-compose.yml"), []byte("services: {}\n"), 0o644) + os.WriteFile(filepath.Join(sd, "app.yaml"), []byte("deployed: true\n"), 0o644) + _ = s.stackMgr.ScanStacks() + s.loadTemplates() + render := func(app, lang string) string { + rr := httptest.NewRecorder() + s.appPageHandler(rr, httptest.NewRequest("GET", "/apps/"+app+"?lang="+lang, nil), app, "") + return html.UnescapeString(rr.Body.String()) + } + view := upwatch.View{LineMbps: 100, LineKnown: true, QBLimitBps: 6249472} + s.SetUpwatchView(func() upwatch.View { return view }) + hu, en := asciiFold(render("qbittorrent", "hu")), render("qbittorrent", "en") + if !strings.Contains(hu, "data-qb-limit") || !strings.Contains(hu, "legfeljebb 50 Mbit/s") || !strings.Contains(hu, "atallithatod") { + t.Fatalf("the Hungarian qBittorrent page lacks the limit line") + } + if !strings.Contains(en, "uploads at most 50 Mbit/s") { + t.Fatalf("the English qBittorrent page lacks the limit line") + } + if strings.Contains(render("crafty", "hu"), "data-qb-limit") { + t.Fatal("another app's page carries qBittorrent's limit line") + } + view.QBLimitBps = 0 // the household owns it + if strings.Contains(render("qbittorrent", "hu"), "data-qb-limit") { + t.Fatal("a household-owned limit is described as the box's") + } + s.upwatchView = nil + if strings.Contains(render("qbittorrent", "hu"), "data-qb-limit") { + t.Fatal("no view wired, yet a line") + } +} diff --git a/controller/scripts/i18n_go_keys.json b/controller/scripts/i18n_go_keys.json index 28ca40b..c18e490 100644 --- a/controller/scripts/i18n_go_keys.json +++ b/controller/scripts/i18n_go_keys.json @@ -244,6 +244,12 @@ "poweroff.agent_down": "BORN AS A KEY (R-998, v0.310.0) -- „A gép kikapcsolása”: the household sentence for the power-off route's answer. Pinned by TestPowerOff_EveryReasonHasASentence.", "poweroff.agent_outdated": "BORN AS A KEY (R-998, v0.310.0) -- „A gép kikapcsolása”: the household sentence for the power-off route's answer. Pinned by TestPowerOff_HiddenForOldAgent.", "event.app_repaired_after_crash": "BORN AS A KEY (R-1007, v0.311.0) -- the operator's counts after an unclean boot (app_repaired_after_crash). Pinned by TestAppRepairedAfterCrash_KeysInBothLanguages.", + "event.app_unexplained_upload": "BORN AS A KEY (upload watch, 2026-10-23 night) -- operator-only event of internal/upwatch (felhom.eu 08 \u00a79). Pinned by TestNotifyUnexplained_TypesSentencesAndDetails.", + "event.app_unexplained_upload_noapp": "BORN AS A KEY (upload watch, 2026-10-23 night) -- operator-only event of internal/upwatch (felhom.eu 08 \u00a79). Pinned by TestNotifyUnexplained_TypesSentencesAndDetails.", + "event.app_unexplained_upload_cleared": "BORN AS A KEY (upload watch, 2026-10-23 night) -- operator-only event of internal/upwatch (felhom.eu 08 \u00a79). Pinned by TestNotifyUnexplained_TypesSentencesAndDetails.", + "event.app_unexplained_cpu": "BORN AS A KEY (upload watch, 2026-10-23 night) -- operator-only event of internal/upwatch (felhom.eu 08 \u00a79). Pinned by TestNotifyUnexplained_TypesSentencesAndDetails.", + "event.app_unexplained_cpu_cleared": "BORN AS A KEY (upload watch, 2026-10-23 night) -- operator-only event of internal/upwatch (felhom.eu 08 \u00a79). Pinned by TestNotifyUnexplained_TypesSentencesAndDetails.", + "upwatch.qb_limit_note": "BORN AS A KEY (upload watch, 2026-10-23 night) -- the household line on qBittorrent's page while the box owns its upload limit. Pinned by TestUpwatch_QBLimitLineOnQBittorrentsPageOnly.", "event.app_repair_after_crash_skipped": "BORN AS A KEY (R-1007, v0.311.0) -- the same event when the run was skipped, with its reason. Pinned by TestAppRepairedAfterCrash_KeysInBothLanguages.", "err.lost_files.nothing_to_dismiss": "BORN AS A KEY (R-1007, v0.311.0) -- the dismiss route's answer when there is no lost-files note. Pinned by TestAppRepairedAfterCrash_KeysInBothLanguages.", "poweroff.refused.too_soon": "BORN AS A KEY (R-998, v0.310.0) -- „A gép kikapcsolása”: the household sentence for the power-off route's answer. Pinned by TestPowerOff_EveryReasonHasASentence.",