ferrum/internal/api/overview.go

package api

import (
	"context"
	"fmt"
	"log/slog"
	"net/http"
	"sync"
	"time"

	"ferrum/internal/pve"
)

// fleetFanoutLimit bounds how many connections are polled concurrently per
// request, so a fleet of many hosts can't open hundreds of simultaneous
// dials at once.
const fleetFanoutLimit = 8

// The fleet overview is the normalized data model behind the multi-connection
// landing page: one call that rolls every configured Proxmox cluster/server
// up into comparable per-connection summaries — node/guest counts, weighted
// CPU utilization, memory, storage by plugin type, cluster identity, and
// active Ferrum alerts. Widgets and the Overview page render directly from
// this instead of re-aggregating raw cluster/resources client-side.
type fleetOverview struct {
	// PBS entries carry only the identity/health fields (Online, Error,
	// LatencyMs, Alerts) — nodes/guests/storage are PVE concepts and stay
	// zeroed rather than being mislabeled with a failed PVE probe.
	Type         string `json:"type"`
	ConnectionID string `json:"connectionId"`
	Name         string `json:"name"`
	Host         string `json:"host"`
	Port         int    `json:"port"`
	Online       bool   `json:"online"`
	Error        string `json:"error,omitempty"`
	LatencyMs    int64  `json:"latencyMs,omitempty"` // round-trip time of the cluster/resources call
	CheckedAt    string `json:"checkedAt"`           // RFC3339 timestamp of this poll, regardless of outcome

	// Nil for standalone servers (no cluster manager); quorate mirrors
	// /cluster/status.
	Cluster *fleetCluster `json:"cluster,omitempty"`

	Nodes     fleetNodeSummary  `json:"nodes"`
	VMs       fleetGuestSummary `json:"vms"`
	LXCs      fleetGuestSummary `json:"lxcs"`
	Templates int               `json:"templates"`
	HAGuests  int               `json:"haGuests"`

	CPU fleetCPUSummary     `json:"cpu"`
	Mem fleetMemorySummary  `json:"memory"`
	Sto fleetStorageSummary `json:"storage"`

	Alerts fleetAlertSummary `json:"alerts"`
}

type fleetCluster struct {
	Name    string `json:"name"`
	Quorate bool   `json:"quorate"`
	Nodes   int    `json:"nodes"`
}

type fleetNodeSummary struct {
	Total  int `json:"total"`
	Online int `json:"online"`
	Cores  int `json:"cores"`
}

type fleetGuestSummary struct {
	Total   int `json:"total"`
	Running int `json:"running"`
	Stopped int `json:"stopped"`
}

type fleetCPUSummary struct {
	Cores    int     `json:"cores"`     // physical cores across online nodes
	UsedCore float64 `json:"usedCores"` // cpu fraction × cores, summed
	Pct      float64 `json:"pct"`       // weighted utilization 0..100
}

type fleetMemorySummary struct {
	Total int64   `json:"total"`
	Used  int64   `json:"used"`
	Pct   float64 `json:"pct"`
}

type fleetStorageSummary struct {
	Total  int64            `json:"total"`
	Used   int64            `json:"used"`
	Pct    float64          `json:"pct"`
	ByType map[string]int64 `json:"byType"` // used bytes per plugin type (dir, zfspool, rbd, nfs, ...)
}

type fleetAlertSummary struct {
	Critical int `json:"critical"`
	Warning  int `json:"warning"`
}

// fleetOverviewHandler aggregates every connection. Like /inventory, one
// unreachable Proxmox host doesn't blank the fleet — it's reported inline.
func (s *Server) fleetOverviewHandler(w http.ResponseWriter, r *http.Request) {
	rows, err := s.db.QueryContext(r.Context(), `SELECT id, name, type, host, port FROM connections ORDER BY name`)
	if err != nil {
		s.writeError(w, http.StatusInternalServerError, err)
		return
	}
	type conn struct {
		ID, Name, Type, Host string
		Port                 int
	}
	var conns []conn
	for rows.Next() {
		var c conn
		if err := rows.Scan(&c.ID, &c.Name, &c.Type, &c.Host, &c.Port); err != nil {
			rows.Close()
			s.writeError(w, http.StatusInternalServerError, err)
			return
		}
		conns = append(conns, c)
	}
	rows.Close()

	// Active alerts grouped per connection (Ferrum's own alert engine).
	alertsByConn := map[string]fleetAlertSummary{}
	alertRows, err := s.db.QueryContext(r.Context(), `
		SELECT connection_id, severity, COUNT(*) FROM alert_instances
		WHERE status = 'active' GROUP BY connection_id, severity`)
	if err == nil {
		for alertRows.Next() {
			var connID, severity string
			var n int
			if err := alertRows.Scan(&connID, &severity, &n); err == nil {
				a := alertsByConn[connID]
				if severity == "critical" {
					a.Critical += n
				} else {
					a.Warning += n
				}
				alertsByConn[connID] = a
			}
		}
		alertRows.Close()
	} else {
		slog.Warn("fleet overview: alert query failed", "error", err)
	}

	// Fan out to every connection concurrently — sequential polling means one
	// slow or unreachable host (its own 15s HTTP timeout) stalls the whole
	// fleet view on every refresh. Each host is independent, so run them in
	// parallel and let the slowest one bound the request instead of the sum.
	out := make([]fleetOverview, len(conns))
	var wg sync.WaitGroup
	sem := make(chan struct{}, fleetFanoutLimit)
	for i, c := range conns {
		wg.Add(1)
		sem <- struct{}{}
		go func(i int, c conn) {
			defer wg.Done()
			defer func() { <-sem }()
			out[i] = s.buildFleetEntry(r.Context(), c.ID, c.Type, c.Name, c.Host, c.Port, alertsByConn[c.ID])
		}(i, c)
	}
	wg.Wait()

	writeJSON(w, http.StatusOK, out)
}

func (s *Server) buildFleetEntry(ctx context.Context, id, connType, name, host string, port int, alerts fleetAlertSummary) fleetOverview {
	entry := fleetOverview{
		Type:         connType,
		ConnectionID: id, Name: name, Host: host, Port: port,
		Alerts: alerts,
	}

	start := time.Now()
	if connType == "pbs" {
		// PBS hosts don't serve /cluster/resources — probed through the PVE
		// client they'd all report offline. Check reachability with the
		// cheapest authenticated PBS call instead; the PBS pages supply the
		// usage detail, the fleet grid only needs up/down + latency here.
		err := s.connections.RetryOnUnauthorized(id, func() error {
			client, cerr := s.connections.PBSClientFor(ctx, id)
			if cerr != nil {
				return cerr
			}
			_, cerr = client.ListDatastores(ctx)
			return cerr
		})
		entry.LatencyMs = time.Since(start).Milliseconds()
		entry.CheckedAt = time.Now().UTC().Format(time.RFC3339)
		if err != nil {
			entry.Error = err.Error()
			return entry
		}
		entry.Online = true
		return entry
	}

	client, err := s.clientFor(ctx, id)
	if err != nil {
		entry.Error = err.Error()
		entry.CheckedAt = time.Now().UTC().Format(time.RFC3339)
		return entry
	}

	resources, err := client.ClusterResources(ctx)
	entry.LatencyMs = time.Since(start).Milliseconds()
	entry.CheckedAt = time.Now().UTC().Format(time.RFC3339)
	if err != nil {
		entry.Error = err.Error()
		return entry
	}
	entry.Online = true

	if status, err := client.ClusterStatus(ctx); err == nil {
		for _, row := range status {
			if row.Type == "cluster" {
				entry.Cluster = &fleetCluster{
					Name:    row.Name,
					Quorate: row.Quorate == 1,
					Nodes:   countType(status, "node"),
				}
				break
			}
		}
	}

	entry.Sto.ByType = map[string]int64{}
	var storageRows []pve.ClusterResource
	for _, res := range resources {
		switch res.Type {
		case "node":
			entry.Nodes.Total++
			if res.Status == "online" {
				entry.Nodes.Online++
				entry.Nodes.Cores += res.MaxCPU
				entry.CPU.Cores += res.MaxCPU
				entry.CPU.UsedCore += res.CPU * float64(res.MaxCPU)
				entry.Mem.Total += res.MaxMem
				entry.Mem.Used += res.Mem
			}
		case "qemu", "lxc":
			if res.Template == 1 {
				// Templates are qemu rows in /cluster/resources.
				entry.Templates++
				continue
			}
			summary := &entry.VMs
			if res.Type == "lxc" {
				summary = &entry.LXCs
			}
			summary.Total++
			switch res.Status {
			case "running":
				summary.Running++
			case "stopped":
				summary.Stopped++
			}
			if res.HAState != "" {
				entry.HAGuests++
			}
		case "storage":
			if res.MaxDisk > 0 {
				storageRows = append(storageRows, res)
			}
		}
	}
	for _, res := range dedupeSharedStorage(storageRows) {
		entry.Sto.Total += res.MaxDisk
		entry.Sto.Used += res.Disk
		plugin := res.PluginType
		if plugin == "" {
			plugin = "unknown"
		}
		entry.Sto.ByType[plugin] += res.Disk
	}
	if entry.CPU.Cores > 0 {
		entry.CPU.Pct = entry.CPU.UsedCore / float64(entry.CPU.Cores) * 100
	}
	if entry.Mem.Total > 0 {
		entry.Mem.Pct = float64(entry.Mem.Used) / float64(entry.Mem.Total) * 100
	}
	if entry.Sto.Total > 0 {
		entry.Sto.Pct = float64(entry.Sto.Used) / float64(entry.Sto.Total) * 100
	}

	return entry
}

// localOnlyPluginTypes never represent a physical volume shared across
// nodes — every other plugin type (nfs, cifs, pbs, cephfs, rbd, iscsi, ...)
// is treated as network storage even when PVE's own Shared flag is 0, since
// plenty of real setups add the same NFS/CIFS target as a near-identical
// per-node definition without ever ticking "Shared" in the storage config.
var localOnlyPluginTypes = map[string]bool{"dir": true, "lvm": true, "lvmthin": true, "zfspool": true, "btrfs": true}

// dedupeSharedStorage collapses cluster/resources storage rows that
// represent the *same* physical volume reported once per node it's mounted
// on — summing them as-is (the naive approach) multiplied a shared pool's
// capacity by the node count, e.g. an 8-node cluster made a 2TB NFS share
// read as 16TB. Local-only storage (dir/lvm/zfspool/...) is left exactly as
// reported: it's genuinely separate capacity per node.
//
// Within the "could be shared" group, rows are deduped by an exact
// (name, total, used) match: two truly independent volumes are vanishingly
// unlikely to report byte-for-byte identical total *and* used capacity at
// the same instant, while duplicate reports of one physical volume always do.
func dedupeSharedStorage(rows []pve.ClusterResource) []pve.ClusterResource {
	out := make([]pve.ClusterResource, 0, len(rows))
	seen := map[string]bool{}
	for _, r := range rows {
		if r.Shared == 0 && localOnlyPluginTypes[r.PluginType] {
			out = append(out, r)
			continue
		}
		key := fmt.Sprintf("%s|%d|%d", r.Storage, r.MaxDisk, r.Disk)
		if seen[key] {
			continue
		}
		seen[key] = true
		out = append(out, r)
	}
	return out
}

func countType(rows []pve.ClusterStatus, typ string) int {
	n := 0
	for _, r := range rows {
		if r.Type == typ {
			n++
		}
	}
	return n
}