felhom-agent/internal/storage/observe.go

package storage

import (
	"context"
	"fmt"
	"log/slog"
	"regexp"
	"strings"

	"gitea.dooplex.hu/admin/felhom-agent/internal/hub"
	"gitea.dooplex.hu/admin/felhom-agent/internal/proxmox"
)

// thinPoolWarnFraction is the lvmthin DATA-fill level above which the observer logs a
// prominent warning. A full thin-pool corrupts EVERY guest on it (the storage analog of a
// single-node OOM), so it must be visible early — well before slice-10 policy exists.
const thinPoolWarnFraction = 0.85

// StorageAPI is the read-only Proxmox surface the observer needs. *proxmox.Client
// satisfies it. ListStorage (cluster) carries the type-specific config (server/export/
// vgname/thinpool/fingerprint) that NodeStorage may omit; NodeStorage carries live usage
// + the per-node active flag. The observer joins them by storage name.
type StorageAPI interface {
	Node() string
	ListStorage(ctx context.Context) ([]proxmox.Storage, error)
	NodeStorage(ctx context.Context) ([]proxmox.Storage, error)
}

// Observer builds the observed storage view from Proxmox + non-privileged host reads.
// In Phase B it also (optionally) enriches the reported view with the privileged reads —
// SMART + thin-pool metadata — via HostOps; a nil ops keeps the Phase-A behaviour
// (SMART UNKNOWN, metadata null).
type Observer struct {
	api    StorageAPI
	host   HostReader
	ops    HostOps
	logger *slog.Logger
}

// NewObserver builds an Observer. host defaults to a ProcHostReader; logger to the
// default. ops is the privileged surface for SMART/lvs — nil disables those (Phase-A
// behaviour). A nil api makes Observe/Known return an error (misconfiguration), never panic.
func NewObserver(api StorageAPI, host HostReader, ops HostOps, logger *slog.Logger) *Observer {
	if host == nil {
		host = NewProcHostReader()
	}
	if logger == nil {
		logger = slog.Default()
	}
	return &Observer{api: api, host: host, ops: ops, logger: logger}
}

// observed is the rich internal view of one target, from which both the reported
// hub.StorageTarget and the watchdog's KnownTarget are projected. src/cat are kept for
// Observe-time privileged enrichment (NOT used by the watchdog's Known path).
type observed struct {
	target hub.StorageTarget
	known  KnownTarget
	src    proxmox.Storage
	cat    storageCategory
	// smartHint (v0.95.0) is a SMART-ONLY whole-disk device for a dir-storage whose own backing is
	// empty (the builtin `local` on the shared LVM root). Never assigned to BackingDevice/durable_id.
	smartHint string
}

// Observe builds the reported []hub.StorageTarget. A non-nil error means the Proxmox read
// failed (the collector then omits storage from this cycle's report but still sends the
// rest). The returned slice is always non-nil so it marshals as [] when empty.
func (o *Observer) Observe(ctx context.Context) ([]hub.StorageTarget, error) {
	snap, err := o.snapshot(ctx)
	if err != nil {
		return nil, err
	}
	out := make([]hub.StorageTarget, 0, len(snap))
	for _, s := range snap {
		out = append(out, o.enrich(ctx, s))
	}
	return out, nil
}

// enrich adds the PRIVILEGED reads (SMART, thin-pool metadata) on top of the base target.
// Only Observe calls this (the watchdog's Known path skips it — these are the slow,
// root-shelling reads). A nil ops or a per-target failure degrades gracefully: SMART stays
// UNKNOWN, metadata stays null.
func (o *Observer) enrich(ctx context.Context, ob observed) hub.StorageTarget {
	t := ob.target
	if o.ops == nil {
		return t
	}
	// SMART: for dir-backed targets. The device is the target's own backing (a USB/local-dir exact
	// mount) OR, for a dir on a shared filesystem whose backing is deliberately empty (the builtin
	// `local` on the LVM root — removable-safety guard in build), the SMART-only hint build() resolved
	// from the containing filesystem. smartDeviceFor then resolves dm/LVM/partition to the whole disk.
	if ob.cat == catDir {
		smartDev := t.BackingDevice
		if smartDev == "" {
			smartDev = ob.smartHint
		}
		if smartDev != "" {
			if dev, ok := smartDeviceFor(smartDev); ok {
				if sm, err := o.ops.SMART(ctx, dev); err != nil {
					o.logger.Warn("storage: SMART read failed; health UNKNOWN", "device", dev, "err", err)
				} else {
					t.Smart = sm
				}
			}
		}
	}
	// Thin-pool metadata fill (the value Phase A left null): lvs on the vg/pool.
	if t.Type == hub.StorageTypeLVMThin && t.ThinPool != nil && ob.src.VGName != "" && ob.src.ThinPool != "" {
		if frac, ok := o.ops.ThinPoolMetadata(ctx, ob.src.VGName, ob.src.ThinPool); ok {
			t.ThinPool.MetadataUsedFraction = &frac
			if frac >= thinPoolWarnFraction {
				o.logger.Warn("storage: lvmthin pool METADATA fill is high (exhaustion corrupts the pool like data exhaustion)",
					"storage", t.Name, "metadata_used_fraction", frac)
			}
		}
	}
	return t
}

// Known projects the snapshot to the watchdog's lightweight KnownTarget set. Same Proxmox
// + host reads as Observe — callers that poll it fast should wrap it in a cache (the
// watchdog uses CachingKnownTargets).
func (o *Observer) Known(ctx context.Context) ([]KnownTarget, error) {
	snap, err := o.snapshot(ctx)
	if err != nil {
		return nil, err
	}
	out := make([]KnownTarget, 0, len(snap))
	for _, s := range snap {
		out = append(out, s.known)
	}
	return out, nil
}

// snapshot does the full build: join cluster config + node usage, then derive each
// target's identity, state, class hint, and (for lvmthin) thin-pool fill from host reads.
func (o *Observer) snapshot(ctx context.Context) ([]observed, error) {
	if o.api == nil {
		return nil, fmt.Errorf("storage: no proxmox api configured")
	}
	cluster, err := o.api.ListStorage(ctx)
	if err != nil {
		return nil, fmt.Errorf("storage: ListStorage: %w", err)
	}
	cfgByName := make(map[string]proxmox.Storage, len(cluster))
	for _, c := range cluster {
		cfgByName[c.Storage] = c
	}
	nodeStores, err := o.api.NodeStorage(ctx)
	if err != nil {
		return nil, fmt.Errorf("storage: NodeStorage: %w", err)
	}
	mounts, err := o.host.Mounts()
	if err != nil {
		// Host mount read failed: degrade rather than fail the whole report — Proxmox
		// usage/active is still meaningful; we just lose mount-derived fields.
		o.logger.Warn("storage: reading mounts failed; mount/device fields degraded", "err", err)
		mounts = nil
	}

	out := make([]observed, 0, len(nodeStores))
	for _, ns := range nodeStores {
		// Overlay the cluster config (server/export/vgname/...) onto the node entry.
		s := mergeConfig(ns, cfgByName[ns.Storage])
		out = append(out, o.build(s, mounts))
	}
	return out, nil
}

// build derives one observed target from a merged Storage entry + the mount table.
func (o *Observer) build(s proxmox.Storage, mounts []Mount) observed {
	category := categorize(s.Type)

	// Resolve the backing device + mount path for dir-like targets — ONLY from the
	// target's OWN mountpoint. We deliberately do NOT fall through to the containing
	// filesystem (e.g. root): an unmounted removable dir-storage's mountpoint reverts to a
	// bare directory on root, and resolving its UUID/durable_id to ROOT's UUID would be a
	// catastrophic DR mis-id (the hub would re-attach the wrong disk). When the target is
	// not its own mount, we leave the device/UUID unknown and the durable_id falls back to a
	// stable store id (never another fs's UUID). The authoritative UUID for re-attach comes
	// from a prior attached observation (watchdog memory) or the hub manifest (slice 10).
	var backingDevice, mountPath string
	var exactMount bool
	if category == catDir {
		if dev, mp, ok := exactMountDevice(mounts, s.Path); ok {
			backingDevice, mountPath, exactMount = dev, mp, true
		}
	}

	// Type: distinguish builtin local / removable USB / fixed local-dir within "dir".
	removable, removableKnown := false, false
	if category == catDir && backingDevice != "" {
		removable, removableKnown = o.host.Removable(backingDevice)
	}
	typ := reportType(s, category, removable, removableKnown)

	// durable_id (DR-load-bearing).
	var uuid string
	if category == catDir && backingDevice != "" {
		uuid, _ = o.host.ResolveUUID(backingDevice)
	}
	durableID := deriveDurableID(typ, s, backingDevice, uuid)

	// Reachability + state.
	reachable := o.reachable(typ, category, s, backingDevice, exactMount)
	state := hub.StorageStateAttached
	if !reachable {
		state = hub.StorageStateDisconnected
	}

	// Class hint (rotational; local block-backed only — a HINT, never authoritative).
	classHint := ""
	if category == catDir && backingDevice != "" {
		if rot, ok := o.host.Rotational(backingDevice); ok {
			if rot {
				classHint = "slow"
			} else {
				classHint = "fast"
			}
		}
	}

	tgt := hub.StorageTarget{
		Name:          s.Storage,
		Type:          typ,
		DurableID:     durableID,
		State:         state,
		Reachable:     reachable,
		TotalBytes:    s.Total,
		UsedBytes:     s.Used,
		AvailBytes:    s.Avail,
		UsedFraction:  usedFraction(s),
		Content:       s.Content,
		MountPath:     mountPath,
		BackingDevice: backingDevice,
		// R-116: the CONFIGURED path, carried verbatim and never resolved. This is emphatically NOT the
		// fallthrough the comment above forbids — that prohibition is about resolving a device or a UUID
		// from the CONTAINING filesystem when the target is not its own mount, which would hand back
		// root's identity and mis-target a DR re-attach. `s.Path` is the storage's own declaration of
		// where it lives; it identifies nothing but itself, and it is not used for device or UUID
		// resolution anywhere. MountPath stays empty when the mount is gone, which is the truth.
		ConfigPath: s.Path,
		// R-106: the CONFIGURED PBS namespace, carried verbatim from storage.cfg. Empty for every
		// non-pbs storage (Proxmox only emits it on pbs), and empty for a pbs storage in the root
		// namespace — the DR recipe distinguishes those two cases by the storage's TYPE, never by
		// guessing from this string.
		PBSNamespace: s.Namespace,
		ClassHint:    classHint,
		Role:         "", // hub-owned; not derivable from a Proxmox def (slice 10)
		Smart:        hub.SmartSummary{Health: hub.SmartUnknown},
	}

	// Thin-pool DATA fill: surfaced prominently for lvmthin (metadata fill is Phase B/lvs).
	if typ == hub.StorageTypeLVMThin {
		frac := usedFraction(s)
		tgt.ThinPool = &hub.ThinPoolFill{DataUsedFraction: frac}
		if frac >= thinPoolWarnFraction {
			o.logger.Warn("storage: lvmthin pool data fill is high (a full pool corrupts every guest on it)",
				"storage", s.Storage, "data_used_fraction", frac)
		}
	}

	// SMART-only device hint (v0.95.0): a dir-storage that lives INSIDE a shared filesystem (the
	// builtin `local` on the LVM root) has empty backing by design, yet we can still read the PHYSICAL
	// disk's SMART by resolving its containing filesystem. Gated to catDir + no own backing + reachable,
	// so an UNPLUGGED removable (disconnected) never reads root's SMART, and a mounted removable uses
	// its own backing instead.
	smartHint := ""
	if category == catDir && backingDevice == "" && reachable {
		if dev, ok := containingMountDevice(mounts, s.Path); ok {
			smartHint = dev
		}
	}

	return observed{
		target:    tgt,
		src:       s,
		cat:       category,
		smartHint: smartHint,
		known: KnownTarget{
			Name:          s.Storage,
			Type:          typ,
			DurableID:     durableID,
			UUID:          uuid,
			Network:       category == catNetwork,
			MountBacked:   typ == hub.StorageTypeUSB || typ == hub.StorageTypeLocalDir,
			BackingDevice: backingDevice,
			MountPath:     s.Path,
			ReachEndpoint: reachEndpoint(typ, s),
		},
	}
}

// smartDeviceFor maps a backing device (possibly a partition) to its whole-disk path for
// smartctl (which targets the disk, not the partition). Returns ok=false when the result
// isn't a recognized raw disk (e.g. device-mapper / LVM), so SMART is simply skipped.
func smartDeviceFor(device string) (string, bool) {
	// Device-mapper / LVM (Fix A, v0.95.0): resolve to the single backing whole disk via sysfs
	// slaves. This is what finally covers the system SSD under `pve-root`. The sysfs resolution IS
	// the existence check, so we do NOT re-run ValidateSMARTDevice on its result.
	if strings.HasPrefix(device, "/dev/dm-") || strings.HasPrefix(device, "/dev/mapper/") {
		return dmWholeDisk(device)
	}
	dev := device
	if m := reNVMePart.FindStringSubmatch(device); m != nil {
		dev = m[1] // /dev/nvme0n1p2 -> /dev/nvme0n1
	} else if m := reSDPart.FindStringSubmatch(device); m != nil {
		dev = m[1] // /dev/sdb1 -> /dev/sdb
	}
	if ValidateSMARTDevice(dev) != nil {
		return "", false
	}
	return dev, true
}

var (
	reNVMePart = regexp.MustCompile(`^(/dev/nvme[0-9]+n[0-9]+)p[0-9]+$`)
	reSDPart   = regexp.MustCompile(`^(/dev/(?:sd|hd|vd)[a-z]+)[0-9]+$`)
)

// reachable decides whether the target is currently usable.
//   - usb / local-dir: a Felhom extra/removable dir storage is realized as its OWN
//     mountpoint, so reachable = it is currently an exact mount AND its device node exists.
//     Not-its-own-mount = unplugged/unmounted = disconnected. This is the fast USB-drop
//     signal — we deliberately do NOT fall through to PVE's active flag, because the
//     mountpoint directory still exists on the root fs when the device is gone, so active
//     can read stale-attached.
//   - local (builtin PVE "local"): lives within the root fs by design, so trust active.
//   - network (nfs/cifs/pbs) and block-pool (lvmthin/lvm): trust PVE's active flag — PVE
//     actively probes these and flips active=0 when down.
func (o *Observer) reachable(typ string, category storageCategory, s proxmox.Storage, backingDevice string, exactMount bool) bool {
	switch typ {
	case hub.StorageTypeUSB, hub.StorageTypeLocalDir:
		return exactMount && (backingDevice == "" || o.host.DeviceExists(backingDevice))
	default:
		// local, lvmthin, lvm, nfs, cifs, pbs.
		_ = category
		return s.Active == 1
	}
}

// usedFraction prefers Proxmox's reported used_fraction, falling back to used/total.
func usedFraction(s proxmox.Storage) float64 {
	if s.UsedFraction > 0 {
		return s.UsedFraction
	}
	if s.Total > 0 {
		return float64(s.Used) / float64(s.Total)
	}
	return 0
}

// storageCategory groups Proxmox storage types by how state/identity are derived.
type storageCategory int

const (
	catDir     storageCategory = iota // dir-backed (local/usb/local-dir)
	catNetwork                        // nfs/cifs/pbs
	catBlock                          // lvmthin/lvm
	catOther
)

func categorize(pxType string) storageCategory {
	switch pxType {
	case "dir":
		return catDir
	case "nfs", "cifs", "smb", "pbs":
		return catNetwork
	case "lvmthin", "lvm":
		return catBlock
	default:
		return catOther
	}
}

// reportType maps a Proxmox storage type to the reported vocabulary, splitting "dir" into
// builtin local / removable usb / fixed local-dir.
func reportType(s proxmox.Storage, category storageCategory, removable, removableKnown bool) string {
	switch category {
	case catDir:
		if s.Storage == "local" {
			return hub.StorageTypeLocal
		}
		if removableKnown && removable {
			return hub.StorageTypeUSB
		}
		return hub.StorageTypeLocalDir
	case catNetwork:
		switch s.Type {
		case "nfs":
			return hub.StorageTypeNFS
		case "cifs", "smb":
			return hub.StorageTypeCIFS
		case "pbs":
			return hub.StorageTypePBS
		}
	case catBlock:
		if s.Type == "lvmthin" {
			return hub.StorageTypeLVMThin
		}
		return s.Type // "lvm" (thick) passes through
	}
	return s.Type
}

// reachEndpoint builds the host:port the watchdog dials for a network target's
// reachability check (default ports per protocol). "" for non-network targets.
func reachEndpoint(typ string, s proxmox.Storage) string {
	if s.Server == "" {
		return ""
	}
	switch typ {
	case hub.StorageTypeNFS:
		return netJoin(s.Server, "2049")
	case hub.StorageTypeCIFS:
		return netJoin(s.Server, "445")
	case hub.StorageTypePBS:
		return netJoin(s.Server, "8007")
	}
	return ""
}

func netJoin(host, port string) string {
	if strings.Contains(host, ":") && !strings.HasPrefix(host, "[") {
		host = "[" + host + "]" // IPv6 literal
	}
	return host + ":" + port
}

// exactMountDevice finds the mount whose mountpoint EXACTLY equals path (the target is its
// own mount — the meaningful state for a USB/extra disk).
func exactMountDevice(mounts []Mount, path string) (device, mountPoint string, ok bool) {
	if path == "" {
		return "", "", false
	}
	clean := cleanMountPath(path)
	for _, m := range mounts {
		if cleanMountPath(m.MountPoint) == clean {
			return m.Device, m.MountPoint, true
		}
	}
	return "", "", false
}

func cleanMountPath(p string) string {
	p = strings.TrimRight(p, "/")
	if p == "" {
		return "/"
	}
	return p
}

// mergeConfig overlays the cluster-def config fields (which the per-node entry may omit)
// onto a node-storage entry, keeping the node's live usage/active values.
func mergeConfig(node, cluster proxmox.Storage) proxmox.Storage {
	if cluster.Storage == "" {
		return node
	}
	if node.Type == "" {
		node.Type = cluster.Type
	}
	if node.Server == "" {
		node.Server = cluster.Server
	}
	if node.Export == "" {
		node.Export = cluster.Export
	}
	if node.Share == "" {
		node.Share = cluster.Share
	}
	if node.Datastore == "" {
		node.Datastore = cluster.Datastore
	}
	if node.Fingerprint == "" {
		node.Fingerprint = cluster.Fingerprint
	}
	if node.VGName == "" {
		node.VGName = cluster.VGName
	}
	if node.ThinPool == "" {
		node.ThinPool = cluster.ThinPool
	}
	if node.Path == "" {
		node.Path = cluster.Path
	}
	if node.Content == "" {
		node.Content = cluster.Content
	}
	// R-106: the pbs namespace. `NodeStorage` does not return it AT ALL — it is cluster-config only —
	// so without this line `Namespace` is always empty on the merged entry and every consumer sees a
	// root-namespace box. Found by LIVE VALIDATION, not by the unit tests: the DR-recipe tests supply
	// StorageTarget values directly, so they never crossed this merge.
	//
	// This function is a copy-only-what-is-needed allow-list, which is exactly how the gap arose. If you
	// add a consumer of any other type-specific field (`Username` is the remaining unmerged one), add it
	// here too and pin it in TestMergeConfig_CarriesPBSNamespace's table.
	if node.Namespace == "" {
		node.Namespace = cluster.Namespace
	}
	return node
}