pulse/internal/monitoring/monitor_polling_node.go

package monitoring

import (
	"context"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/rcourtman/pulse-go-rewrite/internal/config"
	"github.com/rcourtman/pulse-go-rewrite/internal/models"
	"github.com/rcourtman/pulse-go-rewrite/pkg/proxmox"
	"github.com/rs/zerolog/log"
)

const (
	nodeMemoryCarryForwardMaxAge  = 2 * time.Minute
	pveNodeNetworkRefreshInterval = 5 * time.Minute
)

type pveNodeNetworkGetter interface {
	GetNodeNetworkInterfaces(context.Context, string) ([]proxmox.NodeNetworkInterface, error)
}

func previousNodeNetworkInterfaces(previous []models.Node, instance, nodeID, nodeName string) ([]models.HostNetworkInterface, time.Time, bool) {
	for _, candidate := range previous {
		if strings.TrimSpace(candidate.Instance) != strings.TrimSpace(instance) {
			continue
		}
		if strings.TrimSpace(candidate.ID) == strings.TrimSpace(nodeID) ||
			strings.EqualFold(strings.TrimSpace(candidate.Name), strings.TrimSpace(nodeName)) {
			interfaces := make([]models.HostNetworkInterface, len(candidate.NetworkInterfaces))
			for i, iface := range candidate.NetworkInterfaces {
				interfaces[i] = iface
				interfaces[i].Addresses = append([]string(nil), iface.Addresses...)
			}
			return interfaces, candidate.LastSeen, true
		}
	}
	return nil, time.Time{}, false
}

func mapPVENodeNetworkInterfaces(interfaces []proxmox.NodeNetworkInterface) []models.HostNetworkInterface {
	mapped := make([]models.HostNetworkInterface, 0, len(interfaces))
	for _, iface := range interfaces {
		name := strings.TrimSpace(iface.Iface)
		if name == "" {
			continue
		}
		addresses := make([]string, 0, 2)
		seen := make(map[string]struct{}, 2)
		ipv4 := strings.TrimSpace(iface.CIDR)
		if ipv4 == "" {
			ipv4 = strings.TrimSpace(iface.Address)
		}
		for _, address := range []string{ipv4, iface.Address6} {
			address = strings.TrimSpace(address)
			if address == "" {
				continue
			}
			if _, ok := seen[address]; ok {
				continue
			}
			seen[address] = struct{}{}
			addresses = append(addresses, address)
		}
		mapped = append(mapped, models.HostNetworkInterface{Name: name, Addresses: addresses})
	}
	sort.SliceStable(mapped, func(i, j int) bool {
		return strings.ToLower(mapped[i].Name) < strings.ToLower(mapped[j].Name)
	})
	return mapped
}

func collectPVENodeNetworkInterfaces(
	ctx context.Context,
	client PVEClientInterface,
	instance, nodeID, nodeName string,
	previous []models.Node,
) []models.HostNetworkInterface {
	fallback, previousPollAt, foundPrevious := previousNodeNetworkInterfaces(previous, instance, nodeID, nodeName)
	if foundPrevious && !previousPollAt.IsZero() &&
		previousPollAt.Truncate(pveNodeNetworkRefreshInterval) == time.Now().Truncate(pveNodeNetworkRefreshInterval) {
		return fallback
	}
	getter, ok := client.(pveNodeNetworkGetter)
	if !ok {
		return fallback
	}
	interfaces, err := getter.GetNodeNetworkInterfaces(ctx, nodeName)
	if err != nil {
		log.Debug().Err(err).Str("instance", instance).Str("node", nodeName).
			Msg("Could not refresh Proxmox node network interfaces; retaining last-known inventory")
		return fallback
	}
	return mapPVENodeNetworkInterfaces(interfaces)
}

// carryForwardNodeMemory returns the node's last memory reading from this
// poller's own snapshot when that reading came from a trusted source recently
// enough to stand in for one failed status read. The unified read state is not
// used here: on a node merged with a host agent it holds the agent's reading,
// not the Proxmox value this snapshot vouches for.
func (m *Monitor) carryForwardNodeMemory(instance, node string, now time.Time) (models.Memory, bool) {
	if m == nil {
		return models.Memory{}, false
	}
	m.diagMu.RLock()
	snapshot, ok := m.nodeSnapshots[makeNodeSnapshotKey(instance, node)]
	m.diagMu.RUnlock()
	if !ok || !snapshot.Memory.HasKnownUsage() || snapshot.RetrievedAt.IsZero() {
		return models.Memory{}, false
	}
	age := now.Sub(snapshot.RetrievedAt)
	if age < 0 || age > nodeMemoryCarryForwardMaxAge {
		return models.Memory{}, false
	}
	switch CanonicalMemorySource(snapshot.MemorySource) {
	case "available-field", "derived-free-buffers-cached", "derived-total-minus-used",
		"rrd-memavailable", "rrd-memused":
		return snapshot.Memory, true
	default:
		return models.Memory{}, false
	}
}

func (m *Monitor) pollPVENode(
	ctx context.Context,
	instanceName string,
	instanceCfg *config.PVEInstance,
	client PVEClientInterface,
	node proxmox.Node,
	connectionHealthStr string,
	prevInstanceNodes []models.Node,
) (models.Node, string, string, error) {
	nodeStart := time.Now()
	displayName := getNodeDisplayName(instanceCfg, node.Node)
	connectionHost, guestURL := resolveNodeConnectionInfo(instanceCfg, monitorDiscoveryConfig(m), node.Node)
	nodeID, effectiveStatus := m.determineNodeIDAndStatus(instanceName, instanceCfg, node)
	nodeFallbackFree := uint64(0)
	if node.MaxMem >= node.Mem {
		nodeFallbackFree = node.MaxMem - node.Mem
	}

	modelNode := models.Node{
		ID:                           nodeID,
		NodeIdentity:                 nodeID,
		Name:                         node.Node,
		NativeNameAliases:            config.PVEClusterNodeNativeAliases(instanceCfg, node.Node),
		DisplayName:                  displayName,
		Instance:                     instanceName,
		Host:                         connectionHost,
		GuestURL:                     guestURL,
		Status:                       effectiveStatus,
		Type:                         "node",
		CPU:                          safeFloat(node.CPU), // Proxmox returns 0-1 ratio (e.g., 0.15 = 15%)
		Memory:                       models.UnavailableMemory(clampToInt64(node.MaxMem)),
		Uptime:                       int64(node.Uptime),
		LoadAverage:                  []float64{},
		LastSeen:                     time.Now(),
		ConnectionHealth:             connectionHealthStr, // Use the determined health status
		IsClusterMember:              instanceCfg.IsCluster,
		ClusterName:                  instanceCfg.ClusterName,
		ProviderScopedIdentity:       m.pveNodeUsesProviderScopedIdentity(instanceName, instanceCfg),
		TLSFingerprint:               pveNodeTLSFingerprint(instanceCfg, node.Node),
		TemperatureMonitoringEnabled: instanceCfg.TemperatureMonitoringEnabled,
	}
	var nodeDiskSource string
	modelNode.Disk, nodeDiskSource = m.resolveNodeDisk(instanceName, nodeID, node.Node, node, nil)

	nodeSnapshotRaw := NodeMemoryRaw{
		Total:               node.MaxMem,
		Used:                node.Mem,
		Free:                nodeFallbackFree,
		FallbackTotal:       node.MaxMem,
		FallbackUsed:        node.Mem,
		FallbackFree:        nodeFallbackFree,
		FallbackCalculated:  true,
		ProxmoxMemorySource: "nodes-endpoint",
	}
	nodeMemorySource := "unavailable"
	nodeFallbackReason := "cache-aware-memory-unavailable"

	// Debug logging for disk metrics - note that these values can fluctuate
	// due to thin provisioning and dynamic allocation
	if node.Disk > 0 && node.MaxDisk > 0 {
		log.Debug().
			Str("node", node.Node).
			Uint64("disk", node.Disk).
			Uint64("maxDisk", node.MaxDisk).
			Float64("diskUsage", safePercentage(float64(node.Disk), float64(node.MaxDisk))).
			Msg("Node disk metrics from /nodes endpoint")
	}

	// Track whether we successfully replaced memory metrics with detailed status data
	memoryUpdated := false

	// Get detailed node info if available (skip for offline nodes)
	if effectiveStatus == "online" {
		nodeInfo, nodeErr := client.GetNodeStatus(ctx, node.Node)
		if nodeErr != nil {
			nodeFallbackReason = "node-status-unavailable"
			// If we can't get node status, log but continue with data from /nodes endpoint
			if node.Disk > 0 && node.MaxDisk > 0 {
				log.Warn().
					Str("instance", instanceName).
					Str("node", node.Node).
					Err(nodeErr).
					Uint64("usingDisk", node.Disk).
					Uint64("usingMaxDisk", node.MaxDisk).
					Msg("Could not get node status - disk fallback retained; memory usage unavailable")
			} else {
				log.Warn().
					Str("instance", instanceName).
					Str("node", node.Node).
					Err(nodeErr).
					Uint64("disk", node.Disk).
					Uint64("maxDisk", node.MaxDisk).
					Msg("Could not get node status - memory usage unavailable")
			}
		} else if nodeInfo != nil {
			if nodeInfo.Memory != nil {
				nodeSnapshotRaw.Total = nodeInfo.Memory.Total
				nodeSnapshotRaw.Used = nodeInfo.Memory.Used
				nodeSnapshotRaw.Free = nodeInfo.Memory.Free
				nodeSnapshotRaw.Available = nodeInfo.Memory.Available
				nodeSnapshotRaw.Avail = nodeInfo.Memory.Avail
				nodeSnapshotRaw.Buffers = nodeInfo.Memory.Buffers
				nodeSnapshotRaw.Cached = nodeInfo.Memory.Cached
				nodeSnapshotRaw.Shared = nodeInfo.Memory.Shared
				nodeSnapshotRaw.EffectiveAvailable = nodeInfo.Memory.EffectiveAvailable()
				nodeSnapshotRaw.ProxmoxMemorySource = "node-status"
				nodeSnapshotRaw.FallbackCalculated = false
			}

			// Convert LoadAvg from interface{} to float64
			loadAvg := make([]float64, 0, len(nodeInfo.LoadAvg))
			for _, val := range nodeInfo.LoadAvg {
				switch v := val.(type) {
				case float64:
					loadAvg = append(loadAvg, v)
				case string:
					if f, err := strconv.ParseFloat(v, 64); err == nil {
						loadAvg = append(loadAvg, f)
					}
				}
			}
			modelNode.LoadAverage = loadAvg
			modelNode.KernelVersion = nodeInfo.KernelVersion
			modelNode.PVEVersion = nodeInfo.PVEVersion

			if resolvedDisk, diskSource := m.resolveNodeDisk(instanceName, nodeID, node.Node, node, nodeInfo); diskSource != "" {
				modelNode.Disk = resolvedDisk
				nodeDiskSource = diskSource
			} else {
				log.Warn().
					Str("node", node.Node).
					Bool("rootfsNil", nodeInfo.RootFS == nil).
					Uint64("nodeDisk", node.Disk).
					Uint64("nodeMaxDisk", node.MaxDisk).
					Msg("No valid disk metrics available for node")
			}

			// Update memory metrics to use Available field for more accurate usage
			if nodeInfo.Memory != nil && nodeInfo.Memory.Total > 0 {
				resolvedMemory, resolvedSource, resolvedFallback, resolvedRaw, ok := m.resolveNodeMemory(
					ctx,
					client,
					instanceName,
					node.Node,
					nodeInfo.Memory,
					nodeSnapshotRaw,
				)
				if ok {
					modelNode.Memory = resolvedMemory
					nodeMemorySource = resolvedSource
					nodeFallbackReason = resolvedFallback
					nodeSnapshotRaw = resolvedRaw
					memoryUpdated = resolvedMemory.HasKnownUsage()
				}
			}

			if nodeInfo.CPUInfo != nil {
				// Use MaxCPU from node data for logical CPU count (includes hyperthreading)
				// If MaxCPU is not available or 0, fall back to physical cores
				logicalCores := node.MaxCPU
				if logicalCores == 0 {
					logicalCores = nodeInfo.CPUInfo.Cores
				}

				mhzStr := nodeInfo.CPUInfo.GetMHzString()
				log.Debug().
					Str("node", node.Node).
					Str("model", nodeInfo.CPUInfo.Model).
					Int("cores", nodeInfo.CPUInfo.Cores).
					Int("logicalCores", logicalCores).
					Int("sockets", nodeInfo.CPUInfo.Sockets).
					Str("mhz", mhzStr).
					Msg("Node CPU info from Proxmox")
				modelNode.CPUInfo = models.CPUInfo{
					Model:   nodeInfo.CPUInfo.Model,
					Cores:   logicalCores, // Use logical cores for display
					Sockets: nodeInfo.CPUInfo.Sockets,
					MHz:     mhzStr,
				}
			}
		}
	}

	if effectiveStatus == "online" {
		modelNode.NetworkInterfaces = collectPVENodeNetworkInterfaces(
			ctx, client, instanceName, nodeID, node.Node, prevInstanceNodes,
		)
	} else {
		modelNode.NetworkInterfaces, _, _ = previousNodeNetworkInterfaces(
			prevInstanceNodes, instanceName, nodeID, node.Node,
		)
	}

	// If we couldn't update memory metrics using detailed status, preserve previous accurate values if available
	if !memoryUpdated && effectiveStatus == "online" {
		if prevMem, ok := m.carryForwardNodeMemory(instanceName, node.Node, time.Now()); ok {
			total := int64(node.MaxMem)
			if total == 0 {
				total = prevMem.Total
			}
			used := prevMem.Used
			if total > 0 && used > total {
				used = total
			}
			// Keep the snapshot's used | cache | free split summing to total.
			cache := min(max(prevMem.Cache, 0), max(total-used, 0))
			free := max(total-used-cache, 0)

			preserved := prevMem
			preserved.Total = total
			preserved.Used = used
			preserved.Cache = cache
			preserved.Free = free
			preserved.Usage = safePercentage(float64(used), float64(total))

			modelNode.Memory = preserved
			log.Debug().
				Str("instance", instanceName).
				Str("node", node.Node).
				Msg("Preserving previous memory metrics - node status unavailable this cycle")

			nodeFallbackReason = "preserved-previous-snapshot"
			nodeMemorySource = "previous-snapshot"
			if nodeSnapshotRaw.ProxmoxMemorySource == "node-status" && nodeSnapshotRaw.Total == 0 {
				nodeSnapshotRaw.ProxmoxMemorySource = "previous-snapshot"
			}
		}
	}

	m.recordNodeSnapshot(instanceName, node.Node, NodeMemorySnapshot{
		RetrievedAt:    time.Now(),
		MemorySource:   nodeMemorySource,
		FallbackReason: nodeFallbackReason,
		Memory:         modelNode.Memory,
		Raw:            nodeSnapshotRaw,
	})

	m.collectNodeTemperatureData(ctx, instanceName, instanceCfg, node, &modelNode, prevInstanceNodes, effectiveStatus)
	m.applyNodePendingUpdates(ctx, instanceName, client, node, nodeID, effectiveStatus, &modelNode)
	m.recordNodePollMetrics(instanceName, node, &modelNode, nodeStart)

	return modelNode, effectiveStatus, nodeDiskSource, nil
}