8f5b258424
Compute strings.ToLower(errStr) once and reuse errStrLower instead of calling ToLower three times on the same string in permission check.
2241 lines
72 KiB
Go
2241 lines
72 KiB
Go
package monitoring
|
|
|
|
import (
|
|
"context"
|
|
stderrors "errors"
|
|
"fmt"
|
|
"math"
|
|
"os"
|
|
"sort"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/rcourtman/pulse-go-rewrite/internal/config"
|
|
"github.com/rcourtman/pulse-go-rewrite/internal/errors"
|
|
"github.com/rcourtman/pulse-go-rewrite/internal/logging"
|
|
"github.com/rcourtman/pulse-go-rewrite/internal/models"
|
|
"github.com/rcourtman/pulse-go-rewrite/pkg/proxmox"
|
|
"github.com/rs/zerolog"
|
|
"github.com/rs/zerolog/log"
|
|
)
|
|
|
|
func (m *Monitor) describeInstancesForScheduler() []InstanceDescriptor {
|
|
total := len(m.pveClients) + len(m.pbsClients) + len(m.pmgClients)
|
|
if total == 0 {
|
|
return nil
|
|
}
|
|
|
|
descriptors := make([]InstanceDescriptor, 0, total)
|
|
|
|
if len(m.pveClients) > 0 {
|
|
names := make([]string, 0, len(m.pveClients))
|
|
for name := range m.pveClients {
|
|
names = append(names, name)
|
|
}
|
|
sort.Strings(names)
|
|
for _, name := range names {
|
|
desc := InstanceDescriptor{
|
|
Name: name,
|
|
Type: InstanceTypePVE,
|
|
}
|
|
if m.scheduler != nil {
|
|
if last, ok := m.scheduler.LastScheduled(InstanceTypePVE, name); ok {
|
|
desc.LastScheduled = last.NextRun
|
|
desc.LastInterval = last.Interval
|
|
}
|
|
}
|
|
if m.stalenessTracker != nil {
|
|
if snap, ok := m.stalenessTracker.snapshot(InstanceTypePVE, name); ok {
|
|
desc.LastSuccess = snap.LastSuccess
|
|
desc.LastFailure = snap.LastError
|
|
desc.Metadata = map[string]any{"changeHash": snap.ChangeHash}
|
|
}
|
|
}
|
|
descriptors = append(descriptors, desc)
|
|
}
|
|
}
|
|
|
|
if len(m.pbsClients) > 0 {
|
|
names := make([]string, 0, len(m.pbsClients))
|
|
for name := range m.pbsClients {
|
|
names = append(names, name)
|
|
}
|
|
sort.Strings(names)
|
|
for _, name := range names {
|
|
desc := InstanceDescriptor{
|
|
Name: name,
|
|
Type: InstanceTypePBS,
|
|
}
|
|
if m.scheduler != nil {
|
|
if last, ok := m.scheduler.LastScheduled(InstanceTypePBS, name); ok {
|
|
desc.LastScheduled = last.NextRun
|
|
desc.LastInterval = last.Interval
|
|
}
|
|
}
|
|
if m.stalenessTracker != nil {
|
|
if snap, ok := m.stalenessTracker.snapshot(InstanceTypePBS, name); ok {
|
|
desc.LastSuccess = snap.LastSuccess
|
|
desc.LastFailure = snap.LastError
|
|
desc.Metadata = map[string]any{"changeHash": snap.ChangeHash}
|
|
}
|
|
}
|
|
descriptors = append(descriptors, desc)
|
|
}
|
|
}
|
|
|
|
if len(m.pmgClients) > 0 {
|
|
names := make([]string, 0, len(m.pmgClients))
|
|
for name := range m.pmgClients {
|
|
names = append(names, name)
|
|
}
|
|
sort.Strings(names)
|
|
for _, name := range names {
|
|
desc := InstanceDescriptor{
|
|
Name: name,
|
|
Type: InstanceTypePMG,
|
|
}
|
|
if m.scheduler != nil {
|
|
if last, ok := m.scheduler.LastScheduled(InstanceTypePMG, name); ok {
|
|
desc.LastScheduled = last.NextRun
|
|
desc.LastInterval = last.Interval
|
|
}
|
|
}
|
|
if m.stalenessTracker != nil {
|
|
if snap, ok := m.stalenessTracker.snapshot(InstanceTypePMG, name); ok {
|
|
desc.LastSuccess = snap.LastSuccess
|
|
desc.LastFailure = snap.LastError
|
|
desc.Metadata = map[string]any{"changeHash": snap.ChangeHash}
|
|
}
|
|
}
|
|
descriptors = append(descriptors, desc)
|
|
}
|
|
}
|
|
|
|
return descriptors
|
|
}
|
|
|
|
func (m *Monitor) buildScheduledTasks(now time.Time) []ScheduledTask {
|
|
descriptors := m.describeInstancesForScheduler()
|
|
if len(descriptors) == 0 {
|
|
return nil
|
|
}
|
|
|
|
queueDepth := 0
|
|
if m.taskQueue != nil {
|
|
queueDepth = m.taskQueue.Size()
|
|
}
|
|
|
|
if m.scheduler == nil {
|
|
tasks := make([]ScheduledTask, 0, len(descriptors))
|
|
for _, desc := range descriptors {
|
|
interval := m.baseIntervalForInstanceType(desc.Type)
|
|
if interval <= 0 {
|
|
interval = DefaultSchedulerConfig().BaseInterval
|
|
}
|
|
tasks = append(tasks, ScheduledTask{
|
|
InstanceName: desc.Name,
|
|
InstanceType: desc.Type,
|
|
NextRun: now,
|
|
Interval: interval,
|
|
})
|
|
}
|
|
return tasks
|
|
}
|
|
|
|
return m.scheduler.BuildPlan(now, descriptors, queueDepth)
|
|
}
|
|
|
|
// convertPoolInfoToModel converts Proxmox ZFS pool info to our model
|
|
func convertPoolInfoToModel(poolInfo *proxmox.ZFSPoolInfo) *models.ZFSPool {
|
|
if poolInfo == nil {
|
|
return nil
|
|
}
|
|
|
|
// Use the converter from the proxmox package
|
|
proxmoxPool := poolInfo.ConvertToModelZFSPool()
|
|
|
|
// Convert to our internal model
|
|
modelPool := &models.ZFSPool{
|
|
Name: proxmoxPool.Name,
|
|
State: proxmoxPool.State,
|
|
Status: proxmoxPool.Status,
|
|
Scan: proxmoxPool.Scan,
|
|
ReadErrors: proxmoxPool.ReadErrors,
|
|
WriteErrors: proxmoxPool.WriteErrors,
|
|
ChecksumErrors: proxmoxPool.ChecksumErrors,
|
|
Devices: make([]models.ZFSDevice, 0, len(proxmoxPool.Devices)),
|
|
}
|
|
|
|
// Convert devices
|
|
for _, dev := range proxmoxPool.Devices {
|
|
modelPool.Devices = append(modelPool.Devices, models.ZFSDevice{
|
|
Name: dev.Name,
|
|
Type: dev.Type,
|
|
State: dev.State,
|
|
ReadErrors: dev.ReadErrors,
|
|
WriteErrors: dev.WriteErrors,
|
|
ChecksumErrors: dev.ChecksumErrors,
|
|
Message: dev.Message,
|
|
})
|
|
}
|
|
|
|
return modelPool
|
|
}
|
|
|
|
// pollVMsWithNodes polls VMs from all nodes in parallel using goroutines
|
|
func (m *Monitor) pollVMsWithNodes(ctx context.Context, instanceName string, client PVEClientInterface, nodes []proxmox.Node, nodeEffectiveStatus map[string]string) {
|
|
startTime := time.Now()
|
|
|
|
// Channel to collect VM results from each node
|
|
type nodeResult struct {
|
|
node string
|
|
vms []models.VM
|
|
err error
|
|
}
|
|
|
|
resultChan := make(chan nodeResult, len(nodes))
|
|
var wg sync.WaitGroup
|
|
|
|
// Count online nodes for logging
|
|
onlineNodes := 0
|
|
for _, node := range nodes {
|
|
if nodeEffectiveStatus[node.Node] == "online" {
|
|
onlineNodes++
|
|
}
|
|
}
|
|
|
|
log.Info().
|
|
Str("instance", instanceName).
|
|
Int("totalNodes", len(nodes)).
|
|
Int("onlineNodes", onlineNodes).
|
|
Msg("Starting parallel VM polling")
|
|
|
|
// Launch a goroutine for each online node
|
|
for _, node := range nodes {
|
|
// Skip offline nodes
|
|
if nodeEffectiveStatus[node.Node] != "online" {
|
|
log.Debug().
|
|
Str("node", node.Node).
|
|
Str("status", node.Status).
|
|
Msg("Skipping offline node for VM polling")
|
|
continue
|
|
}
|
|
|
|
wg.Add(1)
|
|
go func(n proxmox.Node) {
|
|
defer wg.Done()
|
|
|
|
nodeStart := time.Now()
|
|
|
|
// Fetch VMs for this node
|
|
vms, err := client.GetVMs(ctx, n.Node)
|
|
if err != nil {
|
|
monErr := errors.NewMonitorError(errors.ErrorTypeAPI, "get_vms", instanceName, err).WithNode(n.Node)
|
|
log.Error().Err(monErr).Str("node", n.Node).Msg("Failed to get VMs; deferring node poll until next cycle")
|
|
resultChan <- nodeResult{node: n.Node, err: err}
|
|
return
|
|
}
|
|
|
|
var nodeVMs []models.VM
|
|
|
|
// Process each VM
|
|
for _, vm := range vms {
|
|
// Skip templates
|
|
if vm.Template == 1 {
|
|
continue
|
|
}
|
|
|
|
// Parse tags
|
|
var tags []string
|
|
if vm.Tags != "" {
|
|
tags = strings.Split(vm.Tags, ";")
|
|
}
|
|
|
|
// Create guest ID (stable across node migrations)
|
|
// Format: instance-VMID
|
|
guestID := fmt.Sprintf("%s-%d", instanceName, vm.VMID)
|
|
|
|
guestRaw := VMMemoryRaw{
|
|
ListingMem: vm.Mem,
|
|
ListingMaxMem: vm.MaxMem,
|
|
Agent: vm.Agent,
|
|
}
|
|
memorySource := "listing-mem"
|
|
|
|
// Initialize metrics from VM listing (may be 0 for disk I/O)
|
|
diskReadBytes := int64(vm.DiskRead)
|
|
diskWriteBytes := int64(vm.DiskWrite)
|
|
networkInBytes := int64(vm.NetIn)
|
|
networkOutBytes := int64(vm.NetOut)
|
|
|
|
// Get memory info for running VMs (and agent status for disk)
|
|
memUsed := uint64(0)
|
|
memTotal := vm.MaxMem
|
|
var vmStatus *proxmox.VMStatus
|
|
var ipAddresses []string
|
|
var networkInterfaces []models.GuestNetworkInterface
|
|
var osName, osVersion, guestAgentVersion string
|
|
|
|
if vm.Status == "running" {
|
|
// Try to get detailed VM status (but don't wait too long)
|
|
statusCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
|
|
if status, err := client.GetVMStatus(statusCtx, n.Node, vm.VMID); err == nil {
|
|
vmStatus = status
|
|
guestRaw.StatusMaxMem = status.MaxMem
|
|
guestRaw.StatusMem = status.Mem
|
|
guestRaw.StatusFreeMem = status.FreeMem
|
|
guestRaw.Balloon = status.Balloon
|
|
guestRaw.BalloonMin = status.BalloonMin
|
|
guestRaw.Agent = status.Agent.Value
|
|
memAvailable := uint64(0)
|
|
if status.MemInfo != nil {
|
|
guestRaw.MemInfoUsed = status.MemInfo.Used
|
|
guestRaw.MemInfoFree = status.MemInfo.Free
|
|
guestRaw.MemInfoTotal = status.MemInfo.Total
|
|
guestRaw.MemInfoAvailable = status.MemInfo.Available
|
|
guestRaw.MemInfoBuffers = status.MemInfo.Buffers
|
|
guestRaw.MemInfoCached = status.MemInfo.Cached
|
|
guestRaw.MemInfoShared = status.MemInfo.Shared
|
|
componentAvailable := status.MemInfo.Free
|
|
if status.MemInfo.Buffers > 0 {
|
|
if math.MaxUint64-componentAvailable < status.MemInfo.Buffers {
|
|
componentAvailable = math.MaxUint64
|
|
} else {
|
|
componentAvailable += status.MemInfo.Buffers
|
|
}
|
|
}
|
|
if status.MemInfo.Cached > 0 {
|
|
if math.MaxUint64-componentAvailable < status.MemInfo.Cached {
|
|
componentAvailable = math.MaxUint64
|
|
} else {
|
|
componentAvailable += status.MemInfo.Cached
|
|
}
|
|
}
|
|
if status.MemInfo.Total > 0 && componentAvailable > status.MemInfo.Total {
|
|
componentAvailable = status.MemInfo.Total
|
|
}
|
|
|
|
availableFromUsed := uint64(0)
|
|
if status.MemInfo.Total > 0 && status.MemInfo.Used > 0 && status.MemInfo.Total >= status.MemInfo.Used {
|
|
availableFromUsed = status.MemInfo.Total - status.MemInfo.Used
|
|
guestRaw.MemInfoTotalMinusUsed = availableFromUsed
|
|
}
|
|
|
|
missingCacheMetrics := status.MemInfo.Available == 0 &&
|
|
status.MemInfo.Buffers == 0 &&
|
|
status.MemInfo.Cached == 0
|
|
|
|
switch {
|
|
case status.MemInfo.Available > 0:
|
|
memAvailable = status.MemInfo.Available
|
|
memorySource = "meminfo-available"
|
|
case status.MemInfo.Free > 0 ||
|
|
status.MemInfo.Buffers > 0 ||
|
|
status.MemInfo.Cached > 0:
|
|
memAvailable = status.MemInfo.Free +
|
|
status.MemInfo.Buffers +
|
|
status.MemInfo.Cached
|
|
memorySource = "meminfo-derived"
|
|
}
|
|
|
|
if memAvailable == 0 && availableFromUsed > 0 && missingCacheMetrics {
|
|
const vmTotalMinusUsedGapTolerance uint64 = 4 * 1024 * 1024
|
|
if availableFromUsed > componentAvailable {
|
|
gap := availableFromUsed - componentAvailable
|
|
if componentAvailable == 0 || gap >= vmTotalMinusUsedGapTolerance {
|
|
memAvailable = availableFromUsed
|
|
memorySource = "meminfo-total-minus-used"
|
|
}
|
|
}
|
|
}
|
|
}
|
|
if vmStatus.Balloon > 0 && vmStatus.Balloon < vmStatus.MaxMem {
|
|
memTotal = vmStatus.Balloon
|
|
guestRaw.DerivedFromBall = true
|
|
}
|
|
switch {
|
|
case memAvailable > 0:
|
|
if memAvailable > memTotal {
|
|
memAvailable = memTotal
|
|
}
|
|
memUsed = memTotal - memAvailable
|
|
case vmStatus.FreeMem > 0:
|
|
memUsed = memTotal - vmStatus.FreeMem
|
|
memorySource = "status-freemem"
|
|
case vmStatus.Mem > 0:
|
|
memUsed = vmStatus.Mem
|
|
memorySource = "status-mem"
|
|
default:
|
|
memUsed = 0
|
|
memorySource = "status-unavailable"
|
|
}
|
|
if memUsed > memTotal {
|
|
memUsed = memTotal
|
|
}
|
|
// Use actual disk I/O values from detailed status
|
|
diskReadBytes = int64(vmStatus.DiskRead)
|
|
diskWriteBytes = int64(vmStatus.DiskWrite)
|
|
networkInBytes = int64(vmStatus.NetIn)
|
|
networkOutBytes = int64(vmStatus.NetOut)
|
|
}
|
|
cancel()
|
|
}
|
|
|
|
if vm.Status != "running" {
|
|
memorySource = "powered-off"
|
|
} else if vmStatus == nil {
|
|
memorySource = "status-unavailable"
|
|
}
|
|
|
|
if vm.Status == "running" && vmStatus != nil {
|
|
guestIPs, guestIfaces, guestOSName, guestOSVersion, agentVersion := m.fetchGuestAgentMetadata(ctx, client, instanceName, n.Node, vm.Name, vm.VMID, vmStatus)
|
|
if len(guestIPs) > 0 {
|
|
ipAddresses = guestIPs
|
|
}
|
|
if len(guestIfaces) > 0 {
|
|
networkInterfaces = guestIfaces
|
|
}
|
|
if guestOSName != "" {
|
|
osName = guestOSName
|
|
}
|
|
if guestOSVersion != "" {
|
|
osVersion = guestOSVersion
|
|
}
|
|
if agentVersion != "" {
|
|
guestAgentVersion = agentVersion
|
|
}
|
|
}
|
|
|
|
// Calculate I/O rates after we have the actual values
|
|
sampleTime := time.Now()
|
|
currentMetrics := IOMetrics{
|
|
DiskRead: diskReadBytes,
|
|
DiskWrite: diskWriteBytes,
|
|
NetworkIn: networkInBytes,
|
|
NetworkOut: networkOutBytes,
|
|
Timestamp: sampleTime,
|
|
}
|
|
diskReadRate, diskWriteRate, netInRate, netOutRate := m.rateTracker.CalculateRates(guestID, currentMetrics)
|
|
|
|
// Debug log disk I/O rates
|
|
if diskReadRate > 0 || diskWriteRate > 0 {
|
|
log.Debug().
|
|
Str("vm", vm.Name).
|
|
Int("vmid", vm.VMID).
|
|
Float64("diskReadRate", diskReadRate).
|
|
Float64("diskWriteRate", diskWriteRate).
|
|
Int64("diskReadBytes", diskReadBytes).
|
|
Int64("diskWriteBytes", diskWriteBytes).
|
|
Msg("VM disk I/O rates calculated")
|
|
}
|
|
|
|
// Set CPU to 0 for non-running VMs
|
|
cpuUsage := safeFloat(vm.CPU)
|
|
if vm.Status != "running" {
|
|
cpuUsage = 0
|
|
}
|
|
|
|
// Calculate disk usage - start with allocated disk size
|
|
// NOTE: The Proxmox cluster/resources API always returns 0 for VM disk usage
|
|
// We must query the guest agent to get actual disk usage
|
|
diskUsed := vm.Disk
|
|
diskTotal := vm.MaxDisk
|
|
diskFree := diskTotal - diskUsed
|
|
diskUsage := safePercentage(float64(diskUsed), float64(diskTotal))
|
|
diskStatusReason := ""
|
|
var individualDisks []models.Disk
|
|
|
|
// For stopped VMs, we can't get guest agent data
|
|
if vm.Status != "running" {
|
|
// Show allocated disk size for stopped VMs
|
|
if diskTotal > 0 {
|
|
diskUsage = -1 // Indicates "allocated size only"
|
|
diskStatusReason = "vm-stopped"
|
|
}
|
|
}
|
|
|
|
// For running VMs, ALWAYS try to get filesystem info from guest agent
|
|
// The cluster/resources endpoint always returns 0 for disk usage
|
|
if vm.Status == "running" && vmStatus != nil && diskTotal > 0 {
|
|
// Log the initial state
|
|
if logging.IsLevelEnabled(zerolog.DebugLevel) {
|
|
log.Debug().
|
|
Str("instance", instanceName).
|
|
Str("vm", vm.Name).
|
|
Int("vmid", vm.VMID).
|
|
Int("agent", vmStatus.Agent.Value).
|
|
Uint64("diskUsed", diskUsed).
|
|
Uint64("diskTotal", diskTotal).
|
|
Msg("VM has 0 disk usage, checking guest agent")
|
|
}
|
|
|
|
// Check if agent is enabled
|
|
if vmStatus.Agent.Value == 0 {
|
|
diskStatusReason = "agent-disabled"
|
|
if logging.IsLevelEnabled(zerolog.DebugLevel) {
|
|
log.Debug().
|
|
Str("instance", instanceName).
|
|
Str("vm", vm.Name).
|
|
Msg("Guest agent disabled in VM config")
|
|
}
|
|
} else if vmStatus.Agent.Value > 0 || diskUsed == 0 {
|
|
if logging.IsLevelEnabled(zerolog.DebugLevel) {
|
|
log.Debug().
|
|
Str("instance", instanceName).
|
|
Str("vm", vm.Name).
|
|
Int("vmid", vm.VMID).
|
|
Msg("Guest agent enabled, fetching filesystem info")
|
|
}
|
|
|
|
// Filesystem info with configurable timeout and retry (refs #592)
|
|
fsInfoRaw, err := m.retryGuestAgentCall(ctx, m.guestAgentFSInfoTimeout, m.guestAgentRetries, func(ctx context.Context) (interface{}, error) {
|
|
return client.GetVMFSInfo(ctx, n.Node, vm.VMID)
|
|
})
|
|
var fsInfo []proxmox.VMFileSystem
|
|
if err == nil {
|
|
if fs, ok := fsInfoRaw.([]proxmox.VMFileSystem); ok {
|
|
fsInfo = fs
|
|
}
|
|
}
|
|
if err != nil {
|
|
// Handle errors
|
|
errStr := err.Error()
|
|
errStrLower := strings.ToLower(errStr)
|
|
log.Warn().
|
|
Str("instance", instanceName).
|
|
Str("vm", vm.Name).
|
|
Int("vmid", vm.VMID).
|
|
Str("error", errStr).
|
|
Msg("Failed to get VM filesystem info from guest agent")
|
|
|
|
// Classify the error type for better user messaging
|
|
// Order matters: check most specific patterns first
|
|
if strings.Contains(errStr, "QEMU guest agent is not running") {
|
|
diskStatusReason = "agent-not-running"
|
|
log.Info().
|
|
Str("instance", instanceName).
|
|
Str("vm", vm.Name).
|
|
Int("vmid", vm.VMID).
|
|
Msg("Guest agent enabled in VM config but not running inside guest OS. Install and start qemu-guest-agent in the VM")
|
|
} else if strings.Contains(errStr, "timeout") {
|
|
diskStatusReason = "agent-timeout"
|
|
} else if strings.Contains(errStr, "500") && (strings.Contains(errStr, "not running") || strings.Contains(errStr, "not available")) {
|
|
// Proxmox API error 500 with "not running"/"not available" indicates guest agent issue, not permissions
|
|
// This commonly happens when guest agent is not installed or not running
|
|
diskStatusReason = "agent-not-running"
|
|
log.Info().
|
|
Str("instance", instanceName).
|
|
Str("vm", vm.Name).
|
|
Int("vmid", vm.VMID).
|
|
Msg("Guest agent communication failed (API error 500). Install and start qemu-guest-agent in the VM")
|
|
} else if (strings.Contains(errStr, "403") || strings.Contains(errStr, "401")) &&
|
|
(strings.Contains(errStrLower, "permission") || strings.Contains(errStrLower, "forbidden") || strings.Contains(errStrLower, "not allowed")) {
|
|
// Only treat as permission-denied if we get explicit auth/permission error codes (401/403)
|
|
// This distinguishes actual permission issues from guest agent unavailability
|
|
diskStatusReason = "permission-denied"
|
|
log.Warn().
|
|
Str("instance", instanceName).
|
|
Str("vm", vm.Name).
|
|
Int("vmid", vm.VMID).
|
|
Msg("Permission denied accessing guest agent. Verify Pulse user has VM.Monitor (PVE 8) or VM.Audit+VM.GuestAgent.Audit (PVE 9) permissions")
|
|
} else if strings.Contains(errStr, "500") {
|
|
// Generic 500 error without clear indicators - likely agent unavailable
|
|
// Refs #596: Proxmox returns 500 errors when guest agent isn't installed/running
|
|
diskStatusReason = "agent-not-running"
|
|
log.Info().
|
|
Str("instance", instanceName).
|
|
Str("vm", vm.Name).
|
|
Int("vmid", vm.VMID).
|
|
Msg("Failed to communicate with guest agent (API error 500). This usually means qemu-guest-agent is not installed or not running in the VM")
|
|
} else {
|
|
diskStatusReason = "agent-error"
|
|
}
|
|
} else if len(fsInfo) == 0 {
|
|
diskStatusReason = "no-filesystems"
|
|
log.Warn().
|
|
Str("instance", instanceName).
|
|
Str("vm", vm.Name).
|
|
Int("vmid", vm.VMID).
|
|
Msg("Guest agent returned empty filesystem list")
|
|
} else {
|
|
log.Info().
|
|
Str("instance", instanceName).
|
|
Str("vm", vm.Name).
|
|
Int("vmid", vm.VMID).
|
|
Int("filesystems", len(fsInfo)).
|
|
Msg("Got filesystem info from guest agent")
|
|
// Aggregate disk usage from all filesystems
|
|
// Fix for #425: Track seen devices to avoid counting duplicates
|
|
var totalBytes, usedBytes uint64
|
|
seenDevices := make(map[string]bool)
|
|
|
|
for _, fs := range fsInfo {
|
|
// Log each filesystem for debugging
|
|
log.Debug().
|
|
Str("vm", vm.Name).
|
|
Str("mountpoint", fs.Mountpoint).
|
|
Str("type", fs.Type).
|
|
Str("disk", fs.Disk).
|
|
Uint64("total", fs.TotalBytes).
|
|
Uint64("used", fs.UsedBytes).
|
|
Msg("Processing filesystem from guest agent")
|
|
|
|
// Skip special filesystems and Windows System Reserved
|
|
// For Windows, mountpoints are like "C:\\" or "D:\\" - don't skip those
|
|
isWindowsDrive := len(fs.Mountpoint) >= 2 && fs.Mountpoint[1] == ':' && strings.Contains(fs.Mountpoint, "\\")
|
|
|
|
if !isWindowsDrive {
|
|
if reason, skip := readOnlyFilesystemReason(fs.Type, fs.TotalBytes, fs.UsedBytes); skip {
|
|
log.Debug().
|
|
Str("vm", vm.Name).
|
|
Str("mountpoint", fs.Mountpoint).
|
|
Str("type", fs.Type).
|
|
Str("skipReason", reason).
|
|
Uint64("total", fs.TotalBytes).
|
|
Uint64("used", fs.UsedBytes).
|
|
Msg("Skipping read-only filesystem from guest agent")
|
|
continue
|
|
}
|
|
|
|
if fs.Type == "tmpfs" || fs.Type == "devtmpfs" ||
|
|
strings.HasPrefix(fs.Mountpoint, "/dev") ||
|
|
strings.HasPrefix(fs.Mountpoint, "/proc") ||
|
|
strings.HasPrefix(fs.Mountpoint, "/sys") ||
|
|
strings.HasPrefix(fs.Mountpoint, "/run") ||
|
|
fs.Mountpoint == "/boot/efi" ||
|
|
fs.Mountpoint == "System Reserved" ||
|
|
strings.Contains(fs.Mountpoint, "System Reserved") ||
|
|
strings.HasPrefix(fs.Mountpoint, "/snap") { // Skip snap mounts
|
|
log.Debug().
|
|
Str("vm", vm.Name).
|
|
Str("mountpoint", fs.Mountpoint).
|
|
Str("type", fs.Type).
|
|
Msg("Skipping special filesystem")
|
|
continue
|
|
}
|
|
}
|
|
|
|
// Skip if we've already seen this device (duplicate mount point)
|
|
if fs.Disk != "" && seenDevices[fs.Disk] {
|
|
log.Debug().
|
|
Str("vm", vm.Name).
|
|
Str("mountpoint", fs.Mountpoint).
|
|
Str("disk", fs.Disk).
|
|
Msg("Skipping duplicate mount of same device")
|
|
continue
|
|
}
|
|
|
|
// Only count real filesystems with valid data
|
|
if fs.TotalBytes > 0 {
|
|
// Mark this device as seen
|
|
if fs.Disk != "" {
|
|
seenDevices[fs.Disk] = true
|
|
}
|
|
|
|
totalBytes += fs.TotalBytes
|
|
usedBytes += fs.UsedBytes
|
|
individualDisks = append(individualDisks, models.Disk{
|
|
Total: int64(fs.TotalBytes),
|
|
Used: int64(fs.UsedBytes),
|
|
Free: int64(fs.TotalBytes - fs.UsedBytes),
|
|
Usage: safePercentage(float64(fs.UsedBytes), float64(fs.TotalBytes)),
|
|
Mountpoint: fs.Mountpoint,
|
|
Type: fs.Type,
|
|
Device: fs.Disk,
|
|
})
|
|
log.Debug().
|
|
Str("vm", vm.Name).
|
|
Str("mountpoint", fs.Mountpoint).
|
|
Str("disk", fs.Disk).
|
|
Uint64("added_total", fs.TotalBytes).
|
|
Uint64("added_used", fs.UsedBytes).
|
|
Msg("Adding filesystem to total")
|
|
} else {
|
|
log.Debug().
|
|
Str("vm", vm.Name).
|
|
Str("mountpoint", fs.Mountpoint).
|
|
Msg("Skipping filesystem with 0 total bytes")
|
|
}
|
|
}
|
|
|
|
// If we got valid data from guest agent, use it
|
|
if totalBytes > 0 {
|
|
diskTotal = totalBytes
|
|
diskUsed = usedBytes
|
|
diskFree = totalBytes - usedBytes
|
|
diskUsage = safePercentage(float64(usedBytes), float64(totalBytes))
|
|
diskStatusReason = "" // Clear reason on success
|
|
|
|
log.Info().
|
|
Str("instance", instanceName).
|
|
Str("vm", vm.Name).
|
|
Int("vmid", vm.VMID).
|
|
Uint64("totalBytes", totalBytes).
|
|
Uint64("usedBytes", usedBytes).
|
|
Float64("usage", diskUsage).
|
|
Msg("✓ Successfully retrieved disk usage from guest agent")
|
|
} else {
|
|
// Only special filesystems found - show allocated disk size instead
|
|
diskStatusReason = "special-filesystems-only"
|
|
if diskTotal > 0 {
|
|
diskUsage = -1 // Show as allocated size
|
|
}
|
|
log.Info().
|
|
Str("instance", instanceName).
|
|
Str("vm", vm.Name).
|
|
Int("filesystems_found", len(fsInfo)).
|
|
Msg("Guest agent provided filesystem info but no usable filesystems found (all were special mounts)")
|
|
}
|
|
}
|
|
} else {
|
|
// No vmStatus available or agent disabled - show allocated disk
|
|
if diskTotal > 0 {
|
|
diskUsage = -1 // Show as allocated size
|
|
diskStatusReason = "no-agent"
|
|
}
|
|
}
|
|
} else if vm.Status == "running" && diskTotal > 0 {
|
|
// Running VM but no vmStatus - show allocated disk
|
|
diskUsage = -1
|
|
diskStatusReason = "no-status"
|
|
}
|
|
|
|
memTotalBytes := clampToInt64(memTotal)
|
|
memUsedBytes := clampToInt64(memUsed)
|
|
if memTotalBytes > 0 && memUsedBytes > memTotalBytes {
|
|
memUsedBytes = memTotalBytes
|
|
}
|
|
memFreeBytes := memTotalBytes - memUsedBytes
|
|
if memFreeBytes < 0 {
|
|
memFreeBytes = 0
|
|
}
|
|
memory := models.Memory{
|
|
Total: memTotalBytes,
|
|
Used: memUsedBytes,
|
|
Free: memFreeBytes,
|
|
Usage: safePercentage(float64(memUsed), float64(memTotal)),
|
|
}
|
|
if guestRaw.Balloon > 0 {
|
|
memory.Balloon = clampToInt64(guestRaw.Balloon)
|
|
}
|
|
|
|
// Create VM model
|
|
modelVM := models.VM{
|
|
ID: guestID,
|
|
VMID: vm.VMID,
|
|
Name: vm.Name,
|
|
Node: n.Node,
|
|
Instance: instanceName,
|
|
Status: vm.Status,
|
|
Type: "qemu",
|
|
CPU: cpuUsage,
|
|
CPUs: vm.CPUs,
|
|
Memory: memory,
|
|
Disk: models.Disk{
|
|
Total: int64(diskTotal),
|
|
Used: int64(diskUsed),
|
|
Free: int64(diskFree),
|
|
Usage: diskUsage,
|
|
},
|
|
Disks: individualDisks,
|
|
DiskStatusReason: diskStatusReason,
|
|
NetworkIn: max(0, int64(netInRate)),
|
|
NetworkOut: max(0, int64(netOutRate)),
|
|
DiskRead: max(0, int64(diskReadRate)),
|
|
DiskWrite: max(0, int64(diskWriteRate)),
|
|
Uptime: int64(vm.Uptime),
|
|
Template: vm.Template == 1,
|
|
LastSeen: sampleTime,
|
|
Tags: tags,
|
|
IPAddresses: ipAddresses,
|
|
OSName: osName,
|
|
OSVersion: osVersion,
|
|
AgentVersion: guestAgentVersion,
|
|
NetworkInterfaces: networkInterfaces,
|
|
}
|
|
|
|
// Zero out metrics for non-running VMs
|
|
if vm.Status != "running" {
|
|
modelVM.CPU = 0
|
|
modelVM.Memory.Usage = 0
|
|
modelVM.Disk.Usage = 0
|
|
modelVM.NetworkIn = 0
|
|
modelVM.NetworkOut = 0
|
|
modelVM.DiskRead = 0
|
|
modelVM.DiskWrite = 0
|
|
}
|
|
|
|
// Trigger guest metadata migration if old format exists
|
|
if m.guestMetadataStore != nil {
|
|
m.guestMetadataStore.GetWithLegacyMigration(guestID, instanceName, n.Node, vm.VMID)
|
|
}
|
|
|
|
nodeVMs = append(nodeVMs, modelVM)
|
|
|
|
m.recordGuestSnapshot(instanceName, modelVM.Type, n.Node, vm.VMID, GuestMemorySnapshot{
|
|
Name: vm.Name,
|
|
Status: vm.Status,
|
|
RetrievedAt: sampleTime,
|
|
MemorySource: memorySource,
|
|
Memory: modelVM.Memory,
|
|
Raw: guestRaw,
|
|
})
|
|
|
|
// Check alerts
|
|
m.alertManager.CheckGuest(modelVM, instanceName)
|
|
}
|
|
|
|
nodeDuration := time.Since(nodeStart)
|
|
log.Debug().
|
|
Str("node", n.Node).
|
|
Int("vms", len(nodeVMs)).
|
|
Dur("duration", nodeDuration).
|
|
Msg("Node VM polling completed")
|
|
|
|
resultChan <- nodeResult{node: n.Node, vms: nodeVMs}
|
|
}(node)
|
|
}
|
|
|
|
// Close channel when all goroutines complete
|
|
go func() {
|
|
wg.Wait()
|
|
close(resultChan)
|
|
}()
|
|
|
|
// Collect results from all nodes
|
|
var allVMs []models.VM
|
|
successfulNodes := 0
|
|
failedNodes := 0
|
|
|
|
for result := range resultChan {
|
|
if result.err != nil {
|
|
failedNodes++
|
|
} else {
|
|
successfulNodes++
|
|
allVMs = append(allVMs, result.vms...)
|
|
}
|
|
}
|
|
|
|
// If we got ZERO VMs but had VMs before (likely cluster health issue),
|
|
// preserve previous VMs instead of clearing them
|
|
if len(allVMs) == 0 && len(nodes) > 0 {
|
|
prevState := m.GetState()
|
|
prevVMCount := 0
|
|
for _, vm := range prevState.VMs {
|
|
if vm.Instance == instanceName {
|
|
allVMs = append(allVMs, vm)
|
|
prevVMCount++
|
|
}
|
|
}
|
|
if prevVMCount > 0 {
|
|
log.Warn().
|
|
Str("instance", instanceName).
|
|
Int("prevVMs", prevVMCount).
|
|
Int("successfulNodes", successfulNodes).
|
|
Int("totalNodes", len(nodes)).
|
|
Msg("Traditional polling returned zero VMs but had VMs before - preserving previous VMs")
|
|
}
|
|
}
|
|
|
|
// Update state with all VMs
|
|
m.state.UpdateVMsForInstance(instanceName, allVMs)
|
|
|
|
duration := time.Since(startTime)
|
|
log.Info().
|
|
Str("instance", instanceName).
|
|
Int("totalVMs", len(allVMs)).
|
|
Int("successfulNodes", successfulNodes).
|
|
Int("failedNodes", failedNodes).
|
|
Dur("duration", duration).
|
|
Msg("Parallel VM polling completed")
|
|
}
|
|
|
|
// pollContainersWithNodes polls containers from all nodes in parallel using goroutines
|
|
func (m *Monitor) pollContainersWithNodes(ctx context.Context, instanceName string, client PVEClientInterface, nodes []proxmox.Node, nodeEffectiveStatus map[string]string) {
|
|
startTime := time.Now()
|
|
|
|
// Channel to collect container results from each node
|
|
type nodeResult struct {
|
|
node string
|
|
containers []models.Container
|
|
err error
|
|
}
|
|
|
|
resultChan := make(chan nodeResult, len(nodes))
|
|
var wg sync.WaitGroup
|
|
|
|
// Count online nodes for logging
|
|
onlineNodes := 0
|
|
for _, node := range nodes {
|
|
if nodeEffectiveStatus[node.Node] == "online" {
|
|
onlineNodes++
|
|
}
|
|
}
|
|
|
|
log.Info().
|
|
Str("instance", instanceName).
|
|
Int("totalNodes", len(nodes)).
|
|
Int("onlineNodes", onlineNodes).
|
|
Msg("Starting parallel container polling")
|
|
|
|
// Launch a goroutine for each online node
|
|
for _, node := range nodes {
|
|
// Skip offline nodes
|
|
if nodeEffectiveStatus[node.Node] != "online" {
|
|
log.Debug().
|
|
Str("node", node.Node).
|
|
Str("status", node.Status).
|
|
Msg("Skipping offline node for container polling")
|
|
continue
|
|
}
|
|
|
|
wg.Add(1)
|
|
go func(n proxmox.Node) {
|
|
defer wg.Done()
|
|
|
|
nodeStart := time.Now()
|
|
|
|
// Fetch containers for this node
|
|
containers, err := client.GetContainers(ctx, n.Node)
|
|
if err != nil {
|
|
monErr := errors.NewMonitorError(errors.ErrorTypeAPI, "get_containers", instanceName, err).WithNode(n.Node)
|
|
log.Error().Err(monErr).Str("node", n.Node).Msg("Failed to get containers")
|
|
resultChan <- nodeResult{node: n.Node, err: err}
|
|
return
|
|
}
|
|
|
|
vmIDs := make([]int, 0, len(containers))
|
|
for _, ct := range containers {
|
|
if ct.Template == 1 {
|
|
continue
|
|
}
|
|
vmIDs = append(vmIDs, int(ct.VMID))
|
|
}
|
|
|
|
rootUsageOverrides := m.collectContainerRootUsage(ctx, client, n.Node, vmIDs)
|
|
|
|
var nodeContainers []models.Container
|
|
|
|
// Process each container
|
|
for _, container := range containers {
|
|
// Skip templates
|
|
if container.Template == 1 {
|
|
continue
|
|
}
|
|
|
|
// Parse tags
|
|
var tags []string
|
|
if container.Tags != "" {
|
|
tags = strings.Split(container.Tags, ";")
|
|
}
|
|
|
|
// Create guest ID (stable across node migrations)
|
|
// Format: instance-VMID
|
|
guestID := fmt.Sprintf("%s-%d", instanceName, container.VMID)
|
|
|
|
// Calculate I/O rates
|
|
currentMetrics := IOMetrics{
|
|
DiskRead: int64(container.DiskRead),
|
|
DiskWrite: int64(container.DiskWrite),
|
|
NetworkIn: int64(container.NetIn),
|
|
NetworkOut: int64(container.NetOut),
|
|
Timestamp: time.Now(),
|
|
}
|
|
diskReadRate, diskWriteRate, netInRate, netOutRate := m.rateTracker.CalculateRates(guestID, currentMetrics)
|
|
|
|
// Set CPU to 0 for non-running containers
|
|
cpuUsage := safeFloat(container.CPU)
|
|
if container.Status != "running" {
|
|
cpuUsage = 0
|
|
}
|
|
|
|
memTotalBytes := clampToInt64(container.MaxMem)
|
|
memUsedBytes := clampToInt64(container.Mem)
|
|
if memTotalBytes > 0 && memUsedBytes > memTotalBytes {
|
|
memUsedBytes = memTotalBytes
|
|
}
|
|
memFreeBytes := memTotalBytes - memUsedBytes
|
|
if memFreeBytes < 0 {
|
|
memFreeBytes = 0
|
|
}
|
|
memUsagePercent := safePercentage(float64(memUsedBytes), float64(memTotalBytes))
|
|
|
|
diskTotalBytes := clampToInt64(container.MaxDisk)
|
|
diskUsedBytes := clampToInt64(container.Disk)
|
|
if diskTotalBytes > 0 && diskUsedBytes > diskTotalBytes {
|
|
diskUsedBytes = diskTotalBytes
|
|
}
|
|
diskFreeBytes := diskTotalBytes - diskUsedBytes
|
|
if diskFreeBytes < 0 {
|
|
diskFreeBytes = 0
|
|
}
|
|
diskUsagePercent := safePercentage(float64(diskUsedBytes), float64(diskTotalBytes))
|
|
|
|
// Create container model
|
|
modelContainer := models.Container{
|
|
ID: guestID,
|
|
VMID: int(container.VMID),
|
|
Name: container.Name,
|
|
Node: n.Node,
|
|
Instance: instanceName,
|
|
Status: container.Status,
|
|
Type: "lxc",
|
|
CPU: cpuUsage,
|
|
CPUs: int(container.CPUs),
|
|
Memory: models.Memory{
|
|
Total: memTotalBytes,
|
|
Used: memUsedBytes,
|
|
Free: memFreeBytes,
|
|
Usage: memUsagePercent,
|
|
},
|
|
Disk: models.Disk{
|
|
Total: diskTotalBytes,
|
|
Used: diskUsedBytes,
|
|
Free: diskFreeBytes,
|
|
Usage: diskUsagePercent,
|
|
},
|
|
NetworkIn: max(0, int64(netInRate)),
|
|
NetworkOut: max(0, int64(netOutRate)),
|
|
DiskRead: max(0, int64(diskReadRate)),
|
|
DiskWrite: max(0, int64(diskWriteRate)),
|
|
Uptime: int64(container.Uptime),
|
|
Template: container.Template == 1,
|
|
LastSeen: time.Now(),
|
|
Tags: tags,
|
|
}
|
|
|
|
if override, ok := rootUsageOverrides[int(container.VMID)]; ok {
|
|
overrideUsed := clampToInt64(override.Used)
|
|
overrideTotal := clampToInt64(override.Total)
|
|
|
|
if overrideUsed > 0 && (modelContainer.Disk.Used == 0 || overrideUsed < modelContainer.Disk.Used) {
|
|
modelContainer.Disk.Used = overrideUsed
|
|
}
|
|
|
|
if overrideTotal > 0 {
|
|
modelContainer.Disk.Total = overrideTotal
|
|
}
|
|
|
|
if modelContainer.Disk.Total > 0 && modelContainer.Disk.Used > modelContainer.Disk.Total {
|
|
modelContainer.Disk.Used = modelContainer.Disk.Total
|
|
}
|
|
|
|
modelContainer.Disk.Free = modelContainer.Disk.Total - modelContainer.Disk.Used
|
|
if modelContainer.Disk.Free < 0 {
|
|
modelContainer.Disk.Free = 0
|
|
}
|
|
|
|
modelContainer.Disk.Usage = safePercentage(float64(modelContainer.Disk.Used), float64(modelContainer.Disk.Total))
|
|
}
|
|
|
|
m.enrichContainerMetadata(ctx, client, instanceName, n.Node, &modelContainer)
|
|
|
|
// Zero out metrics for non-running containers
|
|
if container.Status != "running" {
|
|
modelContainer.CPU = 0
|
|
modelContainer.Memory.Usage = 0
|
|
modelContainer.Disk.Usage = 0
|
|
modelContainer.NetworkIn = 0
|
|
modelContainer.NetworkOut = 0
|
|
modelContainer.DiskRead = 0
|
|
modelContainer.DiskWrite = 0
|
|
}
|
|
|
|
// Trigger guest metadata migration if old format exists
|
|
if m.guestMetadataStore != nil {
|
|
m.guestMetadataStore.GetWithLegacyMigration(guestID, instanceName, n.Node, int(container.VMID))
|
|
}
|
|
|
|
nodeContainers = append(nodeContainers, modelContainer)
|
|
|
|
// Check alerts
|
|
m.alertManager.CheckGuest(modelContainer, instanceName)
|
|
}
|
|
|
|
nodeDuration := time.Since(nodeStart)
|
|
log.Debug().
|
|
Str("node", n.Node).
|
|
Int("containers", len(nodeContainers)).
|
|
Dur("duration", nodeDuration).
|
|
Msg("Node container polling completed")
|
|
|
|
resultChan <- nodeResult{node: n.Node, containers: nodeContainers}
|
|
}(node)
|
|
}
|
|
|
|
// Close channel when all goroutines complete
|
|
go func() {
|
|
wg.Wait()
|
|
close(resultChan)
|
|
}()
|
|
|
|
// Collect results from all nodes
|
|
var allContainers []models.Container
|
|
successfulNodes := 0
|
|
failedNodes := 0
|
|
|
|
for result := range resultChan {
|
|
if result.err != nil {
|
|
failedNodes++
|
|
} else {
|
|
successfulNodes++
|
|
allContainers = append(allContainers, result.containers...)
|
|
}
|
|
}
|
|
|
|
// If we got ZERO containers but had containers before (likely cluster health issue),
|
|
// preserve previous containers instead of clearing them
|
|
if len(allContainers) == 0 && len(nodes) > 0 {
|
|
prevState := m.GetState()
|
|
prevContainerCount := 0
|
|
for _, container := range prevState.Containers {
|
|
if container.Instance == instanceName {
|
|
allContainers = append(allContainers, container)
|
|
prevContainerCount++
|
|
}
|
|
}
|
|
if prevContainerCount > 0 {
|
|
log.Warn().
|
|
Str("instance", instanceName).
|
|
Int("prevContainers", prevContainerCount).
|
|
Int("successfulNodes", successfulNodes).
|
|
Int("totalNodes", len(nodes)).
|
|
Msg("Traditional polling returned zero containers but had containers before - preserving previous containers")
|
|
}
|
|
}
|
|
|
|
// Update state with all containers
|
|
m.state.UpdateContainersForInstance(instanceName, allContainers)
|
|
|
|
duration := time.Since(startTime)
|
|
log.Info().
|
|
Str("instance", instanceName).
|
|
Int("totalContainers", len(allContainers)).
|
|
Int("successfulNodes", successfulNodes).
|
|
Int("failedNodes", failedNodes).
|
|
Dur("duration", duration).
|
|
Msg("Parallel container polling completed")
|
|
}
|
|
|
|
// pollStorageWithNodes polls storage from all nodes in parallel using goroutines
|
|
func (m *Monitor) pollStorageWithNodes(ctx context.Context, instanceName string, client PVEClientInterface, nodes []proxmox.Node) {
|
|
startTime := time.Now()
|
|
|
|
instanceCfg := m.getInstanceConfig(instanceName)
|
|
|
|
// Get cluster storage configuration first (single call)
|
|
clusterStorages, err := client.GetAllStorage(ctx)
|
|
clusterStorageAvailable := err == nil
|
|
if err != nil {
|
|
// Provide detailed context about cluster health issues
|
|
if strings.Contains(err.Error(), "no healthy nodes available") {
|
|
log.Warn().
|
|
Err(err).
|
|
Str("instance", instanceName).
|
|
Msg("Cluster health check shows no healthy endpoints - continuing with direct node storage polling. Check network connectivity and API accessibility from Pulse to each cluster node.")
|
|
} else {
|
|
log.Warn().
|
|
Err(err).
|
|
Str("instance", instanceName).
|
|
Msg("Failed to get cluster storage config - will continue with node storage only")
|
|
}
|
|
}
|
|
|
|
// Create a map for quick lookup of cluster storage config
|
|
clusterStorageMap := make(map[string]proxmox.Storage)
|
|
cephDetected := false
|
|
if clusterStorageAvailable {
|
|
for _, cs := range clusterStorages {
|
|
clusterStorageMap[cs.Storage] = cs
|
|
if !cephDetected && isCephStorageType(cs.Type) {
|
|
cephDetected = true
|
|
}
|
|
}
|
|
}
|
|
|
|
// Channel to collect storage results from each node
|
|
type nodeResult struct {
|
|
node string
|
|
storage []models.Storage
|
|
err error
|
|
}
|
|
|
|
resultChan := make(chan nodeResult, len(nodes))
|
|
var wg sync.WaitGroup
|
|
|
|
// Count online nodes for logging
|
|
onlineNodes := 0
|
|
for _, node := range nodes {
|
|
if node.Status == "online" {
|
|
onlineNodes++
|
|
}
|
|
}
|
|
|
|
log.Info().
|
|
Str("instance", instanceName).
|
|
Int("totalNodes", len(nodes)).
|
|
Int("onlineNodes", onlineNodes).
|
|
Msg("Starting parallel storage polling")
|
|
|
|
// Get existing storage from state to preserve data for offline nodes
|
|
currentState := m.state.GetSnapshot()
|
|
existingStorageMap := make(map[string]models.Storage)
|
|
for _, storage := range currentState.Storage {
|
|
if storage.Instance == instanceName {
|
|
existingStorageMap[storage.ID] = storage
|
|
}
|
|
}
|
|
|
|
// Track which nodes we successfully polled
|
|
polledNodes := make(map[string]bool)
|
|
|
|
// Launch a goroutine for each online node
|
|
for _, node := range nodes {
|
|
// Skip offline nodes but preserve their existing storage data
|
|
if node.Status != "online" {
|
|
log.Debug().
|
|
Str("node", node.Node).
|
|
Str("status", node.Status).
|
|
Msg("Skipping offline node for storage polling - preserving existing data")
|
|
continue
|
|
}
|
|
|
|
wg.Add(1)
|
|
go func(n proxmox.Node) {
|
|
defer wg.Done()
|
|
|
|
nodeStart := time.Now()
|
|
|
|
// Fetch storage for this node
|
|
nodeStorage, err := client.GetStorage(ctx, n.Node)
|
|
if err != nil {
|
|
if shouldAttemptFallback(err) {
|
|
if fallbackStorage, ferr := m.fetchNodeStorageFallback(ctx, instanceCfg, n.Node); ferr == nil {
|
|
log.Warn().
|
|
Str("instance", instanceName).
|
|
Str("node", n.Node).
|
|
Err(err).
|
|
Msg("Primary storage query failed; using direct node fallback")
|
|
nodeStorage = fallbackStorage
|
|
err = nil
|
|
} else {
|
|
log.Warn().
|
|
Str("instance", instanceName).
|
|
Str("node", n.Node).
|
|
Err(ferr).
|
|
Msg("Storage fallback to direct node query failed")
|
|
}
|
|
}
|
|
}
|
|
if err != nil {
|
|
// Handle timeout gracefully - unavailable storage (e.g., NFS mounts) can cause this
|
|
if strings.Contains(err.Error(), "timeout") || strings.Contains(err.Error(), "deadline exceeded") {
|
|
log.Warn().
|
|
Str("node", n.Node).
|
|
Str("instance", instanceName).
|
|
Msg("Storage query timed out - likely due to unavailable storage mounts. Preserving existing storage data for this node.")
|
|
// Send an error result so the node is marked as failed and preservation logic works
|
|
resultChan <- nodeResult{node: n.Node, err: err}
|
|
return
|
|
}
|
|
// For other errors, log as error
|
|
log.Error().
|
|
Err(err).
|
|
Str("node", n.Node).
|
|
Str("instance", instanceName).
|
|
Msg("Failed to get node storage - check API permissions")
|
|
resultChan <- nodeResult{node: n.Node, err: err}
|
|
return
|
|
}
|
|
|
|
var nodeStorageList []models.Storage
|
|
|
|
// Get ZFS pool status for this node if any storage is ZFS
|
|
// This is now production-ready with proper API integration
|
|
var zfsPoolMap = make(map[string]*models.ZFSPool)
|
|
enableZFSMonitoring := os.Getenv("PULSE_DISABLE_ZFS_MONITORING") != "true" // Enabled by default
|
|
|
|
if enableZFSMonitoring {
|
|
hasZFSStorage := false
|
|
for _, storage := range nodeStorage {
|
|
if storage.Type == "zfspool" || storage.Type == "zfs" || storage.Type == "local-zfs" {
|
|
hasZFSStorage = true
|
|
break
|
|
}
|
|
}
|
|
|
|
if hasZFSStorage {
|
|
if poolInfos, err := client.GetZFSPoolsWithDetails(ctx, n.Node); err == nil {
|
|
log.Debug().
|
|
Str("node", n.Node).
|
|
Int("pools", len(poolInfos)).
|
|
Msg("Successfully fetched ZFS pool details")
|
|
|
|
// Convert to our model format
|
|
for _, poolInfo := range poolInfos {
|
|
modelPool := convertPoolInfoToModel(&poolInfo)
|
|
if modelPool != nil {
|
|
zfsPoolMap[poolInfo.Name] = modelPool
|
|
}
|
|
}
|
|
} else {
|
|
// Log but don't fail - ZFS monitoring is optional
|
|
log.Debug().
|
|
Err(err).
|
|
Str("node", n.Node).
|
|
Str("instance", instanceName).
|
|
Msg("Could not get ZFS pool status (may require additional permissions)")
|
|
}
|
|
}
|
|
}
|
|
|
|
// Process each storage
|
|
for _, storage := range nodeStorage {
|
|
if reason, skip := readOnlyFilesystemReason(storage.Type, storage.Total, storage.Used); skip {
|
|
log.Debug().
|
|
Str("node", n.Node).
|
|
Str("storage", storage.Storage).
|
|
Str("type", storage.Type).
|
|
Str("skipReason", reason).
|
|
Uint64("total", storage.Total).
|
|
Uint64("used", storage.Used).
|
|
Msg("Skipping read-only storage mount")
|
|
continue
|
|
}
|
|
|
|
// Create storage ID
|
|
var storageID string
|
|
if instanceName == n.Node {
|
|
storageID = fmt.Sprintf("%s-%s", n.Node, storage.Storage)
|
|
} else {
|
|
storageID = fmt.Sprintf("%s-%s-%s", instanceName, n.Node, storage.Storage)
|
|
}
|
|
|
|
// Get cluster config for this storage
|
|
clusterConfig, hasClusterConfig := clusterStorageMap[storage.Storage]
|
|
|
|
// Determine if shared
|
|
shared := hasClusterConfig && clusterConfig.Shared == 1
|
|
|
|
// Create storage model
|
|
// Initialize Enabled/Active from per-node API response
|
|
modelStorage := models.Storage{
|
|
ID: storageID,
|
|
Name: storage.Storage,
|
|
Node: n.Node,
|
|
Instance: instanceName,
|
|
Type: storage.Type,
|
|
Status: "available",
|
|
Total: int64(storage.Total),
|
|
Used: int64(storage.Used),
|
|
Free: int64(storage.Available),
|
|
Usage: safePercentage(float64(storage.Used), float64(storage.Total)),
|
|
Content: sortContent(storage.Content),
|
|
Shared: shared,
|
|
Enabled: storage.Enabled == 1,
|
|
Active: storage.Active == 1,
|
|
}
|
|
|
|
// If this is ZFS storage, attach pool status information
|
|
if storage.Type == "zfspool" || storage.Type == "zfs" || storage.Type == "local-zfs" {
|
|
// Try to match by storage name or by common ZFS pool names
|
|
poolName := storage.Storage
|
|
|
|
// Common mappings
|
|
if poolName == "local-zfs" {
|
|
poolName = "rpool/data" // Common default
|
|
}
|
|
|
|
// Look for exact match first
|
|
if pool, found := zfsPoolMap[poolName]; found {
|
|
modelStorage.ZFSPool = pool
|
|
} else {
|
|
// Try partial matches for common patterns
|
|
for name, pool := range zfsPoolMap {
|
|
if name == "rpool" && strings.Contains(storage.Storage, "rpool") {
|
|
modelStorage.ZFSPool = pool
|
|
break
|
|
} else if name == "data" && strings.Contains(storage.Storage, "data") {
|
|
modelStorage.ZFSPool = pool
|
|
break
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Override with cluster config if available, but only when the
|
|
// cluster metadata explicitly carries those flags. Some storage
|
|
// types (notably PBS) omit enabled/active, and forcing them to 0
|
|
// would make us skip backup polling even though the node reports
|
|
// the storage as available.
|
|
if hasClusterConfig {
|
|
// Cluster metadata is inconsistent across storage types; PBS storages often omit
|
|
// enabled/active entirely (decode as zero). To avoid marking usable storages as
|
|
// disabled, only override when the cluster explicitly sets the flag to 1.
|
|
if clusterConfig.Enabled == 1 {
|
|
modelStorage.Enabled = true
|
|
}
|
|
if clusterConfig.Active == 1 {
|
|
modelStorage.Active = true
|
|
}
|
|
}
|
|
|
|
// Determine status based on enabled/active flags
|
|
// Priority: disabled storage always shows as "disabled", regardless of active state
|
|
if !modelStorage.Enabled {
|
|
modelStorage.Status = "disabled"
|
|
} else if modelStorage.Active {
|
|
modelStorage.Status = "available"
|
|
} else {
|
|
modelStorage.Status = "inactive"
|
|
}
|
|
|
|
nodeStorageList = append(nodeStorageList, modelStorage)
|
|
}
|
|
|
|
nodeDuration := time.Since(nodeStart)
|
|
log.Debug().
|
|
Str("node", n.Node).
|
|
Int("storage", len(nodeStorageList)).
|
|
Dur("duration", nodeDuration).
|
|
Msg("Node storage polling completed")
|
|
|
|
// If we got empty storage but have existing storage for this node, don't mark as successfully polled
|
|
// This allows preservation logic to keep the existing storage
|
|
if len(nodeStorageList) == 0 {
|
|
// Check if we have existing storage for this node
|
|
hasExisting := false
|
|
for _, existing := range existingStorageMap {
|
|
if existing.Node == n.Node {
|
|
hasExisting = true
|
|
break
|
|
}
|
|
}
|
|
if hasExisting {
|
|
log.Warn().
|
|
Str("node", n.Node).
|
|
Str("instance", instanceName).
|
|
Msg("Node returned empty storage but has existing storage - preserving existing data")
|
|
// Don't send result, allowing preservation logic to work
|
|
return
|
|
}
|
|
}
|
|
|
|
resultChan <- nodeResult{node: n.Node, storage: nodeStorageList}
|
|
}(node)
|
|
}
|
|
|
|
// Close channel when all goroutines complete
|
|
go func() {
|
|
wg.Wait()
|
|
close(resultChan)
|
|
}()
|
|
|
|
// Collect results from all nodes
|
|
var allStorage []models.Storage
|
|
type sharedStorageAggregation struct {
|
|
storage models.Storage
|
|
nodes map[string]struct{}
|
|
nodeIDs map[string]struct{}
|
|
}
|
|
sharedStorageMap := make(map[string]*sharedStorageAggregation) // Map to keep shared storage entries with node affiliations
|
|
|
|
toSortedSlice := func(set map[string]struct{}) []string {
|
|
slice := make([]string, 0, len(set))
|
|
for value := range set {
|
|
slice = append(slice, value)
|
|
}
|
|
sort.Strings(slice)
|
|
return slice
|
|
}
|
|
successfulNodes := 0
|
|
failedNodes := 0
|
|
|
|
for result := range resultChan {
|
|
if result.err != nil {
|
|
failedNodes++
|
|
} else {
|
|
successfulNodes++
|
|
polledNodes[result.node] = true // Mark this node as successfully polled
|
|
for _, storage := range result.storage {
|
|
if storage.Shared {
|
|
// For shared storage, aggregate by storage name so we can retain the reporting nodes
|
|
key := storage.Name
|
|
nodeIdentifier := fmt.Sprintf("%s-%s", storage.Instance, storage.Node)
|
|
|
|
if entry, exists := sharedStorageMap[key]; exists {
|
|
entry.nodes[storage.Node] = struct{}{}
|
|
entry.nodeIDs[nodeIdentifier] = struct{}{}
|
|
|
|
// Prefer the entry with the most up-to-date utilization data
|
|
if storage.Used > entry.storage.Used || (storage.Total > entry.storage.Total && storage.Used == entry.storage.Used) {
|
|
entry.storage.Total = storage.Total
|
|
entry.storage.Used = storage.Used
|
|
entry.storage.Free = storage.Free
|
|
entry.storage.Usage = storage.Usage
|
|
entry.storage.ZFSPool = storage.ZFSPool
|
|
entry.storage.Status = storage.Status
|
|
entry.storage.Enabled = storage.Enabled
|
|
entry.storage.Active = storage.Active
|
|
entry.storage.Content = storage.Content
|
|
entry.storage.Type = storage.Type
|
|
}
|
|
} else {
|
|
sharedStorageMap[key] = &sharedStorageAggregation{
|
|
storage: storage,
|
|
nodes: map[string]struct{}{storage.Node: {}},
|
|
nodeIDs: map[string]struct{}{nodeIdentifier: {}},
|
|
}
|
|
}
|
|
} else {
|
|
// Non-shared storage goes directly to results
|
|
allStorage = append(allStorage, storage)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Add deduplicated shared storage to results
|
|
for _, entry := range sharedStorageMap {
|
|
entry.storage.Node = "cluster"
|
|
entry.storage.Nodes = toSortedSlice(entry.nodes)
|
|
entry.storage.NodeIDs = toSortedSlice(entry.nodeIDs)
|
|
entry.storage.NodeCount = len(entry.storage.Nodes)
|
|
allStorage = append(allStorage, entry.storage)
|
|
}
|
|
|
|
// Preserve existing storage data for nodes that weren't polled (offline or error)
|
|
preservedCount := 0
|
|
for _, existingStorage := range existingStorageMap {
|
|
// Only preserve if we didn't poll this node
|
|
if !polledNodes[existingStorage.Node] && existingStorage.Node != "cluster" {
|
|
allStorage = append(allStorage, existingStorage)
|
|
preservedCount++
|
|
log.Debug().
|
|
Str("node", existingStorage.Node).
|
|
Str("storage", existingStorage.Name).
|
|
Msg("Preserving existing storage data for unpolled node")
|
|
}
|
|
}
|
|
|
|
// Record metrics and check alerts for all storage devices
|
|
for _, storage := range allStorage {
|
|
if m.metricsHistory != nil {
|
|
timestamp := time.Now()
|
|
m.metricsHistory.AddStorageMetric(storage.ID, "usage", storage.Usage, timestamp)
|
|
m.metricsHistory.AddStorageMetric(storage.ID, "used", float64(storage.Used), timestamp)
|
|
m.metricsHistory.AddStorageMetric(storage.ID, "total", float64(storage.Total), timestamp)
|
|
m.metricsHistory.AddStorageMetric(storage.ID, "avail", float64(storage.Free), timestamp)
|
|
}
|
|
|
|
if m.alertManager != nil {
|
|
m.alertManager.CheckStorage(storage)
|
|
}
|
|
}
|
|
|
|
if !cephDetected {
|
|
for _, storage := range allStorage {
|
|
if isCephStorageType(storage.Type) {
|
|
cephDetected = true
|
|
break
|
|
}
|
|
}
|
|
}
|
|
|
|
// Update state with all storage
|
|
m.state.UpdateStorageForInstance(instanceName, allStorage)
|
|
|
|
// Poll Ceph cluster data after refreshing storage information
|
|
m.pollCephCluster(ctx, instanceName, client, cephDetected)
|
|
|
|
duration := time.Since(startTime)
|
|
|
|
// Warn if all nodes failed to get storage
|
|
if successfulNodes == 0 && failedNodes > 0 {
|
|
log.Error().
|
|
Str("instance", instanceName).
|
|
Int("failedNodes", failedNodes).
|
|
Msg("All nodes failed to retrieve storage - check Proxmox API permissions for Datastore.Audit on all storage")
|
|
} else {
|
|
log.Info().
|
|
Str("instance", instanceName).
|
|
Int("totalStorage", len(allStorage)).
|
|
Int("successfulNodes", successfulNodes).
|
|
Int("failedNodes", failedNodes).
|
|
Int("preservedStorage", preservedCount).
|
|
Dur("duration", duration).
|
|
Msg("Parallel storage polling completed")
|
|
}
|
|
}
|
|
|
|
func shouldAttemptFallback(err error) bool {
|
|
if err == nil {
|
|
return false
|
|
}
|
|
msg := strings.ToLower(err.Error())
|
|
return strings.Contains(msg, "timeout") || strings.Contains(msg, "deadline exceeded") || strings.Contains(msg, "context canceled")
|
|
}
|
|
|
|
func (m *Monitor) fetchNodeStorageFallback(ctx context.Context, instanceCfg *config.PVEInstance, nodeName string) ([]proxmox.Storage, error) {
|
|
if m == nil || instanceCfg == nil || !instanceCfg.IsCluster || len(instanceCfg.ClusterEndpoints) == 0 {
|
|
return nil, fmt.Errorf("fallback unavailable")
|
|
}
|
|
|
|
var target string
|
|
hasFingerprint := strings.TrimSpace(instanceCfg.Fingerprint) != ""
|
|
for _, ep := range instanceCfg.ClusterEndpoints {
|
|
if !strings.EqualFold(ep.NodeName, nodeName) {
|
|
continue
|
|
}
|
|
target = clusterEndpointEffectiveURL(ep, instanceCfg.VerifySSL, hasFingerprint)
|
|
if target != "" {
|
|
break
|
|
}
|
|
}
|
|
if strings.TrimSpace(target) == "" {
|
|
return nil, fmt.Errorf("fallback endpoint missing for node %s", nodeName)
|
|
}
|
|
|
|
cfg := proxmox.ClientConfig{
|
|
Host: target,
|
|
VerifySSL: instanceCfg.VerifySSL,
|
|
Fingerprint: instanceCfg.Fingerprint,
|
|
Timeout: m.pollTimeout,
|
|
}
|
|
if instanceCfg.TokenName != "" && instanceCfg.TokenValue != "" {
|
|
cfg.TokenName = instanceCfg.TokenName
|
|
cfg.TokenValue = instanceCfg.TokenValue
|
|
} else {
|
|
cfg.User = instanceCfg.User
|
|
cfg.Password = instanceCfg.Password
|
|
}
|
|
|
|
directClient, err := proxmox.NewClient(cfg)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return directClient.GetStorage(ctx, nodeName)
|
|
}
|
|
|
|
// pollPVENode polls a single PVE node and returns the result
|
|
func (m *Monitor) pollPVENode(
|
|
ctx context.Context,
|
|
instanceName string,
|
|
instanceCfg *config.PVEInstance,
|
|
client PVEClientInterface,
|
|
node proxmox.Node,
|
|
connectionHealthStr string,
|
|
prevNodeMemory map[string]models.Memory,
|
|
prevInstanceNodes []models.Node,
|
|
) (models.Node, string, error) {
|
|
nodeStart := time.Now()
|
|
displayName := getNodeDisplayName(instanceCfg, node.Node)
|
|
connectionHost := instanceCfg.Host
|
|
guestURL := instanceCfg.GuestURL
|
|
if instanceCfg.IsCluster && len(instanceCfg.ClusterEndpoints) > 0 {
|
|
hasFingerprint := instanceCfg.Fingerprint != ""
|
|
for _, ep := range instanceCfg.ClusterEndpoints {
|
|
if strings.EqualFold(ep.NodeName, node.Node) {
|
|
if effective := clusterEndpointEffectiveURL(ep, instanceCfg.VerifySSL, hasFingerprint); effective != "" {
|
|
connectionHost = effective
|
|
}
|
|
if ep.GuestURL != "" {
|
|
guestURL = ep.GuestURL
|
|
}
|
|
break
|
|
}
|
|
}
|
|
}
|
|
|
|
// Apply grace period for node status to prevent flapping
|
|
nodeID := instanceName + "-" + node.Node
|
|
effectiveStatus := node.Status
|
|
now := time.Now()
|
|
|
|
m.mu.Lock()
|
|
if strings.ToLower(node.Status) == "online" {
|
|
// Node is online - update last-online timestamp
|
|
m.nodeLastOnline[nodeID] = now
|
|
} else {
|
|
// Node is reported as offline - check grace period
|
|
lastOnline, exists := m.nodeLastOnline[nodeID]
|
|
if exists && now.Sub(lastOnline) < nodeOfflineGracePeriod {
|
|
// Still within grace period - preserve online status
|
|
effectiveStatus = "online"
|
|
log.Debug().
|
|
Str("instance", instanceName).
|
|
Str("node", node.Node).
|
|
Dur("timeSinceOnline", now.Sub(lastOnline)).
|
|
Dur("gracePeriod", nodeOfflineGracePeriod).
|
|
Msg("Node offline but within grace period - preserving online status")
|
|
} else {
|
|
// Grace period expired or never seen online - mark as offline
|
|
if exists {
|
|
log.Info().
|
|
Str("instance", instanceName).
|
|
Str("node", node.Node).
|
|
Dur("timeSinceOnline", now.Sub(lastOnline)).
|
|
Msg("Node offline and grace period expired - marking as offline")
|
|
}
|
|
}
|
|
}
|
|
m.mu.Unlock()
|
|
|
|
modelNode := models.Node{
|
|
ID: nodeID,
|
|
Name: node.Node,
|
|
DisplayName: displayName,
|
|
Instance: instanceName,
|
|
Host: connectionHost,
|
|
GuestURL: guestURL,
|
|
Status: effectiveStatus,
|
|
Type: "node",
|
|
CPU: safeFloat(node.CPU), // Already in percentage
|
|
Memory: models.Memory{
|
|
Total: int64(node.MaxMem),
|
|
Used: int64(node.Mem),
|
|
Free: int64(node.MaxMem - node.Mem),
|
|
Usage: safePercentage(float64(node.Mem), float64(node.MaxMem)),
|
|
},
|
|
Disk: models.Disk{
|
|
Total: int64(node.MaxDisk),
|
|
Used: int64(node.Disk),
|
|
Free: int64(node.MaxDisk - node.Disk),
|
|
Usage: safePercentage(float64(node.Disk), float64(node.MaxDisk)),
|
|
},
|
|
Uptime: int64(node.Uptime),
|
|
LoadAverage: []float64{},
|
|
LastSeen: time.Now(),
|
|
ConnectionHealth: connectionHealthStr, // Use the determined health status
|
|
IsClusterMember: instanceCfg.IsCluster,
|
|
ClusterName: instanceCfg.ClusterName,
|
|
TemperatureMonitoringEnabled: instanceCfg.TemperatureMonitoringEnabled,
|
|
}
|
|
|
|
nodeSnapshotRaw := NodeMemoryRaw{
|
|
Total: node.MaxMem,
|
|
Used: node.Mem,
|
|
Free: node.MaxMem - node.Mem,
|
|
FallbackTotal: node.MaxMem,
|
|
FallbackUsed: node.Mem,
|
|
FallbackFree: node.MaxMem - node.Mem,
|
|
FallbackCalculated: true,
|
|
ProxmoxMemorySource: "nodes-endpoint",
|
|
}
|
|
nodeMemorySource := "nodes-endpoint"
|
|
var nodeFallbackReason string
|
|
|
|
// 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 - using fallback metrics (memory will include cache/buffers)")
|
|
} 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 - no fallback metrics available (memory will include cache/buffers)")
|
|
}
|
|
} 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
|
|
|
|
// Prefer rootfs data for more accurate disk metrics, but ensure we have valid fallback
|
|
if nodeInfo.RootFS != nil && nodeInfo.RootFS.Total > 0 {
|
|
modelNode.Disk = models.Disk{
|
|
Total: int64(nodeInfo.RootFS.Total),
|
|
Used: int64(nodeInfo.RootFS.Used),
|
|
Free: int64(nodeInfo.RootFS.Free),
|
|
Usage: safePercentage(float64(nodeInfo.RootFS.Used), float64(nodeInfo.RootFS.Total)),
|
|
}
|
|
log.Debug().
|
|
Str("node", node.Node).
|
|
Uint64("rootfsUsed", nodeInfo.RootFS.Used).
|
|
Uint64("rootfsTotal", nodeInfo.RootFS.Total).
|
|
Float64("rootfsUsage", modelNode.Disk.Usage).
|
|
Msg("Using rootfs for disk metrics")
|
|
} else if node.Disk > 0 && node.MaxDisk > 0 {
|
|
// RootFS unavailable but we have valid disk data from /nodes endpoint
|
|
// Keep the values we already set from the nodes list
|
|
log.Debug().
|
|
Str("node", node.Node).
|
|
Bool("rootfsNil", nodeInfo.RootFS == nil).
|
|
Uint64("fallbackDisk", node.Disk).
|
|
Uint64("fallbackMaxDisk", node.MaxDisk).
|
|
Msg("RootFS data unavailable - using /nodes endpoint disk metrics")
|
|
} else {
|
|
// Neither rootfs nor valid node disk data available
|
|
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 {
|
|
var actualUsed uint64
|
|
effectiveAvailable := nodeInfo.Memory.EffectiveAvailable()
|
|
componentAvailable := nodeInfo.Memory.Free
|
|
if nodeInfo.Memory.Buffers > 0 {
|
|
if math.MaxUint64-componentAvailable < nodeInfo.Memory.Buffers {
|
|
componentAvailable = math.MaxUint64
|
|
} else {
|
|
componentAvailable += nodeInfo.Memory.Buffers
|
|
}
|
|
}
|
|
if nodeInfo.Memory.Cached > 0 {
|
|
if math.MaxUint64-componentAvailable < nodeInfo.Memory.Cached {
|
|
componentAvailable = math.MaxUint64
|
|
} else {
|
|
componentAvailable += nodeInfo.Memory.Cached
|
|
}
|
|
}
|
|
if nodeInfo.Memory.Total > 0 && componentAvailable > nodeInfo.Memory.Total {
|
|
componentAvailable = nodeInfo.Memory.Total
|
|
}
|
|
|
|
availableFromUsed := uint64(0)
|
|
if nodeInfo.Memory.Total > 0 && nodeInfo.Memory.Used > 0 && nodeInfo.Memory.Total >= nodeInfo.Memory.Used {
|
|
availableFromUsed = nodeInfo.Memory.Total - nodeInfo.Memory.Used
|
|
}
|
|
nodeSnapshotRaw.TotalMinusUsed = availableFromUsed
|
|
|
|
missingCacheMetrics := nodeInfo.Memory.Available == 0 &&
|
|
nodeInfo.Memory.Avail == 0 &&
|
|
nodeInfo.Memory.Buffers == 0 &&
|
|
nodeInfo.Memory.Cached == 0
|
|
|
|
var rrdMetrics rrdMemCacheEntry
|
|
haveRRDMetrics := false
|
|
usedRRDAvailableFallback := false
|
|
rrdMemUsedFallback := false
|
|
|
|
if effectiveAvailable == 0 && missingCacheMetrics {
|
|
if metrics, err := m.getNodeRRDMetrics(ctx, client, node.Node); err == nil {
|
|
haveRRDMetrics = true
|
|
rrdMetrics = metrics
|
|
if metrics.available > 0 {
|
|
effectiveAvailable = metrics.available
|
|
usedRRDAvailableFallback = true
|
|
}
|
|
if metrics.used > 0 {
|
|
rrdMemUsedFallback = true
|
|
}
|
|
} else if err != nil {
|
|
log.Debug().
|
|
Err(err).
|
|
Str("instance", instanceName).
|
|
Str("node", node.Node).
|
|
Msg("RRD memavailable fallback unavailable")
|
|
}
|
|
}
|
|
|
|
const totalMinusUsedGapTolerance uint64 = 16 * 1024 * 1024
|
|
gapGreaterThanComponents := false
|
|
if availableFromUsed > componentAvailable {
|
|
gap := availableFromUsed - componentAvailable
|
|
if componentAvailable == 0 || gap >= totalMinusUsedGapTolerance {
|
|
gapGreaterThanComponents = true
|
|
}
|
|
}
|
|
|
|
derivedFromTotalMinusUsed := !usedRRDAvailableFallback &&
|
|
missingCacheMetrics &&
|
|
availableFromUsed > 0 &&
|
|
gapGreaterThanComponents &&
|
|
effectiveAvailable == availableFromUsed
|
|
|
|
switch {
|
|
case effectiveAvailable > 0 && effectiveAvailable <= nodeInfo.Memory.Total:
|
|
// Prefer available/avail fields or derived buffers+cache values when present.
|
|
actualUsed = nodeInfo.Memory.Total - effectiveAvailable
|
|
if actualUsed > nodeInfo.Memory.Total {
|
|
actualUsed = nodeInfo.Memory.Total
|
|
}
|
|
|
|
logCtx := log.Debug().
|
|
Str("node", node.Node).
|
|
Uint64("total", nodeInfo.Memory.Total).
|
|
Uint64("effectiveAvailable", effectiveAvailable).
|
|
Uint64("actualUsed", actualUsed).
|
|
Float64("usage", safePercentage(float64(actualUsed), float64(nodeInfo.Memory.Total)))
|
|
if usedRRDAvailableFallback {
|
|
if haveRRDMetrics && rrdMetrics.available > 0 {
|
|
logCtx = logCtx.Uint64("rrdAvailable", rrdMetrics.available)
|
|
}
|
|
logCtx.Msg("Node memory: using RRD memavailable fallback (excludes reclaimable cache)")
|
|
nodeMemorySource = "rrd-memavailable"
|
|
nodeFallbackReason = "rrd-memavailable"
|
|
nodeSnapshotRaw.FallbackCalculated = true
|
|
nodeSnapshotRaw.ProxmoxMemorySource = "rrd-memavailable"
|
|
} else if nodeInfo.Memory.Available > 0 {
|
|
logCtx.Msg("Node memory: using available field (excludes reclaimable cache)")
|
|
nodeMemorySource = "available-field"
|
|
} else if nodeInfo.Memory.Avail > 0 {
|
|
logCtx.Msg("Node memory: using avail field (excludes reclaimable cache)")
|
|
nodeMemorySource = "avail-field"
|
|
} else if derivedFromTotalMinusUsed {
|
|
logCtx.
|
|
Uint64("availableFromUsed", availableFromUsed).
|
|
Uint64("reportedFree", nodeInfo.Memory.Free).
|
|
Msg("Node memory: derived available from total-used gap (cache fields missing)")
|
|
nodeMemorySource = "derived-total-minus-used"
|
|
if nodeFallbackReason == "" {
|
|
nodeFallbackReason = "node-status-total-minus-used"
|
|
}
|
|
nodeSnapshotRaw.FallbackCalculated = true
|
|
nodeSnapshotRaw.ProxmoxMemorySource = "node-status-total-minus-used"
|
|
} else {
|
|
logCtx.
|
|
Uint64("free", nodeInfo.Memory.Free).
|
|
Uint64("buffers", nodeInfo.Memory.Buffers).
|
|
Uint64("cached", nodeInfo.Memory.Cached).
|
|
Msg("Node memory: derived available from free+buffers+cached (excludes reclaimable cache)")
|
|
nodeMemorySource = "derived-free-buffers-cached"
|
|
}
|
|
default:
|
|
switch {
|
|
case rrdMemUsedFallback && haveRRDMetrics && rrdMetrics.used > 0:
|
|
actualUsed = rrdMetrics.used
|
|
if actualUsed > nodeInfo.Memory.Total {
|
|
actualUsed = nodeInfo.Memory.Total
|
|
}
|
|
log.Debug().
|
|
Str("node", node.Node).
|
|
Uint64("total", nodeInfo.Memory.Total).
|
|
Uint64("rrdUsed", rrdMetrics.used).
|
|
Msg("Node memory: using RRD memused fallback (excludes reclaimable cache)")
|
|
nodeMemorySource = "rrd-memused"
|
|
if nodeFallbackReason == "" {
|
|
nodeFallbackReason = "rrd-memused"
|
|
}
|
|
nodeSnapshotRaw.FallbackCalculated = true
|
|
nodeSnapshotRaw.ProxmoxMemorySource = "rrd-memused"
|
|
default:
|
|
// Fallback to traditional used memory if no cache-aware data is exposed
|
|
actualUsed = nodeInfo.Memory.Used
|
|
if actualUsed > nodeInfo.Memory.Total {
|
|
actualUsed = nodeInfo.Memory.Total
|
|
}
|
|
log.Debug().
|
|
Str("node", node.Node).
|
|
Uint64("total", nodeInfo.Memory.Total).
|
|
Uint64("used", actualUsed).
|
|
Msg("Node memory: no cache-aware metrics - using traditional calculation (includes cache)")
|
|
nodeMemorySource = "node-status-used"
|
|
}
|
|
}
|
|
|
|
nodeSnapshotRaw.EffectiveAvailable = effectiveAvailable
|
|
if haveRRDMetrics {
|
|
nodeSnapshotRaw.RRDAvailable = rrdMetrics.available
|
|
nodeSnapshotRaw.RRDUsed = rrdMetrics.used
|
|
nodeSnapshotRaw.RRDTotal = rrdMetrics.total
|
|
}
|
|
|
|
free := int64(nodeInfo.Memory.Total - actualUsed)
|
|
if free < 0 {
|
|
free = 0
|
|
}
|
|
|
|
modelNode.Memory = models.Memory{
|
|
Total: int64(nodeInfo.Memory.Total),
|
|
Used: int64(actualUsed),
|
|
Free: free,
|
|
Usage: safePercentage(float64(actualUsed), float64(nodeInfo.Memory.Total)),
|
|
}
|
|
memoryUpdated = true
|
|
}
|
|
|
|
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 we couldn't update memory metrics using detailed status, preserve previous accurate values if available
|
|
if !memoryUpdated && effectiveStatus == "online" {
|
|
if prevMem, exists := prevNodeMemory[modelNode.ID]; exists && prevMem.Total > 0 {
|
|
total := int64(node.MaxMem)
|
|
if total == 0 {
|
|
total = prevMem.Total
|
|
}
|
|
used := prevMem.Used
|
|
if total > 0 && used > total {
|
|
used = total
|
|
}
|
|
free := total - used
|
|
if free < 0 {
|
|
free = 0
|
|
}
|
|
|
|
preserved := prevMem
|
|
preserved.Total = total
|
|
preserved.Used = used
|
|
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")
|
|
|
|
if nodeFallbackReason == "" {
|
|
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,
|
|
})
|
|
|
|
// Collect temperature data via SSH (non-blocking, best effort)
|
|
// Only attempt for online nodes when temperature monitoring is enabled
|
|
// Check per-node setting first, fall back to global setting
|
|
tempMonitoringEnabled := m.config.TemperatureMonitoringEnabled
|
|
if instanceCfg.TemperatureMonitoringEnabled != nil {
|
|
tempMonitoringEnabled = *instanceCfg.TemperatureMonitoringEnabled
|
|
}
|
|
if effectiveStatus == "online" && m.tempCollector != nil && tempMonitoringEnabled {
|
|
tempCtx, tempCancel := context.WithTimeout(ctx, 30*time.Second) // Increased to accommodate SSH operations via proxy
|
|
defer tempCancel()
|
|
|
|
// Determine SSH hostname to use (most robust approach):
|
|
// Prefer the resolved host for this node, with cluster overrides when available.
|
|
sshHost := modelNode.Host
|
|
foundNodeEndpoint := false
|
|
|
|
if modelNode.IsClusterMember && instanceCfg.IsCluster {
|
|
// Try to find specific endpoint configuration for this node
|
|
if len(instanceCfg.ClusterEndpoints) > 0 {
|
|
hasFingerprint := instanceCfg.Fingerprint != ""
|
|
for _, ep := range instanceCfg.ClusterEndpoints {
|
|
if strings.EqualFold(ep.NodeName, node.Node) {
|
|
if effective := clusterEndpointEffectiveURL(ep, instanceCfg.VerifySSL, hasFingerprint); effective != "" {
|
|
sshHost = effective
|
|
foundNodeEndpoint = true
|
|
}
|
|
break
|
|
}
|
|
}
|
|
}
|
|
|
|
// If no specific endpoint found, fall back to node name
|
|
if !foundNodeEndpoint {
|
|
sshHost = node.Node
|
|
log.Debug().
|
|
Str("node", node.Node).
|
|
Str("instance", instanceCfg.Name).
|
|
Msg("Node endpoint not found in cluster metadata - falling back to node name for temperature collection")
|
|
}
|
|
}
|
|
|
|
if strings.TrimSpace(sshHost) == "" {
|
|
sshHost = node.Node
|
|
}
|
|
|
|
// Use HTTP proxy if configured for this instance, otherwise fall back to socket/SSH
|
|
temp, err := m.tempCollector.CollectTemperatureWithProxy(tempCtx, sshHost, node.Node, instanceCfg.TemperatureProxyURL, instanceCfg.TemperatureProxyToken)
|
|
|
|
if err == nil && temp != nil && temp.Available {
|
|
// Get the current CPU temperature (prefer package, fall back to max)
|
|
currentTemp := temp.CPUPackage
|
|
if currentTemp == 0 && temp.CPUMax > 0 {
|
|
currentTemp = temp.CPUMax
|
|
}
|
|
|
|
// Find previous temperature data for this node to preserve min/max
|
|
var prevTemp *models.Temperature
|
|
for _, prevNode := range prevInstanceNodes {
|
|
if prevNode.ID == modelNode.ID && prevNode.Temperature != nil {
|
|
prevTemp = prevNode.Temperature
|
|
break
|
|
}
|
|
}
|
|
|
|
// Initialize or update min/max tracking
|
|
if prevTemp != nil && prevTemp.CPUMin > 0 {
|
|
// Preserve existing min/max and update if necessary
|
|
temp.CPUMin = prevTemp.CPUMin
|
|
temp.CPUMaxRecord = prevTemp.CPUMaxRecord
|
|
temp.MinRecorded = prevTemp.MinRecorded
|
|
temp.MaxRecorded = prevTemp.MaxRecorded
|
|
|
|
// Update min if current is lower
|
|
if currentTemp > 0 && currentTemp < temp.CPUMin {
|
|
temp.CPUMin = currentTemp
|
|
temp.MinRecorded = time.Now()
|
|
}
|
|
|
|
// Update max if current is higher
|
|
if currentTemp > temp.CPUMaxRecord {
|
|
temp.CPUMaxRecord = currentTemp
|
|
temp.MaxRecorded = time.Now()
|
|
}
|
|
} else if currentTemp > 0 {
|
|
// First reading - initialize min/max to current value
|
|
temp.CPUMin = currentTemp
|
|
temp.CPUMaxRecord = currentTemp
|
|
temp.MinRecorded = time.Now()
|
|
temp.MaxRecorded = time.Now()
|
|
}
|
|
|
|
modelNode.Temperature = temp
|
|
log.Debug().
|
|
Str("node", node.Node).
|
|
Str("sshHost", sshHost).
|
|
Float64("cpuPackage", temp.CPUPackage).
|
|
Float64("cpuMax", temp.CPUMax).
|
|
Float64("cpuMin", temp.CPUMin).
|
|
Float64("cpuMaxRecord", temp.CPUMaxRecord).
|
|
Int("nvmeCount", len(temp.NVMe)).
|
|
Msg("Collected temperature data")
|
|
} else if err != nil {
|
|
log.Debug().
|
|
Str("node", node.Node).
|
|
Str("sshHost", sshHost).
|
|
Bool("isCluster", modelNode.IsClusterMember).
|
|
Int("endpointCount", len(instanceCfg.ClusterEndpoints)).
|
|
Msg("Temperature collection failed - check SSH access")
|
|
} else if temp != nil {
|
|
log.Debug().
|
|
Str("node", node.Node).
|
|
Str("sshHost", sshHost).
|
|
Bool("available", temp.Available).
|
|
Msg("Temperature data unavailable after collection")
|
|
}
|
|
}
|
|
|
|
if m.pollMetrics != nil {
|
|
nodeNameLabel := strings.TrimSpace(node.Node)
|
|
if nodeNameLabel == "" {
|
|
nodeNameLabel = strings.TrimSpace(modelNode.DisplayName)
|
|
}
|
|
if nodeNameLabel == "" {
|
|
nodeNameLabel = "unknown-node"
|
|
}
|
|
|
|
success := true
|
|
nodeErrReason := ""
|
|
health := strings.ToLower(strings.TrimSpace(modelNode.ConnectionHealth))
|
|
if health != "" && health != "healthy" {
|
|
success = false
|
|
nodeErrReason = fmt.Sprintf("connection health %s", health)
|
|
}
|
|
|
|
status := strings.ToLower(strings.TrimSpace(modelNode.Status))
|
|
if success && status != "" && status != "online" {
|
|
success = false
|
|
nodeErrReason = fmt.Sprintf("status %s", status)
|
|
}
|
|
|
|
var nodeErr error
|
|
if !success {
|
|
if nodeErrReason == "" {
|
|
nodeErrReason = "unknown node error"
|
|
}
|
|
nodeErr = stderrors.New(nodeErrReason)
|
|
}
|
|
|
|
m.pollMetrics.RecordNodeResult(NodePollResult{
|
|
InstanceName: instanceName,
|
|
InstanceType: "pve",
|
|
NodeName: nodeNameLabel,
|
|
Success: success,
|
|
Error: nodeErr,
|
|
StartTime: nodeStart,
|
|
EndTime: time.Now(),
|
|
})
|
|
}
|
|
|
|
return modelNode, effectiveStatus, nil
|
|
}
|