pulse/internal/monitoring/monitor_polling_vm.go

package monitoring

import (
	"context"
	"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) collectVMsWithNodes(ctx context.Context, instanceName string, clusterName string, isCluster bool, client PVEClientInterface, nodes []proxmox.Node, nodeEffectiveStatus map[string]string) []models.VM {
	startTime := time.Now()

	type nodeResult struct {
		node             string
		vms              []models.VM
		templateSubjects map[string]struct{}
		err              error
	}

	resultChan := make(chan nodeResult, len(nodes))
	var wg sync.WaitGroup

	onlineNodes := 0
	for _, node := range nodes {
		if nodeEffectiveStatus[node.Node] == "online" {
			onlineNodes++
		}
	}

	prevGuests := m.previousGuestContextForInstance(instanceName)
	prevVMByID := prevGuests.vmsByID
	vmIDToHostAgent := prevGuests.hostAgentsByVMID

	log.Debug().
		Str("instance", instanceName).
		Int("totalNodes", len(nodes)).
		Int("onlineNodes", onlineNodes).
		Msg("Starting parallel VM polling")

	for _, node := range 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()
			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
			}

			nodeVMs, nodeTemplateSubjects := m.pollNodeVMsWithClusterResourceBuilder(ctx, instanceName, n.Node, vms, client, prevVMByID, vmIDToHostAgent)
			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, templateSubjects: nodeTemplateSubjects}
		}(node)
	}

	go func() {
		wg.Wait()
		close(resultChan)
	}()

	var allVMs []models.VM
	qemuTemplateSubjects := 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{}{}
			continue
		}
		successfulNodes++
		allVMs = append(allVMs, result.vms...)
		for key := range result.templateSubjects {
			qemuTemplateSubjects[key] = struct{}{}
		}
	}
	if failedNodes == 0 && successfulNodes > 0 {
		m.updatePVEBackupTemplateSubjectsForType(instanceName, "qemu", qemuTemplateSubjects)
	}

	preservedVMs := 0
	if len(failedNodeNames) > 0 {
		for _, vm := range prevGuests.vms {
			if _, failed := failedNodeNames[vm.Node]; failed {
				allVMs = append(allVMs, vm)
				preservedVMs++
			}
		}
	}
	if preservedVMs > 0 {
		log.Warn().
			Str("instance", instanceName).
			Int("preservedVMs", preservedVMs).
			Int("failedNodes", failedNodes).
			Msg("Preserved prior VMs for nodes whose enumeration failed")
	}

	if !shouldSkipNativeMockStateMetricWrites() {
		now := time.Now()
		for _, vm := range allVMs {
			if !guestObservedInCycle(vm.LastSeen, startTime) {
				continue
			}
			if vm.Status != "running" {
				continue
			}
			diskRead, diskWrite, networkIn, networkOut := guestHistoryRates(
				vm.DiskRead,
				vm.DiskWrite,
				vm.NetworkIn,
				vm.NetworkOut,
				vm.IORateValidity,
			)
			m.recordGuestMetric(
				"vm",
				vm.ID,
				unifiedresources.ProxmoxGuestCPUPercent(vm.CPU),
				historyMemoryUsage(vm.Memory),
				historyMemoryUsed(vm.Memory),
				vm.Disk.Usage,
				diskRead,
				diskWrite,
				networkIn,
				networkOut,
				now,
			)
		}
	}

	duration := time.Since(startTime)
	log.Debug().
		Str("instance", instanceName).
		Int("totalVMs", len(allVMs)).
		Int("successfulNodes", successfulNodes).
		Int("failedNodes", failedNodes).
		Dur("duration", duration).
		Msg("Parallel VM polling completed")

	return allVMs
}

// pollVMsWithNodes retains the focused single-kind polling entry point used by
// tests and maintenance callers. The production guest cycle uses
// collectVMsWithNodes and publishes both guest kinds atomically.
func (m *Monitor) pollVMsWithNodes(ctx context.Context, instanceName string, clusterName string, isCluster bool, client PVEClientInterface, nodes []proxmox.Node, nodeEffectiveStatus map[string]string) {
	vms := m.collectVMsWithNodes(ctx, instanceName, clusterName, isCluster, client, nodes, nodeEffectiveStatus)
	m.state.UpdateVMsForInstance(instanceName, vms)
}