mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-10-14 08:10:26 +00:00
rcourtman
2f4e9957a5
ac0fb263c changed preserved guests to carry their real runtime status
("running") instead of the stringified aggregate status ("online"). The
three guest metric recording loops gate on Status == "running", so guests
carried forward while their node sits in the grace period started passing
that gate. The carried-forward projection has no counters, so every cycle
wrote CPU, disk and network zeroes into the history and the persistent
store. Sparklines and stored metrics showed a collapse to zero for a guest
Pulse could not see, rather than a gap. Memory escaped only by accident,
because the projection leaves usage unknown and historyMemoryUsage returns
the -1 sentinel.
Freshly built guests are stamped with the cycle's sample time while
preserved guests keep the LastSeen of the cycle that observed them, so that
field already distinguishes the two. All three loops now gate on
guestObservedInCycle: the efficient path in recordGuestMetrics, and the
traditional VM and container paths, which had the same defect inline.
The guard fails open when LastSeen is absent. Dropping a real sample is the
worse error and the harder one to notice, so no evidence means record.
Regression coverage asserts both directions, and both were verified by
breaking them: removing the guard reports the fabricated zero sample, and
inverting it reports the observed guest losing its sample entirely.
419 lines
13 KiB
Go
419 lines
13 KiB
Go
package monitoring
|
|
|
|
import (
|
|
"context"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/rcourtman/pulse-go-rewrite/internal/models"
|
|
"github.com/rcourtman/pulse-go-rewrite/internal/monitoring/errors"
|
|
"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
|
|
"github.com/rcourtman/pulse-go-rewrite/pkg/proxmox"
|
|
"github.com/rs/zerolog/log"
|
|
)
|
|
|
|
func (m *Monitor) collectContainersWithNodes(ctx context.Context, instanceName string, clusterName string, isCluster bool, client PVEClientInterface, nodes []proxmox.Node, nodeEffectiveStatus map[string]string) []models.Container {
|
|
startTime := time.Now()
|
|
|
|
// Channel to collect container results from each node
|
|
type nodeResult struct {
|
|
node string
|
|
containers []models.Container
|
|
templateSubjects map[string]struct{}
|
|
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++
|
|
}
|
|
}
|
|
|
|
// Capture the previous guest context once per poll cycle so fallback behavior
|
|
// is based on a consistent pre-poll snapshot.
|
|
prevGuests := m.previousGuestContextForInstance(instanceName)
|
|
prevContainerIsOCI := prevGuests.containerOCIByVMID
|
|
|
|
log.Debug().
|
|
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
|
|
nodeTemplateSubjects := make(map[string]struct{})
|
|
|
|
// Process each container
|
|
for _, container := range containers {
|
|
// Skip templates
|
|
if container.Template == 1 {
|
|
if key := pveBackupTemplateSubjectKey(instanceName, "lxc", n.Node, int(container.VMID)); key != "" {
|
|
nodeTemplateSubjects[key] = struct{}{}
|
|
}
|
|
continue
|
|
}
|
|
|
|
// Parse tags
|
|
var tags []string
|
|
if container.Tags != "" {
|
|
tags = strings.Split(container.Tags, ";")
|
|
}
|
|
|
|
// Generate canonical guest ID: instance:node:vmid
|
|
guestID := makeGuestID(instanceName, n.Node, int(container.VMID))
|
|
|
|
sampleTime := time.Now()
|
|
counterObservedAt := observedAtOr(container.ObservedAt, sampleTime)
|
|
currentMetrics := IOMetrics{
|
|
DiskRead: int64(container.DiskRead),
|
|
DiskWrite: int64(container.DiskWrite),
|
|
NetworkIn: int64(container.NetIn),
|
|
NetworkOut: int64(container.NetOut),
|
|
Timestamp: counterObservedAt,
|
|
Presence: pveCounterPresence(container.IOCounters),
|
|
ObservedAt: counterObservationTimes(counterObservedAt),
|
|
SourceUptime: uint64(container.Uptime),
|
|
}
|
|
statusSnapshot := (*proxmox.Container)(nil)
|
|
if container.Status == "running" {
|
|
statusSnapshot = m.fetchContainerStatusSnapshot(
|
|
ctx,
|
|
client,
|
|
instanceName,
|
|
n.Node,
|
|
container.Name,
|
|
int(container.VMID),
|
|
)
|
|
currentMetrics = mergeContainerRuntimeCounters(currentMetrics, statusSnapshot)
|
|
}
|
|
diskReadRate, diskWriteRate, netInRate, netOutRate := m.rateTracker.CalculateRates(
|
|
makeGuestRateKey(instanceName, "lxc", int(container.VMID)),
|
|
currentMetrics,
|
|
)
|
|
diskReadValue, diskWriteValue, networkInValue, networkOutValue, rateValidity := guestRateValues(
|
|
diskReadRate,
|
|
diskWriteRate,
|
|
netInRate,
|
|
netOutRate,
|
|
)
|
|
|
|
// Set CPU to 0 for non-running containers
|
|
cpuUsage := safeFloat(container.CPU)
|
|
if container.Status != "running" {
|
|
cpuUsage = 0
|
|
}
|
|
|
|
memTotal, memUsed, memorySource, guestRaw := m.calculateLXCMemory(ctx, instanceName, proxmox.ClusterResource{
|
|
Type: "lxc",
|
|
Node: n.Node,
|
|
Name: container.Name,
|
|
Status: container.Status,
|
|
VMID: int(container.VMID),
|
|
MaxMem: container.MaxMem,
|
|
Mem: container.Mem,
|
|
}, client)
|
|
memUsed, memorySource, _ = stabilizeGuestLowTrustMemory(
|
|
m.previousGuestSnapshot(instanceName, "lxc", n.Node, int(container.VMID)),
|
|
container.Status,
|
|
memorySource,
|
|
memTotal,
|
|
memUsed,
|
|
sampleTime,
|
|
false,
|
|
)
|
|
|
|
memTotalBytes := clampToInt64(memTotal)
|
|
memUsedBytes := clampToInt64(memUsed)
|
|
if memTotalBytes > 0 && memUsedBytes > memTotalBytes {
|
|
memUsedBytes = memTotalBytes
|
|
}
|
|
memFreeBytes := memTotalBytes - memUsedBytes
|
|
if memFreeBytes < 0 {
|
|
memFreeBytes = 0
|
|
}
|
|
memUsagePercent := safePercentage(float64(memUsedBytes), float64(memTotalBytes))
|
|
memory := models.UnavailableMemory(memTotalBytes)
|
|
if CanonicalMemorySource(memorySource) != "unavailable" {
|
|
memory = models.Memory{
|
|
Total: memTotalBytes,
|
|
Used: memUsedBytes,
|
|
Free: memFreeBytes,
|
|
Usage: memUsagePercent,
|
|
}
|
|
}
|
|
|
|
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,
|
|
Pool: strings.TrimSpace(container.Pool),
|
|
Instance: instanceName,
|
|
Status: container.Status,
|
|
Type: "lxc",
|
|
CPU: cpuUsage,
|
|
CPUs: int(container.CPUs),
|
|
Memory: memory,
|
|
Disk: models.Disk{
|
|
Total: diskTotalBytes,
|
|
Used: diskUsedBytes,
|
|
Free: diskFreeBytes,
|
|
Usage: diskUsagePercent,
|
|
},
|
|
NetworkIn: networkInValue,
|
|
NetworkOut: networkOutValue,
|
|
DiskRead: diskReadValue,
|
|
DiskWrite: diskWriteValue,
|
|
IORateValidity: rateValidity,
|
|
Uptime: int64(container.Uptime),
|
|
Template: container.Template == 1,
|
|
LastSeen: sampleTime,
|
|
Tags: tags,
|
|
}
|
|
|
|
if prevContainerIsOCI[modelContainer.VMID] {
|
|
modelContainer.IsOCI = true
|
|
modelContainer.Type = "oci"
|
|
}
|
|
|
|
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,
|
|
statusSnapshot,
|
|
)
|
|
|
|
// 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
|
|
modelContainer.IORateValidity = models.IORateValidity{
|
|
Explicit: true,
|
|
DiskRead: true,
|
|
DiskWrite: true,
|
|
NetworkIn: true,
|
|
NetworkOut: true,
|
|
}
|
|
}
|
|
|
|
// Trigger guest metadata migration if old format exists
|
|
if m.guestMetadataStore != nil {
|
|
m.guestMetadataStore.GetWithLegacyMigration(guestID, instanceName, n.Node, int(container.VMID))
|
|
}
|
|
|
|
m.recordGuestSnapshot(instanceName, modelContainer.Type, n.Node, int(container.VMID), GuestMemorySnapshot{
|
|
Name: modelContainer.Name,
|
|
Status: modelContainer.Status,
|
|
RetrievedAt: modelContainer.LastSeen,
|
|
MemorySource: memorySource,
|
|
FallbackReason: guestMemoryFallbackReason(memorySource),
|
|
Memory: modelContainer.Memory,
|
|
Raw: guestRaw,
|
|
})
|
|
|
|
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, templateSubjects: nodeTemplateSubjects}
|
|
}(node)
|
|
}
|
|
|
|
// Close channel when all goroutines complete
|
|
go func() {
|
|
wg.Wait()
|
|
close(resultChan)
|
|
}()
|
|
|
|
// Collect results from all nodes
|
|
var allContainers []models.Container
|
|
lxcTemplateSubjects := make(map[string]struct{})
|
|
successfulNodes := 0
|
|
failedNodes := 0
|
|
failedNodeNames := make(map[string]struct{})
|
|
|
|
for result := range resultChan {
|
|
if result.err != nil {
|
|
failedNodes++
|
|
failedNodeNames[result.node] = struct{}{}
|
|
} else {
|
|
successfulNodes++
|
|
allContainers = append(allContainers, result.containers...)
|
|
for key := range result.templateSubjects {
|
|
lxcTemplateSubjects[key] = struct{}{}
|
|
}
|
|
}
|
|
}
|
|
if failedNodes == 0 && successfulNodes > 0 {
|
|
m.updatePVEBackupTemplateSubjectsForType(instanceName, "lxc", lxcTemplateSubjects)
|
|
}
|
|
|
|
preservedContainers := 0
|
|
if len(failedNodeNames) > 0 {
|
|
for _, container := range prevGuests.containers {
|
|
if _, failed := failedNodeNames[container.Node]; failed {
|
|
allContainers = append(allContainers, container)
|
|
preservedContainers++
|
|
}
|
|
}
|
|
}
|
|
if preservedContainers > 0 {
|
|
log.Warn().
|
|
Str("instance", instanceName).
|
|
Int("preservedContainers", preservedContainers).
|
|
Int("failedNodes", failedNodes).
|
|
Msg("Preserved prior containers for nodes whose enumeration failed")
|
|
}
|
|
|
|
// Check Docker presence for containers that need it (new, restarted, started)
|
|
allContainers = m.CheckContainersForDocker(ctx, allContainers)
|
|
m.CollectProxmoxGuestDockerInventory(ctx, allContainers)
|
|
|
|
// Record guest metrics history for running containers (enables sparkline/trends view)
|
|
if !shouldSkipNativeMockStateMetricWrites() {
|
|
now := time.Now()
|
|
for _, ct := range allContainers {
|
|
if !guestObservedInCycle(ct.LastSeen, startTime) {
|
|
continue
|
|
}
|
|
if ct.Status != "running" {
|
|
continue
|
|
}
|
|
diskRead, diskWrite, networkIn, networkOut := guestHistoryRates(
|
|
ct.DiskRead,
|
|
ct.DiskWrite,
|
|
ct.NetworkIn,
|
|
ct.NetworkOut,
|
|
ct.IORateValidity,
|
|
)
|
|
m.recordGuestMetric(
|
|
"container",
|
|
ct.ID,
|
|
unifiedresources.ProxmoxGuestCPUPercent(ct.CPU),
|
|
historyMemoryUsage(ct.Memory),
|
|
ct.Disk.Usage,
|
|
diskRead,
|
|
diskWrite,
|
|
networkIn,
|
|
networkOut,
|
|
now,
|
|
)
|
|
}
|
|
}
|
|
|
|
duration := time.Since(startTime)
|
|
log.Debug().
|
|
Str("instance", instanceName).
|
|
Int("totalContainers", len(allContainers)).
|
|
Int("successfulNodes", successfulNodes).
|
|
Int("failedNodes", failedNodes).
|
|
Dur("duration", duration).
|
|
Msg("Parallel container polling completed")
|
|
|
|
return allContainers
|
|
}
|
|
|
|
// pollContainersWithNodes retains the focused single-kind polling entry point
|
|
// used by tests and maintenance callers. The production guest cycle uses
|
|
// collectContainersWithNodes and publishes both guest kinds atomically.
|
|
func (m *Monitor) pollContainersWithNodes(ctx context.Context, instanceName string, clusterName string, isCluster bool, client PVEClientInterface, nodes []proxmox.Node, nodeEffectiveStatus map[string]string) {
|
|
containers := m.collectContainersWithNodes(ctx, instanceName, clusterName, isCluster, client, nodes, nodeEffectiveStatus)
|
|
m.state.UpdateContainersForInstance(instanceName, containers)
|
|
}
|
|
|
|
// pollStorageWithNodes polls storage from all nodes in parallel using goroutines
|