hub v0.25.0: per-storage worst-fill alerting (StorageFillChecker)

Generalizes host_disk to any reported storage target (dump/backup volume, data drive,
thin pool, PBS). Per-(host,target) state, born/persistent, natural critical severity,
distinct storage_fill_* events; excludes the root-backed builtin (host_disk owns root).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HxLA1mZurFq9kt8hneFeCs

hub/CHANGELOG.md

@@ -1,5 +1,32 @@
 # Felhom Hub — Changelog
 
+## v0.25.0 — per-storage worst-fill alerting (StorageFillChecker) (2026-06-30)
+
+Generalizes the host-root disk alert (v0.23.0) to ANY reported storage target — so a dedicated
+dump/backup volume, data drive, lvmthin pool, or PBS datastore filling toward failure pages the operator
+with the storage named, even when host root itself is fine.
+
+- **`internal/monitor/storage_fill.go` (NEW) — `StorageFillChecker`.** A per-target mirror of
+  `HostDiskChecker` on the same 60s sweep: born/persistent (already-breached `(host,target)` keys left
+  UNSEEDED → first `Check` emits), escalation-only emit, recovery re-arm, the dispatcher's 1h cooldown.
+  State is keyed per **(host, target)** so targets alert independently. Emits distinct
+  `storage_fill_warning` / `storage_fill_critical` at the **natural `critical` severity** (exercises the
+  v0.24.0 dispatcher fix with a second real caller). Default thresholds 90/95, hub-config overridable
+  (`alerting.storage_fill_warn_percent` / `_crit_percent`), independent of the host-root thresholds.
+- **Root excluded (no double-alert):** the host root-backed builtin (`Type=="local"`, or a target mounted
+  at `/`) is skipped — `HostDiskChecker` owns root. So a root-backed vzdump dump is one alert (from
+  host_disk), and storage_fill uniquely covers OFF-root storage.
+- **`internal/store/store.go`:** `GetHostStorageTargets()` + `HostStorageTargetRow` — parses
+  `report_json.storage_targets[]` of each host's latest report (percent = `used_fraction`×100); modeled on
+  `GetHostDiskUsage`, no denorm column / migration.
+- **`internal/notify/templates.go` + `internal/api/handler.go`:** Hungarian templates + allowlist entries
+  for `storage_fill_warning` / `storage_fill_critical`.
+- **`cmd/hub/main.go`:** register `storageFillChecker` on the 60s tick beside `HostDiskChecker`.
+- Tests: per-target bands (independent warn/escalate/recover/re-arm), **born/persistent companion red-proof**
+  (a seed-all model stays silent on the born-breach), **root-exclusion companion** (without the exclusion a
+  root target IS in the critical band — the exclusion is what suppresses the double-alert), severity
+  `critical`, and the store parse. `go build/vet/test ./...` green.
+
 ## v0.24.0 — dispatcher routes `critical` severity (+ nil-prefs crash guard) (2026-06-30)
 
 NAS Part A2's "Part 0": close the dispatcher's silent drop of `critical`-severity events.

hub/cmd/hub/main.go

@@ -51,6 +51,10 @@ type Config struct {
 		// (normalizeDiskThresholds). Seed-only config, like stale_threshold.
 		HostDiskWarnPercent float64 `yaml:"host_disk_warn_percent"`
 		HostDiskCritPercent float64 `yaml:"host_disk_crit_percent"`
+		// Per-storage-target fill thresholds (percent). Empty/0/invalid → defaults 90/95. Independent of
+		// the host-root thresholds above so a single storage's policy can be tuned separately.
+		StorageFillWarnPercent float64 `yaml:"storage_fill_warn_percent"`
+		StorageFillCritPercent float64 `yaml:"storage_fill_crit_percent"`
 	} `yaml:"alerting"`
 	Registry struct {
 		Image            string `yaml:"image"`
@@ -330,6 +334,10 @@ func main() {
 	// root disk_percent the agent reports; born/persistent (a disk already full at hub restart alerts on
 	// cycle 1). Distinct event types from the controller's GUEST disk_warning/disk_critical. Same 60s sweep.
 	hostDiskChecker := monitor.NewHostDiskChecker(dataStore, cfg.Alerting.HostDiskWarnPercent, cfg.Alerting.HostDiskCritPercent, dispatcher.ProcessEvent, logger)
+	// v0.25.0: PER-STORAGE worst-fill alert — generalizes the host-root check to any reported storage
+	// target (dump/backup volume, data drive, lvmthin pool, PBS datastore). Born/persistent; excludes the
+	// root-backed builtin (hostDiskChecker owns root, no double-alert); emits natural `critical`. Same sweep.
+	storageFillChecker := monitor.NewStorageFillChecker(dataStore, cfg.Alerting.StorageFillWarnPercent, cfg.Alerting.StorageFillCritPercent, dispatcher.ProcessEvent, logger)
 	go func() {
 		ticker := time.NewTicker(60 * time.Second)
 		defer ticker.Stop()
@@ -343,6 +351,7 @@ func main() {
 				hostCapabilityChecker.Check()
 				hostLeafChecker.Check()
 				hostDiskChecker.Check()
+				storageFillChecker.Check()
 			}
 		}
 	}()

hub/internal/api/handler.go

@@ -1098,6 +1098,8 @@ var allowedEventTypes = map[string]bool{
 	// Hub-generated host root-fs disk-pressure (v0.23.0) — distinct from the controller's GUEST disk_*
 	"host_disk_warning":      true,
 	"host_disk_critical":     true,
+	"storage_fill_warning":   true,
+	"storage_fill_critical":  true,
 	"expected_backup_missed": true,
 	"expected_dbdump_missed": true,
 	// Special

hub/internal/monitor/storage_fill.go

@@ -0,0 +1,201 @@
+package monitor
+
+import (
+	"encoding/json"
+	"fmt"
+	"log"
+	"strings"
+	"sync"
+
+	"gitea.dooplex.hu/admin/felhom-hub/internal/store"
+)
+
+// StorageFillChecker generalizes HostDiskChecker from "host root" to "any storage target": it alerts when
+// a SPECIFIC reported storage target (a vzdump/backup volume, a data drive, an lvmthin pool, a PBS
+// datastore, …) crosses a warning/critical fill threshold — naming the target. The felhom-pve incident's
+// root-backed dump volume is already covered by HostDiskChecker (root fills); this adds the case the
+// host-root signal MISSES: a dedicated backup/data storage on its OWN disk filling while root stays fine.
+//
+// It mirrors HostDiskChecker exactly (born/persistent, escalation-only emit, recovery re-arm, the
+// dispatcher's 1h cooldown) but keys state per (host, target) and emits distinct storage_fill_* events at
+// the NATURAL `critical` severity (hub v0.24.0 routes it). It EXCLUDES the host root-backed builtin
+// storage so it never double-alerts what HostDiskChecker already owns.
+type StorageFillChecker struct {
+	store   *store.Store
+	warn    float64
+	crit    float64
+	logger  *log.Logger
+	onEvent EventNotifyFunc
+
+	mu         sync.Mutex
+	states     map[string]string // key(host,target) → band; a breached key is left UNSEEDED at init (F2)
+	customerOf map[string]string // hostID → customerID
+}
+
+const (
+	defaultStorageFillWarnPercent = 90.0
+	defaultStorageFillCritPercent = 95.0
+)
+
+// fillKey is the per-(host,target) state key. A NUL separator can't appear in a host id / storage name.
+func fillKey(hostID, target string) string { return hostID + "\x00" + target }
+
+// excludeFromStorageFill reports whether a target is the host root-backed builtin (HostDiskChecker owns
+// it — alerting here too would double-page for the same filesystem). The PVE builtin "local" lives on the
+// root fs (its used_fraction == root fill); a target mounted at "/" is the root itself.
+func excludeFromStorageFill(typ, mountPath string) bool {
+	return typ == "local" || cleanFillPath(mountPath) == "/"
+}
+
+func cleanFillPath(p string) string {
+	if p == "" {
+		return "" // root-backed dirs report "" — excluded via Type=="local", not here
+	}
+	trimmed := strings.TrimRight(p, "/")
+	if trimmed == "" {
+		return "/" // p was "/" (or "///") — the literal root mount
+	}
+	return trimmed
+}
+
+// NewStorageFillChecker creates the checker (defaults 90/95 when unset/invalid) and seeds state from the
+// latest reports. NO events on init except that already-breached (host,target) keys are left UNSEEDED so
+// their first Check emits (born/persistent — the F2 lesson).
+func NewStorageFillChecker(s *store.Store, warnPercent, critPercent float64, onEvent EventNotifyFunc, logger *log.Logger) *StorageFillChecker {
+	warn, crit := normalizeDiskThresholds(warnPercent, critPercent) // reuse host_disk's sane-defaults guard
+	fc := &StorageFillChecker{
+		store:      s,
+		warn:       warn,
+		crit:       crit,
+		logger:     logger,
+		onEvent:    onEvent,
+		states:     make(map[string]string),
+		customerOf: make(map[string]string),
+	}
+	rows, err := s.GetHostStorageTargets()
+	if err != nil {
+		logger.Printf("[WARN] Storage fill checker: failed to seed states: %v", err)
+		return fc
+	}
+	var okCount, breachedCount, excluded int
+	for _, row := range rows {
+		if s.IsCustomerBlocked(row.CustomerID) {
+			continue
+		}
+		if excludeFromStorageFill(row.Type, row.MountPath) {
+			excluded++
+			continue
+		}
+		fc.customerOf[row.HostID] = row.CustomerID
+		band := bandForPercent(row.Percent, fc.warn, fc.crit)
+		if band != bandOK {
+			breachedCount++
+			continue // leave UNSEEDED → first Check emits (the dispatcher's 1h cooldown dedups a restart)
+		}
+		fc.states[fillKey(row.HostID, row.Name)] = bandOK
+		okCount++
+	}
+	logger.Printf("[INFO] Storage fill checker initialized: warn=%.0f%% crit=%.0f%%, %d ok seeded, %d already-breached left unseeded, %d root-backed excluded", warn, crit, okCount, breachedCount, excluded)
+	return fc
+}
+
+// Check evaluates every (host, non-root target) and emits on each escalation (incl. the born-breach).
+// De-escalation/recovery re-arm silently. Same 60s sweep as the other host checkers.
+func (fc *StorageFillChecker) Check() {
+	rows, err := fc.store.GetHostStorageTargets()
+	if err != nil {
+		fc.logger.Printf("[WARN] Storage fill check failed: %v", err)
+		return
+	}
+	fc.mu.Lock()
+	defer fc.mu.Unlock()
+
+	seen := make(map[string]bool, len(rows))
+	for _, row := range rows {
+		if excludeFromStorageFill(row.Type, row.MountPath) {
+			continue // root-backed builtin — HostDiskChecker owns it
+		}
+		key := fillKey(row.HostID, row.Name)
+		seen[key] = true
+		if fc.store.IsCustomerBlocked(row.CustomerID) {
+			delete(fc.states, key)
+			continue
+		}
+		fc.customerOf[row.HostID] = row.CustomerID
+
+		newBand := bandForPercent(row.Percent, fc.warn, fc.crit)
+		oldBand := fc.states[key] // "" (rank 0) for an unseen / breached-at-init key
+		if bandRank(newBand) > bandRank(oldBand) {
+			fc.emit(row, oldBand, newBand)
+		}
+		fc.states[key] = newBand
+	}
+
+	// Drop state for targets that vanished from the latest report (storage removed) so a later re-add
+	// re-arms cleanly.
+	for k := range fc.states {
+		if !seen[k] {
+			delete(fc.states, k)
+		}
+	}
+}
+
+// GetState returns the current band for a (host, target) ("unknown" if unseen). For tests.
+func (fc *StorageFillChecker) GetState(hostID, target string) string {
+	fc.mu.Lock()
+	defer fc.mu.Unlock()
+	s := fc.states[fillKey(hostID, target)]
+	if s == "" {
+		return "unknown"
+	}
+	return s
+}
+
+func (fc *StorageFillChecker) emit(row store.HostStorageTargetRow, oldBand, newBand string) {
+	var eventType, severity, message string
+	switch newBand {
+	case bandCritical:
+		eventType = "storage_fill_critical"
+		severity = "critical" // natural critical — hub v0.24.0 routes it; the operator email styles it 🔴
+		message = fmt.Sprintf("Host %s: storage %q CRITICALLY full at %.0f%% (threshold %.0f%%) — backups/writes to it will fail; free space immediately", row.HostID, row.Name, row.Percent, fc.crit)
+	case bandWarning:
+		eventType = "storage_fill_warning"
+		severity = "warning"
+		message = fmt.Sprintf("Host %s: storage %q high at %.0f%% (threshold %.0f%%) — free space before it fills", row.HostID, row.Name, row.Percent, fc.warn)
+	default:
+		return
+	}
+
+	details, _ := json.Marshal(map[string]any{
+		"host_id":      row.HostID,
+		"storage":      row.Name,
+		"storage_type": row.Type,
+		"percent":      row.Percent,
+		"total_bytes":  row.TotalBytes,
+		"used_bytes":   row.UsedBytes,
+		"warn_percent": fc.warn,
+		"crit_percent": fc.crit,
+	})
+
+	fc.logger.Printf("[INFO] Storage fill: %s %q %.0f%% %s→%s (%s)", row.HostID, row.Name, row.Percent, bandLabel(oldBand), newBand, eventType)
+
+	if _, err := fc.store.SaveEvent(row.CustomerID, eventType, severity, message, string(details), "hub"); err != nil {
+		fc.logger.Printf("[WARN] Failed to save storage fill event for %s/%s: %v", row.HostID, row.Name, err)
+		return
+	}
+	if fc.onEvent != nil {
+		fc.onEvent(row.CustomerID, eventType, severity, message, string(details), "hub")
+	}
+}
+
+// bandForPercent maps a fill percentage to its band (free function so both checkers share the bands).
+func bandForPercent(pct, warn, crit float64) string {
+	switch {
+	case pct >= crit:
+		return bandCritical
+	case pct >= warn:
+		return bandWarning
+	default:
+		return bandOK
+	}
+}

hub/internal/monitor/storage_fill_test.go

@@ -0,0 +1,147 @@
+package monitor
+
+import (
+	"fmt"
+	"strings"
+	"testing"
+
+	"gitea.dooplex.hu/admin/felhom-hub/internal/store"
+)
+
+// stTarget is a storage-target fixture for saveStorageReport.
+type stTarget struct {
+	name, typ, mount string
+	pct              float64
+}
+
+// saveStorageReport records a host-report whose report_json carries the given storage_targets[].
+func saveStorageReport(t *testing.T, st *store.Store, targets ...stTarget) {
+	t.Helper()
+	var parts []string
+	for _, tg := range targets {
+		parts = append(parts, fmt.Sprintf(
+			`{"name":%q,"type":%q,"mount_path":%q,"used_fraction":%g,"total_bytes":1000000000,"used_bytes":%d}`,
+			tg.name, tg.typ, tg.mount, tg.pct/100, int64(tg.pct*1e7)))
+	}
+	body := `{"host_id":"h1","storage_targets":[` + strings.Join(parts, ",") + `]}`
+	if err := st.SaveHostReport("h1", "c1", []byte(body), store.HostReportDenorm{}); err != nil {
+		t.Fatal(err)
+	}
+}
+
+// TestStorageFill_PerTargetBands: state is keyed per (host, target). A breaches → only A alerts; B silent.
+// A escalates warning→critical; A recovers → clears + re-arms; B never fires (§7-A/C).
+func TestStorageFill_PerTargetBands(t *testing.T) {
+	st := newDiskStore(t)
+	saveStorageReport(t, st, stTarget{"dumpvol", "local-dir", "/mnt/backup", 50}, stTarget{"datadrive", "usb", "/mnt/data", 50})
+	var events, types []string
+	onEvent := func(_, et, sev, _, _, _ string) { events = append(events, sev); types = append(types, et) }
+	fc := NewStorageFillChecker(st, 90, 95, onEvent, quietLog())
+
+	// A → 92 warning; B stays 50.
+	saveStorageReport(t, st, stTarget{"dumpvol", "local-dir", "/mnt/backup", 92}, stTarget{"datadrive", "usb", "/mnt/data", 50})
+	fc.Check()
+	if fc.GetState("h1", "dumpvol") != bandWarning || fc.GetState("h1", "datadrive") != bandOK {
+		t.Fatalf("states: dumpvol=%s datadrive=%s", fc.GetState("h1", "dumpvol"), fc.GetState("h1", "datadrive"))
+	}
+	if len(types) != 1 || types[0] != "storage_fill_warning" {
+		t.Fatalf("want one storage_fill_warning, got %v", types)
+	}
+
+	// A → 96 critical (escalation); B still 50 (silent).
+	saveStorageReport(t, st, stTarget{"dumpvol", "local-dir", "/mnt/backup", 96}, stTarget{"datadrive", "usb", "/mnt/data", 50})
+	fc.Check()
+	if len(types) != 2 || types[1] != "storage_fill_critical" || events[1] != "critical" {
+		t.Fatalf("want storage_fill_critical (severity critical), got types=%v sev=%v", types, events)
+	}
+
+	// A recovers → clears + re-arms; no event.
+	saveStorageReport(t, st, stTarget{"dumpvol", "local-dir", "/mnt/backup", 40}, stTarget{"datadrive", "usb", "/mnt/data", 50})
+	fc.Check()
+	if fc.GetState("h1", "dumpvol") != bandOK || len(types) != 2 {
+		t.Fatalf("recovery should clear without event; state=%s events=%v", fc.GetState("h1", "dumpvol"), types)
+	}
+	// Re-armed: a fresh breach alerts again.
+	saveStorageReport(t, st, stTarget{"dumpvol", "local-dir", "/mnt/backup", 93}, stTarget{"datadrive", "usb", "/mnt/data", 50})
+	fc.Check()
+	if len(types) != 3 || types[2] != "storage_fill_warning" {
+		t.Fatalf("re-arm: a new breach must alert again, got %v", types)
+	}
+}
+
+// TestStorageFill_BornPersistent (F2): a target already over critical at init emits on the FIRST Check.
+// Companion: a seed-all (transition-only) model — emulated by pre-seeding the key to critical — stays silent.
+func TestStorageFill_BornPersistent(t *testing.T) {
+	st := newDiskStore(t)
+	saveStorageReport(t, st, stTarget{"dumpvol", "local-dir", "/mnt/backup", 97})
+	var types, sev []string
+	fc := NewStorageFillChecker(st, 90, 95, func(_, et, s, _, _, _ string) { types = append(types, et); sev = append(sev, s) }, quietLog())
+
+	if fc.GetState("h1", "dumpvol") != "unknown" {
+		t.Fatalf("an already-breached target must be left unseeded, got %s", fc.GetState("h1", "dumpvol"))
+	}
+	fc.Check()
+	if len(types) != 1 || types[0] != "storage_fill_critical" || sev[0] != "critical" {
+		t.Fatalf("born-persistent: first Check must emit storage_fill_critical/critical, got %v %v", types, sev)
+	}
+
+	// COMPANION: a seed-all impl seeds the breached key at init → first Check sees no transition → SILENT.
+	st2 := newDiskStore(t)
+	saveStorageReport(t, st2, stTarget{"dumpvol", "local-dir", "/mnt/backup", 97})
+	var silent []string
+	fc2 := NewStorageFillChecker(st2, 90, 95, func(_, et, _, _, _, _ string) { silent = append(silent, et) }, quietLog())
+	fc2.mu.Lock()
+	fc2.states[fillKey("h1", "dumpvol")] = bandCritical // the WRONG seed-all design's seed
+	fc2.mu.Unlock()
+	fc2.Check()
+	if len(silent) != 0 {
+		t.Fatalf("control: a seed-all model should be silent on the born-breach, got %v", silent)
+	}
+}
+
+// TestStorageFill_RootExcluded (§7-D): a root-backed target at 96% must NOT alert here (HostDiskChecker
+// owns root). Companion: prove the exclusion is what suppresses it (without it the band would be critical).
+func TestStorageFill_RootExcluded(t *testing.T) {
+	st := newDiskStore(t)
+	saveStorageReport(t, st,
+		stTarget{"local", "local", "", 96},       // PVE builtin local = root-backed → excluded by Type
+		stTarget{"rootmnt", "local-dir", "/", 96}, // a target literally at "/" → excluded by mount path
+	)
+	var types []string
+	fc := NewStorageFillChecker(st, 90, 95, func(_, et, _, _, _, _ string) { types = append(types, et) }, quietLog())
+	fc.Check()
+	if len(types) != 0 {
+		t.Fatalf("root-backed targets must NOT alert here (host_disk owns root), got %v", types)
+	}
+	// COMPANION: the exclusion is load-bearing — without it both targets are in the critical band.
+	if !excludeFromStorageFill("local", "") || !excludeFromStorageFill("local-dir", "/") {
+		t.Fatal("root-backed targets must be excluded")
+	}
+	if excludeFromStorageFill("local-dir", "/mnt/backup") {
+		t.Fatal("a real off-root storage must NOT be excluded")
+	}
+	if bandForPercent(96, 90, 95) != bandCritical {
+		t.Fatal("control: 96%% IS the critical band — only the exclusion suppresses the root alert")
+	}
+}
+
+// TestStorageFill_ParseAndExcludeFromStore checks GetHostStorageTargets parses names/percent + the checker
+// keys per target end-to-end (one report → two off-root targets tracked).
+func TestStorageFill_ParseFromStore(t *testing.T) {
+	st := newDiskStore(t)
+	saveStorageReport(t, st, stTarget{"backupdrive", "usb", "/mnt/bk", 80}, stTarget{"local", "local", "", 99})
+	rows, err := st.GetHostStorageTargets()
+	if err != nil {
+		t.Fatal(err)
+	}
+	got := map[string]float64{}
+	for _, r := range rows {
+		got[r.Name] = r.Percent
+	}
+	if v := got["backupdrive"]; v < 79.9 || v > 80.1 {
+		t.Fatalf("backupdrive percent = %v, want ~80", v)
+	}
+	if _, ok := got["local"]; !ok {
+		t.Fatal("parse should include all targets (the checker excludes root, not the store)")
+	}
+}

hub/internal/notify/templates.go

@@ -69,6 +69,10 @@ var customerMessages = map[string]string{
 	"host_disk_warning":  "A házszerver alaprendszerének (Proxmox-gazda) gyökérlemeze 90% felett van — kérjük, szabadíts fel helyet (pl. régi biztonsági mentések).",
 	"host_disk_critical": "A házszerver alaprendszerének gyökérlemeze kritikusan tele van (95%+) — azonnali beavatkozás szükséges, különben a rendszer hibázhat!",
 
+	// Per-storage fill events (a SPECIFIC tároló — mentési kötet, adatmeghajtó, tároló-készlet — telik be)
+	"storage_fill_warning":  "Egy tároló a házszerveren 90% felett telt — kérjük, szabadíts fel helyet, mielőtt megtelik.",
+	"storage_fill_critical": "Egy tároló a házszerveren kritikusan tele van (95%+) — a rá készülő mentések/írások meghiúsulhatnak; azonnali beavatkozás szükséges!",
+
 	// Storage events
 	"storage_disconnected": "Egy meghajtó leválasztva — a mentések szünetelhetnek.",
 	"storage_reconnected":  "A meghajtó újra csatlakoztatva.",

hub/internal/store/store.go

@@ -1731,6 +1731,61 @@ func (s *Store) GetHostDiskUsage() ([]HostDiskRow, error) {
 	return out, rows.Err()
 }
 
+// HostStorageTargetRow is one (host, storage target) fill observation, parsed from the latest report's
+// storage_targets[] (no denorm column — modeled on GetHostDiskUsage's report_json parse). Percent is
+// used_fraction×100 (the agent reports a 0..1 fraction). A target with no usable fraction yields 0 (ok).
+type HostStorageTargetRow struct {
+	HostID     string
+	CustomerID string
+	Name       string  // Proxmox storage id
+	Type       string  // local | local-dir | lvmthin | usb | nfs | cifs | pbs
+	MountPath  string  // host mountpoint ("" for network/lvm/root-backed dir)
+	Percent    float64 // 0..100 (used_fraction × 100)
+	TotalBytes int64
+	UsedBytes  int64
+}
+
+// GetHostStorageTargets returns the per-storage fill of every host's LATEST report (MAX(id) per host),
+// parsed from report_json.storage_targets[] (mirrors GetHostDiskUsage). The StorageFillChecker keys on
+// (host, name) and excludes the root-backed builtin — see monitor/storage_fill.go.
+func (s *Store) GetHostStorageTargets() ([]HostStorageTargetRow, error) {
+	rows, err := s.db.Query(`
+		SELECT hr.host_id, hr.customer_id, hr.report_json
+		FROM host_reports hr
+		JOIN (SELECT host_id, MAX(id) AS mx FROM host_reports GROUP BY host_id) latest
+		  ON hr.id = latest.mx`)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var out []HostStorageTargetRow
+	for rows.Next() {
+		var hostID, customerID, reportJSON string
+		if err := rows.Scan(&hostID, &customerID, &reportJSON); err != nil {
+			return nil, err
+		}
+		var body struct {
+			StorageTargets []struct {
+				Name         string  `json:"name"`
+				Type         string  `json:"type"`
+				MountPath    string  `json:"mount_path"`
+				UsedFraction float64 `json:"used_fraction"`
+				TotalBytes   int64   `json:"total_bytes"`
+				UsedBytes    int64   `json:"used_bytes"`
+			} `json:"storage_targets"`
+		}
+		_ = json.Unmarshal([]byte(reportJSON), &body) // malformed/old body → no targets (never a false alert)
+		for _, t := range body.StorageTargets {
+			out = append(out, HostStorageTargetRow{
+				HostID: hostID, CustomerID: customerID,
+				Name: t.Name, Type: t.Type, MountPath: t.MountPath,
+				Percent: t.UsedFraction * 100, TotalBytes: t.TotalBytes, UsedBytes: t.UsedBytes,
+			})
+		}
+	}
+	return out, rows.Err()
+}
+
 // GetHostLeafFingerprints returns the latest reported local-API leaf fp per host (mirrors
 // GetHostCapabilities — MAX(id) per host, parsed from report_json so there is no schema migration).
 func (s *Store) GetHostLeafFingerprints() ([]HostLeafRow, error) {