package hub import ( "context" "fmt" "log/slog" "time" "gitea.dooplex.hu/admin/felhom-agent/internal/capability" "gitea.dooplex.hu/admin/felhom-agent/internal/proxmox" ) // proxmoxReader is the read-only subset the collector needs. Signatures match the // REAL internal/proxmox.Client surface (slice 1): the node is held by the Client // (no per-call node arg), reads return values (not pointers), and the guest type is // proxmox.Guest. (The task's sketch used node-arg/pointer/LXC shapes; adapted to // the actual exports per its instruction — no proxmox changes were needed: ListLXC // already carries status/maxmem/maxdisk, GuestConfig carries cores.) type proxmoxReader interface { Node() string NodeStatus(ctx context.Context) (proxmox.NodeStatus, error) ListLXC(ctx context.Context) ([]proxmox.Guest, error) GuestConfig(ctx context.Context, vmid int) (proxmox.GuestConfig, error) } // StorageObserver is the seam the storage layer (internal/storage) plugs into to fill the // report's storage_targets. Defined here (consumer-side) so hub does NOT import storage — // storage imports hub for the wire type, and main.go wires the concrete observer in. Same // pattern as proxmoxReader / CloudflaredProber. A nil observer (slice-3 behaviour, or a // host with no storage layer) yields an empty []StorageTarget without error. type StorageObserver interface { Observe(ctx context.Context) ([]StorageTarget, error) } // BackupReporter / RestoreTestReporter are the slice-6 seams the backup layer plugs into // (same consumer-side pattern as StorageObserver — hub does not import the backup package). // They return the agent's LATEST-known backup-per-target / restore-test result (point-in-time // state the backup layer accumulates), not a live scan. A nil reporter → empty slice. type BackupReporter interface { Backups(ctx context.Context) []Backup } type RestoreTestReporter interface { RestoreTests(ctx context.Context) []RestoreTest } // PBSReporter is the slice-6-Phase-B seam the pbs verify loop plugs into (same pattern). // Returns the agent's latest-known PBS snapshot inventory + verify-state. nil → empty. type PBSReporter interface { PBSSnapshots(ctx context.Context) []PBSSnapshot } // WireguardReporter is the S3 seam the wgtunnel loop plugs into (same consumer-side pattern — // hub does not import wgtunnel). nil (feature disabled) → no wireguard stanza on the report. type WireguardReporter interface { WireguardStatus(ctx context.Context) *WireguardStatus } // Collector builds a HostReport from read-only sources. All deps are behind narrow // interfaces for unit testing. type Collector struct { px proxmoxReader cf CloudflaredProber storage StorageObserver backups BackupReporter restoreTests RestoreTestReporter pbs PBSReporter temp TempReader // slice 9: host CPU/chassis temp (nil-safe → nil temp) capProbe func(ctx context.Context) []capability.Status // v0.44.0: privileged-capability self-check (nil → empty) leafFP string // v0.48.0: served local-API leaf fp (static per process; "" when local API disabled) wg WireguardReporter // S3: offsite-tunnel status (nil → stanza omitted) selfUpdate SelfUpdateReporter // D1: agent self-update pending status (nil → false) mgmtPlane MgmtPlaneReporter // G1: management-plane health (nil → stanza omitted) hostID string agentVersion string logger *slog.Logger now func() time.Time } // NewCollector builds a collector. hostID echoes config.Hub.HostID; agentVersion is // the binary version. storage/backups/restoreTests/pbs may be nil (their collections emit empty). func NewCollector(px proxmoxReader, cf CloudflaredProber, storage StorageObserver, backups BackupReporter, restoreTests RestoreTestReporter, pbs PBSReporter, hostID, agentVersion string, logger *slog.Logger) *Collector { if logger == nil { logger = slog.Default() } return &Collector{ px: px, cf: cf, storage: storage, backups: backups, restoreTests: restoreTests, pbs: pbs, temp: SysfsTempReader{}, // slice 9: real sysfs reader by default; tests inject a fake hostID: hostID, agentVersion: agentVersion, logger: logger, now: func() time.Time { return time.Now().UTC() }, } } // SetTempReader overrides the host-temp source (tests inject a fake; a nil reader disables temp). // Returns the collector for chaining. func (c *Collector) SetTempReader(t TempReader) *Collector { c.temp = t return c } // SetCapabilityProber wires the privileged-capability self-check (v0.44.0): each collect runs it // and attaches the snapshot. nil → the report carries an empty []. Returns the collector for chaining. func (c *Collector) SetCapabilityProber(probe func(ctx context.Context) []capability.Status) *Collector { c.capProbe = probe return c } // SetLeafFingerprint records the served local-API leaf fp (v0.48.0) to ride every host report (the hub // watches it for a re-key). Static per process — set once at startup. "" when the local API is // disabled. Returns the collector for chaining. func (c *Collector) SetLeafFingerprint(fp string) *Collector { c.leafFP = fp return c } // SetWireguardReporter wires the offsite-tunnel status source (S3; nil-safe → stanza omitted). // Returns the collector for chaining. func (c *Collector) SetWireguardReporter(w WireguardReporter) *Collector { c.wg = w return c } // SelfUpdateReporter is the D1 seam the selfupdate commit-manager plugs into (same consumer-side // pattern — hub does not import selfupdate). nil (feature not wired) → pending=false on the report. type SelfUpdateReporter interface { // SelfUpdatePending reports whether a signed update has flipped the binary but not yet // committed, and the awaited version. SelfUpdatePending() (pending bool, version string) } // SetSelfUpdateReporter wires the agent self-update pending-status source (D1; nil-safe → false). // Returns the collector for chaining. func (c *Collector) SetSelfUpdateReporter(s SelfUpdateReporter) *Collector { c.selfUpdate = s return c } // MgmtPlaneReporter is the G1 seam the mgmtplane observer plugs into (same consumer-side pattern — // hub does not import mgmtplane). nil (feature not wired) → no mgmt_plane stanza on the report. type MgmtPlaneReporter interface { MgmtPlaneStatus(ctx context.Context) *MgmtPlaneStatus } // SetMgmtPlaneReporter wires the management-plane health source (G1; nil-safe → stanza omitted). // Returns the collector for chaining. func (c *Collector) SetMgmtPlaneReporter(m MgmtPlaneReporter) *Collector { c.mgmtPlane = m return c } // Collect builds the report. Best-effort liveness: a failed NodeStatus is a hard // error (no useful report — the cycle skips the POST); a failed per-guest // GuestConfig degrades that guest to status="unknown" without spec but still sends; // a cloudflared probe failure yields status="unknown" and is never fatal. func (c *Collector) Collect(ctx context.Context) (*HostReport, error) { ns, err := c.px.NodeStatus(ctx) if err != nil { return nil, fmt.Errorf("hub: NodeStatus failed (no useful report): %w", err) } host := hostMetrics(c.px.Node(), ns) host.CPUTempC = c.cpuTempC(ctx) // slice 9: operator freebie — temp now rides the hub report too report := &HostReport{ HostID: c.hostID, ReportedAt: c.now().Format(time.RFC3339), AgentVersion: c.agentVersion, Host: host, Guests: c.collectGuests(ctx), // storage_targets populated this slice (slice 5) via the observer; the rest stay // defined-but-empty (slice 6). Non-nil so they marshal as []. StorageTargets: c.collectStorage(ctx), Backups: c.collectBackups(ctx), RestoreTests: c.collectRestoreTests(ctx), PBSSnapshots: c.collectPBSSnapshots(ctx), AuditTail: []AuditEntry{}, Cloudflared: Cloudflared{Status: c.cloudflaredStatus(ctx)}, Capabilities: c.capabilities(ctx), LeafFingerprint: c.leafFP, } // DR recipe host-half — derived from the just-collected guest/storage/PBS facts (no new reads). // Secret-free by construction (identifiers/intents/sizes/coordinates only). report.DRRecipe = BuildDRRecipeHostHalf(report.Guests, report.StorageTargets, report.PBSSnapshots) // S3: offsite-tunnel status stanza (nil reporter = feature disabled → omitted; the pubkey in // it is the operator's revocation-recovery handle). if c.wg != nil { report.Wireguard = c.wg.WireguardStatus(ctx) } // D1: agent self-update pending status (nil reporter → pending=false, the steady state). if c.selfUpdate != nil { report.SelfUpdatePending, report.SelfUpdatePendingVersion = c.selfUpdate.SelfUpdatePending() } // G1: management-plane health (nil reporter = feature not wired → stanza omitted). if c.mgmtPlane != nil { report.MgmtPlane = c.mgmtPlane.MgmtPlaneStatus(ctx) } return report, nil } // capabilities runs the privileged-capability self-check for this report (v0.44.0), or returns an // empty (non-nil) slice when no prober is wired (dev/test). Never fatal — serve-degraded. func (c *Collector) capabilities(ctx context.Context) []capability.Status { if c.capProbe == nil { return []capability.Status{} } if s := c.capProbe(ctx); s != nil { return s } return []capability.Status{} } // HostMetricsNow does a FRESH NodeStatus + CPU-temp read and returns just the host block (no // guests/storage). It is the source for the local API's GET /host/metrics (slice 9) — current // cpu%/temp, not the 15-min hub-report snapshot. Storage targets come from the observer // separately. A NodeStatus failure is a hard error (no useful host view); a missing temp sensor // degrades to nil (never an error). func (c *Collector) HostMetricsNow(ctx context.Context) (HostMetrics, error) { ns, err := c.px.NodeStatus(ctx) if err != nil { return HostMetrics{}, fmt.Errorf("hub: NodeStatus failed: %w", err) } h := hostMetrics(c.px.Node(), ns) h.CPUTempC = c.cpuTempC(ctx) return h, nil } // cpuTempC reads the host CPU/chassis temp via the TempReader seam (nil-safe → nil). func (c *Collector) cpuTempC(ctx context.Context) *int { if c.temp == nil { return nil } return c.temp.CPUTempC(ctx) } func hostMetrics(node string, ns proxmox.NodeStatus) HostMetrics { h := HostMetrics{ Node: node, CPUPercent: ns.CPU * 100, // PVE cpu is a 0..1 fraction MemoryTotalBytes: ns.Memory.Total, MemoryUsedBytes: ns.Memory.Used, DiskTotalBytes: ns.RootFS.Total, DiskUsedBytes: ns.RootFS.Used, LoadAvg: ns.LoadAvg, UptimeSeconds: ns.Uptime, } h.MemoryPercent = percent(ns.Memory.Used, ns.Memory.Total) h.DiskPercent = percent(ns.RootFS.Used, ns.RootFS.Total) if h.LoadAvg == nil { h.LoadAvg = []string{} } return h } func (c *Collector) collectGuests(ctx context.Context) []Guest { lxc, err := c.px.ListLXC(ctx) if err != nil { // Not fatal: a report with no guest list still carries host liveness. c.logger.Warn("hub: ListLXC failed; reporting no guests", "err", err) return []Guest{} } guests := make([]Guest, 0, len(lxc)) for _, g := range lxc { entry := Guest{VMID: g.VMID, Name: g.Name, Status: g.Status, ControllerVersion: ""} // Normalize an empty run-status to "unknown" so the wire value is always one // of running|stopped|unknown (matches the hub handler's empty→unknown default). if entry.Status == "" { entry.Status = "unknown" } // GuestConfig supplies cores; memory/disk come from the list entry (bytes). // On failure, KEEP the known run-status from ListLXC — only the spec is lost. cfg, err := c.px.GuestConfig(ctx, g.VMID) if err != nil { c.logger.Warn("hub: GuestConfig failed; spec omitted (run-status kept)", "vmid", g.VMID, "err", err) entry.Spec = nil } else { entry.Spec = &GuestSpec{ Cores: cfg.Cores, MemoryBytes: g.MaxMem, DiskBytes: g.MaxDisk, } } guests = append(guests, entry) } return guests } // collectStorage builds the storage_targets via the observer. A nil observer (no storage // layer wired) or an observe error degrades to an empty list — storage detail is // best-effort and must never sink the heartbeat (host liveness is the priority). func (c *Collector) collectStorage(ctx context.Context) []StorageTarget { if c.storage == nil { return []StorageTarget{} } targets, err := c.storage.Observe(ctx) if err != nil { c.logger.Warn("hub: storage observe failed; reporting no storage targets", "err", err) return []StorageTarget{} } if targets == nil { return []StorageTarget{} } return targets } // collectBackups / collectRestoreTests read the agent's latest backup + restore-test state // via the seams. Best-effort: a nil reporter or nil slice degrades to an empty (non-nil) // list so the collection always marshals as []. func (c *Collector) collectBackups(ctx context.Context) []Backup { if c.backups == nil { return []Backup{} } if b := c.backups.Backups(ctx); b != nil { return b } return []Backup{} } func (c *Collector) collectRestoreTests(ctx context.Context) []RestoreTest { if c.restoreTests == nil { return []RestoreTest{} } if r := c.restoreTests.RestoreTests(ctx); r != nil { return r } return []RestoreTest{} } // collectPBSSnapshots reads the latest PBS snapshot inventory via the seam (nil → empty). func (c *Collector) collectPBSSnapshots(ctx context.Context) []PBSSnapshot { if c.pbs == nil { return []PBSSnapshot{} } if s := c.pbs.PBSSnapshots(ctx); s != nil { return s } return []PBSSnapshot{} } func (c *Collector) cloudflaredStatus(ctx context.Context) string { if c.cf == nil { return "unknown" } st, err := c.cf.Status(ctx) if err != nil || st == "" { c.logger.Warn("hub: cloudflared probe failed", "err", err) return "unknown" } return st } func percent(used, total int64) float64 { if total <= 0 { return 0 } return float64(used) / float64(total) * 100 }