From 4eb8bed9b508fba342b9bbd5f6b395a6b8bd2380 Mon Sep 17 00:00:00 2001 From: rcourtman Date: Wed, 22 Oct 2025 07:34:32 +0000 Subject: [PATCH] Fix initial setup caching and container discovery defaults --- .../src/components/FirstRunSetup.tsx | 16 +- .../Settings/QuickSecuritySetup.tsx | 13 + internal/discovery/config_override.go | 278 +++--- internal/monitoring/monitor.go | 803 +++++++++--------- 4 files changed, 582 insertions(+), 528 deletions(-) diff --git a/frontend-modern/src/components/FirstRunSetup.tsx b/frontend-modern/src/components/FirstRunSetup.tsx index 06c7420..4d49ce5 100644 --- a/frontend-modern/src/components/FirstRunSetup.tsx +++ b/frontend-modern/src/components/FirstRunSetup.tsx @@ -131,9 +131,23 @@ export const FirstRunSetup: Component = () => { // Save credentials for display setSavedUsername(username()); - setSavedPassword(useCustomPassword() ? password() : generatedPassword()); + setSavedPassword(finalPassword); setSavedToken(token); + // Clear any cached credentials from prior sessions so a reload doesn't auto-submit again + try { + sessionStorage.removeItem('pulse_auth'); + sessionStorage.removeItem('pulse_auth_user'); + } catch (storageError) { + console.warn('Unable to clear cached auth session storage', storageError); + } + + try { + localStorage.removeItem('apiToken'); + } catch (storageError) { + console.warn('Unable to clear cached API token', storageError); + } + const bootstrapRecord: APITokenRecord = { id: 'bootstrap-token', name: 'Bootstrap token', diff --git a/frontend-modern/src/components/Settings/QuickSecuritySetup.tsx b/frontend-modern/src/components/Settings/QuickSecuritySetup.tsx index 623b440..5e48679 100644 --- a/frontend-modern/src/components/Settings/QuickSecuritySetup.tsx +++ b/frontend-modern/src/components/Settings/QuickSecuritySetup.tsx @@ -117,6 +117,19 @@ export const QuickSecuritySetup: Component = (props) => setCredentials(newCredentials); setShowCredentials(true); + try { + sessionStorage.removeItem('pulse_auth'); + sessionStorage.removeItem('pulse_auth_user'); + } catch (storageError) { + console.warn('Unable to clear cached auth session storage', storageError); + } + + try { + localStorage.removeItem('apiToken'); + } catch (storageError) { + console.warn('Unable to clear cached API token', storageError); + } + // Show success message showSuccess( isRotation diff --git a/internal/discovery/config_override.go b/internal/discovery/config_override.go index 788f70c..08d1f82 100644 --- a/internal/discovery/config_override.go +++ b/internal/discovery/config_override.go @@ -1,64 +1,64 @@ package discovery import ( - "fmt" - "net" - "strings" - "time" + "fmt" + "net" + "strings" + "time" - "github.com/rcourtman/pulse-go-rewrite/internal/config" - pkgdiscovery "github.com/rcourtman/pulse-go-rewrite/pkg/discovery" - "github.com/rcourtman/pulse-go-rewrite/pkg/discovery/envdetect" + "github.com/rcourtman/pulse-go-rewrite/internal/config" + pkgdiscovery "github.com/rcourtman/pulse-go-rewrite/pkg/discovery" + "github.com/rcourtman/pulse-go-rewrite/pkg/discovery/envdetect" ) // BuildScanner creates a discovery scanner configured using the supplied discovery config. func BuildScanner(cfg config.DiscoveryConfig) (*pkgdiscovery.Scanner, error) { - cfg = config.NormalizeDiscoveryConfig(cfg) + cfg = config.NormalizeDiscoveryConfig(cfg) - profile, err := envdetect.DetectEnvironment() - if err != nil { - return nil, err - } + profile, err := envdetect.DetectEnvironment() + if err != nil { + return nil, err + } - ApplyConfigToProfile(profile, cfg) - return pkgdiscovery.NewScannerWithProfile(profile), nil + ApplyConfigToProfile(profile, cfg) + return pkgdiscovery.NewScannerWithProfile(profile), nil } // ApplyConfigToProfile mutates the supplied environment profile according to the discovery config. func ApplyConfigToProfile(profile *envdetect.EnvironmentProfile, cfg config.DiscoveryConfig) { - if profile == nil { - return - } + if profile == nil { + return + } - // Environment override - if env, ok := environmentFromOverride(cfg.EnvironmentOverride); ok { - profile.Type = env - filterPhasesForEnvironment(profile, env) - } else if cfg.EnvironmentOverride != "" && strings.ToLower(cfg.EnvironmentOverride) != "auto" { - profile.Warnings = append(profile.Warnings, fmt.Sprintf("Unknown environment override: %s", cfg.EnvironmentOverride)) - } + // Environment override + if env, ok := environmentFromOverride(cfg.EnvironmentOverride); ok { + profile.Type = env + filterPhasesForEnvironment(profile, env) + } else if cfg.EnvironmentOverride != "" && strings.ToLower(cfg.EnvironmentOverride) != "auto" { + profile.Warnings = append(profile.Warnings, fmt.Sprintf("Unknown environment override: %s", cfg.EnvironmentOverride)) + } - // Apply subnet blocklist - blocked := parseCIDRMap(cfg.SubnetBlocklist, &profile.Warnings) - if len(blocked) > 0 { - var filtered []envdetect.SubnetPhase - for _, phase := range profile.Phases { - var kept []net.IPNet - for _, subnet := range phase.Subnets { - if _, blocked := blocked[subnet.String()]; blocked { - continue - } - kept = append(kept, subnet) - } - if len(kept) > 0 { - phase.Subnets = kept - filtered = append(filtered, phase) - } - } - profile.Phases = filtered - } + // Apply subnet blocklist + blocked := parseCIDRMap(cfg.SubnetBlocklist, &profile.Warnings) + if len(blocked) > 0 { + var filtered []envdetect.SubnetPhase + for _, phase := range profile.Phases { + var kept []net.IPNet + for _, subnet := range phase.Subnets { + if _, blocked := blocked[subnet.String()]; blocked { + continue + } + kept = append(kept, subnet) + } + if len(kept) > 0 { + phase.Subnets = kept + filtered = append(filtered, phase) + } + } + profile.Phases = filtered + } - // Apply subnet allowlist as highest priority phase + // Apply subnet allowlist as highest priority phase if len(cfg.SubnetAllowlist) > 0 { allowlist := parseCIDRs(cfg.SubnetAllowlist, &profile.Warnings) if len(allowlist) > 0 { @@ -79,104 +79,126 @@ func ApplyConfigToProfile(profile *envdetect.EnvironmentProfile, cfg config.Disc Subnets: allowlist, Confidence: 1.0, Priority: 0, - } - profile.Phases = append([]envdetect.SubnetPhase{allowPhase}, profile.Phases...) - } - } + } + profile.Phases = append([]envdetect.SubnetPhase{allowPhase}, profile.Phases...) + } + } - // Override scan policy - if cfg.MaxHostsPerScan > 0 { - profile.Policy.MaxHostsPerScan = cfg.MaxHostsPerScan - } - if cfg.MaxConcurrent > 0 { - profile.Policy.MaxConcurrent = cfg.MaxConcurrent - } - profile.Policy.EnableReverseDNS = cfg.EnableReverseDNS - profile.Policy.ScanGateways = cfg.ScanGateways + if len(cfg.SubnetAllowlist) == 0 && shouldPruneContainerNetworks(profile.Type) { + pruned := make([]envdetect.SubnetPhase, 0, len(profile.Phases)) + for _, phase := range profile.Phases { + if isLikelyContainerPhase(phase.Name) { + continue + } + pruned = append(pruned, phase) + } + if len(pruned) > 0 { + profile.Phases = pruned + } + } - if cfg.DialTimeout > 0 { - profile.Policy.DialTimeout = time.Duration(cfg.DialTimeout) * time.Millisecond - } - if cfg.HTTPTimeout > 0 { - profile.Policy.HTTPTimeout = time.Duration(cfg.HTTPTimeout) * time.Millisecond - } + // Override scan policy + if cfg.MaxHostsPerScan > 0 { + profile.Policy.MaxHostsPerScan = cfg.MaxHostsPerScan + } + if cfg.MaxConcurrent > 0 { + profile.Policy.MaxConcurrent = cfg.MaxConcurrent + } + profile.Policy.EnableReverseDNS = cfg.EnableReverseDNS + profile.Policy.ScanGateways = cfg.ScanGateways + + if cfg.DialTimeout > 0 { + profile.Policy.DialTimeout = time.Duration(cfg.DialTimeout) * time.Millisecond + } + if cfg.HTTPTimeout > 0 { + profile.Policy.HTTPTimeout = time.Duration(cfg.HTTPTimeout) * time.Millisecond + } +} + +func shouldPruneContainerNetworks(env envdetect.Environment) bool { + return env == envdetect.DockerBridge || env == envdetect.LXCUnprivileged +} + +func isLikelyContainerPhase(name string) bool { + name = strings.ToLower(strings.TrimSpace(name)) + return strings.Contains(name, "container") } func parseCIDRs(values []string, warnings *[]string) []net.IPNet { - var subnets []net.IPNet - for _, value := range values { - value = strings.TrimSpace(value) - if value == "" { - continue - } - _, ipNet, err := net.ParseCIDR(value) - if err != nil { - if warnings != nil { - *warnings = append(*warnings, fmt.Sprintf("Invalid CIDR '%s' ignored", value)) - } - continue - } - subnets = append(subnets, *ipNet) - } - return subnets + var subnets []net.IPNet + for _, value := range values { + value = strings.TrimSpace(value) + if value == "" { + continue + } + _, ipNet, err := net.ParseCIDR(value) + if err != nil { + if warnings != nil { + *warnings = append(*warnings, fmt.Sprintf("Invalid CIDR '%s' ignored", value)) + } + continue + } + subnets = append(subnets, *ipNet) + } + return subnets } func parseCIDRMap(values []string, warnings *[]string) map[string]struct{} { - cidrs := parseCIDRs(values, warnings) - result := make(map[string]struct{}, len(cidrs)) - for _, cidr := range cidrs { - result[cidr.String()] = struct{}{} - } - return result + cidrs := parseCIDRs(values, warnings) + result := make(map[string]struct{}, len(cidrs)) + for _, cidr := range cidrs { + result[cidr.String()] = struct{}{} + } + return result } func environmentFromOverride(value string) (envdetect.Environment, bool) { - normalized := strings.ToLower(strings.TrimSpace(value)) - switch normalized { - case "", "auto": - return envdetect.Unknown, false - case "native": - return envdetect.Native, true - case "docker_host": - return envdetect.DockerHost, true - case "docker_bridge": - return envdetect.DockerBridge, true - case "lxc_privileged": - return envdetect.LXCPrivileged, true - case "lxc_unprivileged": - return envdetect.LXCUnprivileged, true - default: - return envdetect.Unknown, false - } + normalized := strings.ToLower(strings.TrimSpace(value)) + switch normalized { + case "", "auto": + return envdetect.Unknown, false + case "native": + return envdetect.Native, true + case "docker_host": + return envdetect.DockerHost, true + case "docker_bridge": + return envdetect.DockerBridge, true + case "lxc_privileged": + return envdetect.LXCPrivileged, true + case "lxc_unprivileged": + return envdetect.LXCUnprivileged, true + default: + return envdetect.Unknown, false + } } func filterPhasesForEnvironment(profile *envdetect.EnvironmentProfile, env envdetect.Environment) { - if len(profile.Phases) == 0 { - return - } + if len(profile.Phases) == 0 { + return + } - var keep []envdetect.SubnetPhase - for _, phase := range profile.Phases { - name := strings.ToLower(phase.Name) - switch env { - case envdetect.Native, envdetect.DockerHost, envdetect.LXCPrivileged: - if strings.Contains(name, "local") || strings.Contains(name, "host") { - keep = append(keep, phase) - } - case envdetect.DockerBridge: - if strings.Contains(name, "container") || strings.Contains(name, "inferred") || strings.Contains(name, "host") { - keep = append(keep, phase) - } - case envdetect.LXCUnprivileged: - if strings.Contains(name, "lxc") || strings.Contains(name, "container") || strings.Contains(name, "parent") { - keep = append(keep, phase) - } - default: - keep = append(keep, phase) - } - } + var keep []envdetect.SubnetPhase + for _, phase := range profile.Phases { + name := strings.ToLower(phase.Name) + switch env { + case envdetect.Native, envdetect.DockerHost, envdetect.LXCPrivileged: + if strings.Contains(name, "local") || strings.Contains(name, "host") { + keep = append(keep, phase) + } + case envdetect.DockerBridge: + if strings.Contains(name, "container") || strings.Contains(name, "inferred") || strings.Contains(name, "host") { + keep = append(keep, phase) + } + case envdetect.LXCUnprivileged: + if strings.Contains(name, "lxc") || strings.Contains(name, "container") || strings.Contains(name, "parent") { + keep = append(keep, phase) + } + default: + keep = append(keep, phase) + } + } - if len(keep) > 0 { - profile.Phases = keep - } + if len(keep) > 0 { + profile.Phases = keep + } } diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index b02ced6..b26da03 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -111,7 +111,7 @@ func mergeNVMeTempsIntoDisks(disks []models.PhysicalDisk, nodes []models.Node) [ } nvmeTempsByNode := make(map[string][]models.NVMeTemp) - for _, node := range nodes { + for _, node := range nodes { if node.Temperature == nil || !node.Temperature.Available || len(node.Temperature.NVMe) == 0 { continue } @@ -371,7 +371,6 @@ type InstanceHealth struct { DeadLetter InstanceDLQ `json:"deadLetter"` } - func schedulerKey(instanceType InstanceType, name string) string { return string(instanceType) + "::" + name } @@ -386,60 +385,60 @@ func timePtr(t time.Time) *time.Time { // Monitor handles all monitoring operations type Monitor struct { - config *config.Config - state *models.State - pveClients map[string]PVEClientInterface - pbsClients map[string]*pbs.Client - pmgClients map[string]*pmg.Client - pollMetrics *PollMetrics - scheduler *AdaptiveScheduler - stalenessTracker *StalenessTracker - taskQueue *TaskQueue - circuitBreakers map[string]*circuitBreaker - deadLetterQueue *TaskQueue - failureCounts map[string]int - lastOutcome map[string]taskOutcome - backoffCfg backoffConfig - rng *rand.Rand - maxRetryAttempts int - tempCollector *TemperatureCollector // SSH-based temperature collector - mu sync.RWMutex - startTime time.Time - rateTracker *RateTracker - metricsHistory *MetricsHistory - alertManager *alerts.Manager - notificationMgr *notifications.NotificationManager - configPersist *config.ConfigPersistence - discoveryService *discovery.Service // Background discovery service - activePollCount int32 // Number of active polling operations - pollCounter int64 // Counter for polling cycles - authFailures map[string]int // Track consecutive auth failures per node - lastAuthAttempt map[string]time.Time // Track last auth attempt time - lastClusterCheck map[string]time.Time // Track last cluster check for standalone nodes - lastPhysicalDiskPoll map[string]time.Time // Track last physical disk poll time per instance - lastPVEBackupPoll map[string]time.Time // Track last PVE backup poll per instance - lastPBSBackupPoll map[string]time.Time // Track last PBS backup poll per instance - persistence *config.ConfigPersistence // Add persistence for saving updated configs - pbsBackupPollers map[string]bool // Track PBS backup polling goroutines per instance - runtimeCtx context.Context // Context used while monitor is running - wsHub *websocket.Hub // Hub used for broadcasting state - diagMu sync.RWMutex // Protects diagnostic snapshot maps - nodeSnapshots map[string]NodeMemorySnapshot - guestSnapshots map[string]GuestMemorySnapshot - rrdCacheMu sync.RWMutex // Protects RRD memavailable cache - nodeRRDMemCache map[string]rrdMemCacheEntry - removedDockerHosts map[string]time.Time // Track deliberately removed Docker hosts (ID -> removal time) - dockerCommands map[string]*dockerHostCommand - dockerCommandIndex map[string]string - guestMetadataMu sync.RWMutex - guestMetadataCache map[string]guestMetadataCacheEntry - executor PollExecutor - breakerBaseRetry time.Duration - breakerMaxDelay time.Duration + config *config.Config + state *models.State + pveClients map[string]PVEClientInterface + pbsClients map[string]*pbs.Client + pmgClients map[string]*pmg.Client + pollMetrics *PollMetrics + scheduler *AdaptiveScheduler + stalenessTracker *StalenessTracker + taskQueue *TaskQueue + circuitBreakers map[string]*circuitBreaker + deadLetterQueue *TaskQueue + failureCounts map[string]int + lastOutcome map[string]taskOutcome + backoffCfg backoffConfig + rng *rand.Rand + maxRetryAttempts int + tempCollector *TemperatureCollector // SSH-based temperature collector + mu sync.RWMutex + startTime time.Time + rateTracker *RateTracker + metricsHistory *MetricsHistory + alertManager *alerts.Manager + notificationMgr *notifications.NotificationManager + configPersist *config.ConfigPersistence + discoveryService *discovery.Service // Background discovery service + activePollCount int32 // Number of active polling operations + pollCounter int64 // Counter for polling cycles + authFailures map[string]int // Track consecutive auth failures per node + lastAuthAttempt map[string]time.Time // Track last auth attempt time + lastClusterCheck map[string]time.Time // Track last cluster check for standalone nodes + lastPhysicalDiskPoll map[string]time.Time // Track last physical disk poll time per instance + lastPVEBackupPoll map[string]time.Time // Track last PVE backup poll per instance + lastPBSBackupPoll map[string]time.Time // Track last PBS backup poll per instance + persistence *config.ConfigPersistence // Add persistence for saving updated configs + pbsBackupPollers map[string]bool // Track PBS backup polling goroutines per instance + runtimeCtx context.Context // Context used while monitor is running + wsHub *websocket.Hub // Hub used for broadcasting state + diagMu sync.RWMutex // Protects diagnostic snapshot maps + nodeSnapshots map[string]NodeMemorySnapshot + guestSnapshots map[string]GuestMemorySnapshot + rrdCacheMu sync.RWMutex // Protects RRD memavailable cache + nodeRRDMemCache map[string]rrdMemCacheEntry + removedDockerHosts map[string]time.Time // Track deliberately removed Docker hosts (ID -> removal time) + dockerCommands map[string]*dockerHostCommand + dockerCommandIndex map[string]string + guestMetadataMu sync.RWMutex + guestMetadataCache map[string]guestMetadataCacheEntry + executor PollExecutor + breakerBaseRetry time.Duration + breakerMaxDelay time.Duration breakerHalfOpenWindow time.Duration - instanceInfoCache map[string]*instanceInfo - pollStatusMap map[string]*pollStatus - dlqInsightMap map[string]*dlqInsight + instanceInfoCache map[string]*instanceInfo + pollStatusMap map[string]*pollStatus + dlqInsightMap map[string]*dlqInsight } type rrdMemCacheEntry struct { @@ -530,10 +529,10 @@ type guestMetadataCacheEntry struct { } type taskOutcome struct { - success bool - transient bool - err error - recordedAt time.Time + success bool + transient bool + err error + recordedAt time.Time } func (m *Monitor) getNodeRRDMemAvailable(ctx context.Context, client PVEClientInterface, nodeName string) (uint64, error) { @@ -1480,8 +1479,8 @@ func New(cfg *config.Config) (*Monitor, error) { taskQueue := NewTaskQueue() deadLetterQueue := NewTaskQueue() breakers := make(map[string]*circuitBreaker) -failureCounts := make(map[string]int) -lastOutcome := make(map[string]taskOutcome) + failureCounts := make(map[string]int) + lastOutcome := make(map[string]taskOutcome) backoff := backoffConfig{ Initial: 5 * time.Second, Multiplier: 2, @@ -1893,7 +1892,6 @@ func (m *Monitor) buildInstanceInfoCache(cfg *config.Config) { } } - func (m *Monitor) getExecutor() PollExecutor { m.mu.RLock() exec := m.executor @@ -1999,12 +1997,12 @@ func (m *Monitor) Start(ctx context.Context, wsHub *websocket.Hub) { // Create separate tickers for polling and broadcasting // Hardcoded to 10 seconds since Proxmox updates cluster/resources every 10 seconds -const pollingInterval = 10 * time.Second + const pollingInterval = 10 * time.Second -workerCount := len(m.pveClients) + len(m.pbsClients) + len(m.pmgClients) -m.startTaskWorkers(ctx, workerCount) + workerCount := len(m.pveClients) + len(m.pbsClients) + len(m.pmgClients) + m.startTaskWorkers(ctx, workerCount) -pollTicker := time.NewTicker(pollingInterval) + pollTicker := time.NewTicker(pollingInterval) defer pollTicker.Stop() broadcastTicker := time.NewTicker(pollingInterval) @@ -2694,30 +2692,30 @@ func (m *Monitor) allowExecution(task ScheduledTask) bool { } func (m *Monitor) ensureBreaker(key string) *circuitBreaker { - m.mu.Lock() - defer m.mu.Unlock() + m.mu.Lock() + defer m.mu.Unlock() - if m.circuitBreakers == nil { - m.circuitBreakers = make(map[string]*circuitBreaker) - } - if breaker, ok := m.circuitBreakers[key]; ok { - return breaker - } - baseRetry := m.breakerBaseRetry - if baseRetry <= 0 { - baseRetry = 5 * time.Second - } - maxDelay := m.breakerMaxDelay - if maxDelay <= 0 { - maxDelay = 5 * time.Minute - } - halfOpen := m.breakerHalfOpenWindow - if halfOpen <= 0 { - halfOpen = 30 * time.Second - } - breaker := newCircuitBreaker(3, baseRetry, maxDelay, halfOpen) - m.circuitBreakers[key] = breaker - return breaker + if m.circuitBreakers == nil { + m.circuitBreakers = make(map[string]*circuitBreaker) + } + if breaker, ok := m.circuitBreakers[key]; ok { + return breaker + } + baseRetry := m.breakerBaseRetry + if baseRetry <= 0 { + baseRetry = 5 * time.Second + } + maxDelay := m.breakerMaxDelay + if maxDelay <= 0 { + maxDelay = 5 * time.Minute + } + halfOpen := m.breakerHalfOpenWindow + if halfOpen <= 0 { + halfOpen = 30 * time.Second + } + breaker := newCircuitBreaker(3, baseRetry, maxDelay, halfOpen) + m.circuitBreakers[key] = breaker + return breaker } func (m *Monitor) recordTaskResult(instanceType InstanceType, instance string, pollErr error) { @@ -2792,13 +2790,13 @@ func (m *Monitor) recordTaskResult(instanceType InstanceType, instance string, p // SchedulerHealthResponse contains complete scheduler health data for API exposure. type SchedulerHealthResponse struct { - UpdatedAt time.Time `json:"updatedAt"` - Enabled bool `json:"enabled"` - Queue QueueSnapshot `json:"queue"` - DeadLetter DeadLetterSnapshot `json:"deadLetter"` - Breakers []BreakerSnapshot `json:"breakers,omitempty"` - Staleness []StalenessSnapshot `json:"staleness,omitempty"` - Instances []InstanceHealth `json:"instances"` + UpdatedAt time.Time `json:"updatedAt"` + Enabled bool `json:"enabled"` + Queue QueueSnapshot `json:"queue"` + DeadLetter DeadLetterSnapshot `json:"deadLetter"` + Breakers []BreakerSnapshot `json:"breakers,omitempty"` + Staleness []StalenessSnapshot `json:"staleness,omitempty"` + Instances []InstanceHealth `json:"instances"` } // DeadLetterSnapshot contains dead-letter queue data. @@ -3667,52 +3665,58 @@ func (m *Monitor) pollPVEInstance(ctx context.Context, instanceName string, clie 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" - } + 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) - } + 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) - } + 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 = fmt.Errorf(nodeErrReason) - } + 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(), - }) - } + m.pollMetrics.RecordNodeResult(NodePollResult{ + InstanceName: instanceName, + InstanceType: "pve", + NodeName: nodeNameLabel, + Success: success, + Error: nodeErr, + StartTime: nodeStart, + EndTime: time.Now(), + }) + } - modelNodes = append(modelNodes, modelNode) + modelNodes = append(modelNodes, modelNode) } if len(modelNodes) == 0 && len(prevInstanceNodes) > 0 { @@ -4138,19 +4142,19 @@ func (m *Monitor) pollPVEInstance(ctx context.Context, instanceName string, clie m.mu.RUnlock() shouldPoll, reason, newLast := m.shouldRunBackupPoll(lastPoll, now) - if !shouldPoll { - if reason != "" { - log.Debug(). - Str("instance", instanceName). - Str("reason", reason). - Msg("Skipping PVE backup polling this cycle") - } - } else { - select { - case <-ctx.Done(): - pollErr = ctx.Err() - return - default: + if !shouldPoll { + if reason != "" { + log.Debug(). + Str("instance", instanceName). + Str("reason", reason). + Msg("Skipping PVE backup polling this cycle") + } + } else { + select { + case <-ctx.Done(): + pollErr = ctx.Err() + return + default: m.mu.Lock() m.lastPVEBackupPoll[instanceName] = newLast m.mu.Unlock() @@ -4939,8 +4943,8 @@ func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, clie } defer m.recordTaskResult(InstanceTypePBS, instanceName, pollErr) - // Check if context is cancelled - select { + // Check if context is cancelled + select { case <-ctx.Done(): pollErr = ctx.Err() if debugEnabled { @@ -4954,285 +4958,286 @@ func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, clie log.Debug().Str("instance", instanceName).Msg("Polling PBS instance") } - // Get instance config - var instanceCfg *config.PBSInstance - for _, cfg := range m.config.PBSInstances { - if cfg.Name == instanceName { - instanceCfg = &cfg - if debugEnabled { - log.Debug(). - Str("instance", instanceName). - Bool("monitorDatastores", cfg.MonitorDatastores). - Msg("Found PBS instance config") - } - break - } - } - if instanceCfg == nil { - log.Error().Str("instance", instanceName).Msg("PBS instance config not found") - return - } + // Get instance config + var instanceCfg *config.PBSInstance + for _, cfg := range m.config.PBSInstances { + if cfg.Name == instanceName { + instanceCfg = &cfg + if debugEnabled { + log.Debug(). + Str("instance", instanceName). + Bool("monitorDatastores", cfg.MonitorDatastores). + Msg("Found PBS instance config") + } + break + } + } + if instanceCfg == nil { + log.Error().Str("instance", instanceName).Msg("PBS instance config not found") + return + } - // Initialize PBS instance with default values - pbsInst := models.PBSInstance{ - ID: "pbs-" + instanceName, - Name: instanceName, - Host: instanceCfg.Host, - Status: "offline", - Version: "unknown", - ConnectionHealth: "unhealthy", - LastSeen: time.Now(), - } + // Initialize PBS instance with default values + pbsInst := models.PBSInstance{ + ID: "pbs-" + instanceName, + Name: instanceName, + Host: instanceCfg.Host, + Status: "offline", + Version: "unknown", + ConnectionHealth: "unhealthy", + LastSeen: time.Now(), + } - // Try to get version first - version, versionErr := client.GetVersion(ctx) - if versionErr == nil { - pbsInst.Status = "online" - pbsInst.Version = version.Version - pbsInst.ConnectionHealth = "healthy" - m.resetAuthFailures(instanceName, "pbs") - m.state.SetConnectionHealth("pbs-"+instanceName, true) + // Try to get version first + version, versionErr := client.GetVersion(ctx) + if versionErr == nil { + pbsInst.Status = "online" + pbsInst.Version = version.Version + pbsInst.ConnectionHealth = "healthy" + m.resetAuthFailures(instanceName, "pbs") + m.state.SetConnectionHealth("pbs-"+instanceName, true) - if debugEnabled { - log.Debug(). - Str("instance", instanceName). - Str("version", version.Version). - Bool("monitorDatastores", instanceCfg.MonitorDatastores). - Msg("PBS version retrieved successfully") - } + if debugEnabled { + log.Debug(). + Str("instance", instanceName). + Str("version", version.Version). + Bool("monitorDatastores", instanceCfg.MonitorDatastores). + Msg("PBS version retrieved successfully") + } } else { - if debugEnabled { - log.Debug().Err(versionErr).Str("instance", instanceName).Msg("Failed to get PBS version, trying fallback") - } + if debugEnabled { + log.Debug().Err(versionErr).Str("instance", instanceName).Msg("Failed to get PBS version, trying fallback") + } - ctx2, cancel2 := context.WithTimeout(context.Background(), 10*time.Second) - defer cancel2() - _, datastoreErr := client.GetDatastores(ctx2) - if datastoreErr == nil { - pbsInst.Status = "online" - pbsInst.Version = "connected" - pbsInst.ConnectionHealth = "healthy" - m.resetAuthFailures(instanceName, "pbs") - m.state.SetConnectionHealth("pbs-"+instanceName, true) + ctx2, cancel2 := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel2() + _, datastoreErr := client.GetDatastores(ctx2) + if datastoreErr == nil { + pbsInst.Status = "online" + pbsInst.Version = "connected" + pbsInst.ConnectionHealth = "healthy" + m.resetAuthFailures(instanceName, "pbs") + m.state.SetConnectionHealth("pbs-"+instanceName, true) - log.Info(). - Str("instance", instanceName). - Msg("PBS connected (version unavailable but datastores accessible)") - } else { - pbsInst.Status = "offline" - pbsInst.ConnectionHealth = "error" - monErr := errors.WrapConnectionError("get_pbs_version", instanceName, versionErr) - log.Error().Err(monErr).Str("instance", instanceName).Msg("Failed to connect to PBS") - m.state.SetConnectionHealth("pbs-"+instanceName, false) + log.Info(). + Str("instance", instanceName). + Msg("PBS connected (version unavailable but datastores accessible)") + } else { + pbsInst.Status = "offline" + pbsInst.ConnectionHealth = "error" + monErr := errors.WrapConnectionError("get_pbs_version", instanceName, versionErr) + log.Error().Err(monErr).Str("instance", instanceName).Msg("Failed to connect to PBS") + m.state.SetConnectionHealth("pbs-"+instanceName, false) - if errors.IsAuthError(versionErr) || errors.IsAuthError(datastoreErr) { - m.recordAuthFailure(instanceName, "pbs") - return - } - } - } + if errors.IsAuthError(versionErr) || errors.IsAuthError(datastoreErr) { + m.recordAuthFailure(instanceName, "pbs") + return + } + } + } - // Get node status (CPU, memory, etc.) - nodeStatus, err := client.GetNodeStatus(ctx) - if err != nil { - if debugEnabled { - log.Debug().Err(err).Str("instance", instanceName).Msg("Could not get PBS node status (may need Sys.Audit permission)") - } - } else if nodeStatus != nil { - pbsInst.CPU = nodeStatus.CPU - if nodeStatus.Memory.Total > 0 { - pbsInst.Memory = float64(nodeStatus.Memory.Used) / float64(nodeStatus.Memory.Total) * 100 - pbsInst.MemoryUsed = nodeStatus.Memory.Used - pbsInst.MemoryTotal = nodeStatus.Memory.Total - } - pbsInst.Uptime = nodeStatus.Uptime + // Get node status (CPU, memory, etc.) + nodeStatus, err := client.GetNodeStatus(ctx) + if err != nil { + if debugEnabled { + log.Debug().Err(err).Str("instance", instanceName).Msg("Could not get PBS node status (may need Sys.Audit permission)") + } + } else if nodeStatus != nil { + pbsInst.CPU = nodeStatus.CPU + if nodeStatus.Memory.Total > 0 { + pbsInst.Memory = float64(nodeStatus.Memory.Used) / float64(nodeStatus.Memory.Total) * 100 + pbsInst.MemoryUsed = nodeStatus.Memory.Used + pbsInst.MemoryTotal = nodeStatus.Memory.Total + } + pbsInst.Uptime = nodeStatus.Uptime - log.Debug(). - Str("instance", instanceName). - Float64("cpu", pbsInst.CPU). - Float64("memory", pbsInst.Memory). - Int64("uptime", pbsInst.Uptime). - Msg("PBS node status retrieved") - } + log.Debug(). + Str("instance", instanceName). + Float64("cpu", pbsInst.CPU). + Float64("memory", pbsInst.Memory). + Int64("uptime", pbsInst.Uptime). + Msg("PBS node status retrieved") + } - // Poll datastores if enabled - if instanceCfg.MonitorDatastores { - datastores, err := client.GetDatastores(ctx) - if err != nil { - monErr := errors.WrapAPIError("get_datastores", instanceName, err, 0) - log.Error().Err(monErr).Str("instance", instanceName).Msg("Failed to get datastores") - } else { - log.Info(). - Str("instance", instanceName). - Int("count", len(datastores)). - Msg("Got PBS datastores") + // Poll datastores if enabled + if instanceCfg.MonitorDatastores { + datastores, err := client.GetDatastores(ctx) + if err != nil { + monErr := errors.WrapAPIError("get_datastores", instanceName, err, 0) + log.Error().Err(monErr).Str("instance", instanceName).Msg("Failed to get datastores") + } else { + log.Info(). + Str("instance", instanceName). + Int("count", len(datastores)). + Msg("Got PBS datastores") - for _, ds := range datastores { - total := ds.Total - if total == 0 && ds.TotalSpace > 0 { - total = ds.TotalSpace - } - used := ds.Used - if used == 0 && ds.UsedSpace > 0 { - used = ds.UsedSpace - } - avail := ds.Avail - if avail == 0 && ds.AvailSpace > 0 { - avail = ds.AvailSpace - } - if total == 0 && used > 0 && avail > 0 { - total = used + avail - } + for _, ds := range datastores { + total := ds.Total + if total == 0 && ds.TotalSpace > 0 { + total = ds.TotalSpace + } + used := ds.Used + if used == 0 && ds.UsedSpace > 0 { + used = ds.UsedSpace + } + avail := ds.Avail + if avail == 0 && ds.AvailSpace > 0 { + avail = ds.AvailSpace + } + if total == 0 && used > 0 && avail > 0 { + total = used + avail + } - log.Debug(). - Str("store", ds.Store). - Int64("total", total). - Int64("used", used). - Int64("avail", avail). - Int64("orig_total", ds.Total). - Int64("orig_total_space", ds.TotalSpace). - Msg("PBS datastore details") + log.Debug(). + Str("store", ds.Store). + Int64("total", total). + Int64("used", used). + Int64("avail", avail). + Int64("orig_total", ds.Total). + Int64("orig_total_space", ds.TotalSpace). + Msg("PBS datastore details") - modelDS := models.PBSDatastore{ - Name: ds.Store, - Total: total, - Used: used, - Free: avail, - Usage: safePercentage(float64(used), float64(total)), - Status: "available", - DeduplicationFactor: ds.DeduplicationFactor, - } + modelDS := models.PBSDatastore{ + Name: ds.Store, + Total: total, + Used: used, + Free: avail, + Usage: safePercentage(float64(used), float64(total)), + Status: "available", + DeduplicationFactor: ds.DeduplicationFactor, + } - namespaces, err := client.ListNamespaces(ctx, ds.Store, "", 0) - if err != nil { - log.Warn().Err(err). - Str("instance", instanceName). - Str("datastore", ds.Store). - Msg("Failed to list namespaces") - } else { - for _, ns := range namespaces { - nsPath := ns.NS - if nsPath == "" { - nsPath = ns.Path - } - if nsPath == "" { - nsPath = ns.Name - } + namespaces, err := client.ListNamespaces(ctx, ds.Store, "", 0) + if err != nil { + log.Warn().Err(err). + Str("instance", instanceName). + Str("datastore", ds.Store). + Msg("Failed to list namespaces") + } else { + for _, ns := range namespaces { + nsPath := ns.NS + if nsPath == "" { + nsPath = ns.Path + } + if nsPath == "" { + nsPath = ns.Name + } - modelNS := models.PBSNamespace{ - Path: nsPath, - Parent: ns.Parent, - Depth: strings.Count(nsPath, "/"), - } - modelDS.Namespaces = append(modelDS.Namespaces, modelNS) - } + modelNS := models.PBSNamespace{ + Path: nsPath, + Parent: ns.Parent, + Depth: strings.Count(nsPath, "/"), + } + modelDS.Namespaces = append(modelDS.Namespaces, modelNS) + } - hasRoot := false - for _, ns := range modelDS.Namespaces { - if ns.Path == "" { - hasRoot = true - break - } - } - if !hasRoot { - modelDS.Namespaces = append([]models.PBSNamespace{{Path: "", Depth: 0}}, modelDS.Namespaces...) - } - } + hasRoot := false + for _, ns := range modelDS.Namespaces { + if ns.Path == "" { + hasRoot = true + break + } + } + if !hasRoot { + modelDS.Namespaces = append([]models.PBSNamespace{{Path: "", Depth: 0}}, modelDS.Namespaces...) + } + } - pbsInst.Datastores = append(pbsInst.Datastores, modelDS) - } - } - } + pbsInst.Datastores = append(pbsInst.Datastores, modelDS) + } + } + } - // Update state and run alerts - m.state.UpdatePBSInstance(pbsInst) - log.Info(). - Str("instance", instanceName). - Str("id", pbsInst.ID). - Int("datastores", len(pbsInst.Datastores)). - Msg("PBS instance updated in state") + // Update state and run alerts + m.state.UpdatePBSInstance(pbsInst) + log.Info(). + Str("instance", instanceName). + Str("id", pbsInst.ID). + Int("datastores", len(pbsInst.Datastores)). + Msg("PBS instance updated in state") - if m.alertManager != nil { - m.alertManager.CheckPBS(pbsInst) - } + if m.alertManager != nil { + m.alertManager.CheckPBS(pbsInst) + } - // Poll backups if enabled - if instanceCfg.MonitorBackups { - if len(pbsInst.Datastores) == 0 { - log.Debug(). - Str("instance", instanceName). - Msg("No PBS datastores available for backup polling") - } else if !m.config.EnableBackupPolling { - log.Debug(). - Str("instance", instanceName). - Msg("Skipping PBS backup polling - globally disabled") - } else { - now := time.Now() + // Poll backups if enabled + if instanceCfg.MonitorBackups { + if len(pbsInst.Datastores) == 0 { + log.Debug(). + Str("instance", instanceName). + Msg("No PBS datastores available for backup polling") + } else if !m.config.EnableBackupPolling { + log.Debug(). + Str("instance", instanceName). + Msg("Skipping PBS backup polling - globally disabled") + } else { + now := time.Now() - m.mu.RLock() - lastPoll := m.lastPBSBackupPoll[instanceName] - inProgress := m.pbsBackupPollers[instanceName] - m.mu.RUnlock() + m.mu.RLock() + lastPoll := m.lastPBSBackupPoll[instanceName] + inProgress := m.pbsBackupPollers[instanceName] + m.mu.RUnlock() - shouldPoll, reason, newLast := m.shouldRunBackupPoll(lastPoll, now) - if !shouldPoll { - if reason != "" { - log.Debug(). - Str("instance", instanceName). - Str("reason", reason). - Msg("Skipping PBS backup polling this cycle") - } - } else if inProgress { - log.Debug(). - Str("instance", instanceName). - Msg("PBS backup polling already in progress") - } else { - datastoreSnapshot := make([]models.PBSDatastore, len(pbsInst.Datastores)) - copy(datastoreSnapshot, pbsInst.Datastores) + shouldPoll, reason, newLast := m.shouldRunBackupPoll(lastPoll, now) + if !shouldPoll { + if reason != "" { + log.Debug(). + Str("instance", instanceName). + Str("reason", reason). + Msg("Skipping PBS backup polling this cycle") + } + } else if inProgress { + log.Debug(). + Str("instance", instanceName). + Msg("PBS backup polling already in progress") + } else { + datastoreSnapshot := make([]models.PBSDatastore, len(pbsInst.Datastores)) + copy(datastoreSnapshot, pbsInst.Datastores) - m.mu.Lock() - if m.pbsBackupPollers == nil { - m.pbsBackupPollers = make(map[string]bool) - } - if m.pbsBackupPollers[instanceName] { - m.mu.Unlock() - } else { - m.pbsBackupPollers[instanceName] = true - m.lastPBSBackupPoll[instanceName] = newLast - m.mu.Unlock() + m.mu.Lock() + if m.pbsBackupPollers == nil { + m.pbsBackupPollers = make(map[string]bool) + } + if m.pbsBackupPollers[instanceName] { + m.mu.Unlock() + } else { + m.pbsBackupPollers[instanceName] = true + m.lastPBSBackupPoll[instanceName] = newLast + m.mu.Unlock() - go func(ds []models.PBSDatastore, inst string, start time.Time, pbsClient *pbs.Client) { - defer func() { - m.mu.Lock() - delete(m.pbsBackupPollers, inst) - m.lastPBSBackupPoll[inst] = time.Now() - m.mu.Unlock() - }() + go func(ds []models.PBSDatastore, inst string, start time.Time, pbsClient *pbs.Client) { + defer func() { + m.mu.Lock() + delete(m.pbsBackupPollers, inst) + m.lastPBSBackupPoll[inst] = time.Now() + m.mu.Unlock() + }() - log.Info(). - Str("instance", inst). - Int("datastores", len(ds)). - Msg("Starting background PBS backup polling") + log.Info(). + Str("instance", inst). + Int("datastores", len(ds)). + Msg("Starting background PBS backup polling") - backupCtx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) - defer cancel() + backupCtx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) + defer cancel() - m.pollPBSBackups(backupCtx, inst, pbsClient, ds) + m.pollPBSBackups(backupCtx, inst, pbsClient, ds) - log.Info(). - Str("instance", inst). - Dur("duration", time.Since(start)). - Msg("Completed background PBS backup polling") - }(datastoreSnapshot, instanceName, now, client) - } - } - } - } else { - log.Debug(). - Str("instance", instanceName). - Msg("PBS backup monitoring disabled") - } + log.Info(). + Str("instance", inst). + Dur("duration", time.Since(start)). + Msg("Completed background PBS backup polling") + }(datastoreSnapshot, instanceName, now, client) + } + } + } + } else { + log.Debug(). + Str("instance", instanceName). + Msg("PBS backup monitoring disabled") + } } + // pollPMGInstance polls a single Proxmox Mail Gateway instance func (m *Monitor) pollPMGInstance(ctx context.Context, instanceName string, client *pmg.Client) { start := time.Now()