danvics/Pulse
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Fix initial setup caching and container discovery defaults

Browse source
This commit is contained in:
rcourtman 2025-10-22 07:34:32 +00:00
parent ff4dc49ae4
commit 4eb8bed9b5
4 changed files with 582 additions and 528 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
frontend-modern/src/components/FirstRunSetup.tsx
View file

@ -131,9 +131,23 @@ export const FirstRunSetup: Component = () => {
// Save credentials for display // Save credentials for display
setSavedUsername(username()); setSavedUsername(username());
setSavedPassword(useCustomPassword() ? password() : generatedPassword()); setSavedPassword(finalPassword);
setSavedToken(token); 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 = { const bootstrapRecord: APITokenRecord = {
id: 'bootstrap-token', id: 'bootstrap-token',
name: 'Bootstrap token', name: 'Bootstrap token',

13
frontend-modern/src/components/Settings/QuickSecuritySetup.tsx
View file

@ -117,6 +117,19 @@ export const QuickSecuritySetup: Component<QuickSecuritySetupProps> = (props) =>
setCredentials(newCredentials); setCredentials(newCredentials);
setShowCredentials(true); 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 // Show success message
showSuccess( showSuccess(
isRotation isRotation

278
internal/discovery/config_override.go
View file

@ -1,64 +1,64 @@
package discovery package discovery
import ( import (
"fmt" "fmt"
"net" "net"
"strings" "strings"
"time" "time"
"github.com/rcourtman/pulse-go-rewrite/internal/config" "github.com/rcourtman/pulse-go-rewrite/internal/config"
pkgdiscovery "github.com/rcourtman/pulse-go-rewrite/pkg/discovery" pkgdiscovery "github.com/rcourtman/pulse-go-rewrite/pkg/discovery"
"github.com/rcourtman/pulse-go-rewrite/pkg/discovery/envdetect" "github.com/rcourtman/pulse-go-rewrite/pkg/discovery/envdetect"
) )
// BuildScanner creates a discovery scanner configured using the supplied discovery config. // BuildScanner creates a discovery scanner configured using the supplied discovery config.
func BuildScanner(cfg config.DiscoveryConfig) (*pkgdiscovery.Scanner, error) { func BuildScanner(cfg config.DiscoveryConfig) (*pkgdiscovery.Scanner, error) {
cfg = config.NormalizeDiscoveryConfig(cfg) cfg = config.NormalizeDiscoveryConfig(cfg)
profile, err := envdetect.DetectEnvironment() profile, err := envdetect.DetectEnvironment()
if err != nil { if err != nil {
return nil, err return nil, err
} }
ApplyConfigToProfile(profile, cfg) ApplyConfigToProfile(profile, cfg)
return pkgdiscovery.NewScannerWithProfile(profile), nil return pkgdiscovery.NewScannerWithProfile(profile), nil
} }
// ApplyConfigToProfile mutates the supplied environment profile according to the discovery config. // ApplyConfigToProfile mutates the supplied environment profile according to the discovery config.
func ApplyConfigToProfile(profile *envdetect.EnvironmentProfile, cfg config.DiscoveryConfig) { func ApplyConfigToProfile(profile *envdetect.EnvironmentProfile, cfg config.DiscoveryConfig) {
if profile == nil { if profile == nil {
return return
} }
// Environment override // Environment override
if env, ok := environmentFromOverride(cfg.EnvironmentOverride); ok { if env, ok := environmentFromOverride(cfg.EnvironmentOverride); ok {
profile.Type = env profile.Type = env
filterPhasesForEnvironment(profile, env) filterPhasesForEnvironment(profile, env)
} else if cfg.EnvironmentOverride != "" && strings.ToLower(cfg.EnvironmentOverride) != "auto" { } else if cfg.EnvironmentOverride != "" && strings.ToLower(cfg.EnvironmentOverride) != "auto" {
profile.Warnings = append(profile.Warnings, fmt.Sprintf("Unknown environment override: %s", cfg.EnvironmentOverride)) profile.Warnings = append(profile.Warnings, fmt.Sprintf("Unknown environment override: %s", cfg.EnvironmentOverride))
} }
// Apply subnet blocklist // Apply subnet blocklist
blocked := parseCIDRMap(cfg.SubnetBlocklist, &profile.Warnings) blocked := parseCIDRMap(cfg.SubnetBlocklist, &profile.Warnings)
if len(blocked) > 0 { if len(blocked) > 0 {
var filtered []envdetect.SubnetPhase var filtered []envdetect.SubnetPhase
for _, phase := range profile.Phases { for _, phase := range profile.Phases {
var kept []net.IPNet var kept []net.IPNet
for _, subnet := range phase.Subnets { for _, subnet := range phase.Subnets {
if _, blocked := blocked[subnet.String()]; blocked { if _, blocked := blocked[subnet.String()]; blocked {
continue continue
} }
kept = append(kept, subnet) kept = append(kept, subnet)
} }
if len(kept) > 0 { if len(kept) > 0 {
phase.Subnets = kept phase.Subnets = kept
filtered = append(filtered, phase) filtered = append(filtered, phase)
} }
} }
profile.Phases = filtered profile.Phases = filtered
} }
// Apply subnet allowlist as highest priority phase // Apply subnet allowlist as highest priority phase
if len(cfg.SubnetAllowlist) > 0 { if len(cfg.SubnetAllowlist) > 0 {
allowlist := parseCIDRs(cfg.SubnetAllowlist, &profile.Warnings) allowlist := parseCIDRs(cfg.SubnetAllowlist, &profile.Warnings)
if len(allowlist) > 0 { if len(allowlist) > 0 {
@ -79,104 +79,126 @@ func ApplyConfigToProfile(profile *envdetect.EnvironmentProfile, cfg config.Disc
Subnets: allowlist, Subnets: allowlist,
Confidence: 1.0, Confidence: 1.0,
Priority: 0, Priority: 0,
} }
profile.Phases = append([]envdetect.SubnetPhase{allowPhase}, profile.Phases...) profile.Phases = append([]envdetect.SubnetPhase{allowPhase}, profile.Phases...)
} }
} }
// Override scan policy if len(cfg.SubnetAllowlist) == 0 && shouldPruneContainerNetworks(profile.Type) {
if cfg.MaxHostsPerScan > 0 { pruned := make([]envdetect.SubnetPhase, 0, len(profile.Phases))
profile.Policy.MaxHostsPerScan = cfg.MaxHostsPerScan for _, phase := range profile.Phases {
} if isLikelyContainerPhase(phase.Name) {
if cfg.MaxConcurrent > 0 { continue
profile.Policy.MaxConcurrent = cfg.MaxConcurrent }
} pruned = append(pruned, phase)
profile.Policy.EnableReverseDNS = cfg.EnableReverseDNS }
profile.Policy.ScanGateways = cfg.ScanGateways if len(pruned) > 0 {
profile.Phases = pruned
}
}
if cfg.DialTimeout > 0 { // Override scan policy
profile.Policy.DialTimeout = time.Duration(cfg.DialTimeout) * time.Millisecond if cfg.MaxHostsPerScan > 0 {
} profile.Policy.MaxHostsPerScan = cfg.MaxHostsPerScan
if cfg.HTTPTimeout > 0 { }
profile.Policy.HTTPTimeout = time.Duration(cfg.HTTPTimeout) * time.Millisecond 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 { func parseCIDRs(values []string, warnings *[]string) []net.IPNet {
var subnets []net.IPNet var subnets []net.IPNet
for _, value := range values { for _, value := range values {
value = strings.TrimSpace(value) value = strings.TrimSpace(value)
if value == "" { if value == "" {
continue continue
} }
_, ipNet, err := net.ParseCIDR(value) _, ipNet, err := net.ParseCIDR(value)
if err != nil { if err != nil {
if warnings != nil { if warnings != nil {
*warnings = append(*warnings, fmt.Sprintf("Invalid CIDR '%s' ignored", value)) *warnings = append(*warnings, fmt.Sprintf("Invalid CIDR '%s' ignored", value))
} }
continue continue
} }
subnets = append(subnets, *ipNet) subnets = append(subnets, *ipNet)
} }
return subnets return subnets
} }
func parseCIDRMap(values []string, warnings *[]string) map[string]struct{} { func parseCIDRMap(values []string, warnings *[]string) map[string]struct{} {
cidrs := parseCIDRs(values, warnings) cidrs := parseCIDRs(values, warnings)
result := make(map[string]struct{}, len(cidrs)) result := make(map[string]struct{}, len(cidrs))
for _, cidr := range cidrs { for _, cidr := range cidrs {
result[cidr.String()] = struct{}{} result[cidr.String()] = struct{}{}
} }
return result return result
} }
func environmentFromOverride(value string) (envdetect.Environment, bool) { func environmentFromOverride(value string) (envdetect.Environment, bool) {
normalized := strings.ToLower(strings.TrimSpace(value)) normalized := strings.ToLower(strings.TrimSpace(value))
switch normalized { switch normalized {
case "", "auto": case "", "auto":
return envdetect.Unknown, false return envdetect.Unknown, false
case "native": case "native":
return envdetect.Native, true return envdetect.Native, true
case "docker_host": case "docker_host":
return envdetect.DockerHost, true return envdetect.DockerHost, true
case "docker_bridge": case "docker_bridge":
return envdetect.DockerBridge, true return envdetect.DockerBridge, true
case "lxc_privileged": case "lxc_privileged":
return envdetect.LXCPrivileged, true return envdetect.LXCPrivileged, true
case "lxc_unprivileged": case "lxc_unprivileged":
return envdetect.LXCUnprivileged, true return envdetect.LXCUnprivileged, true
default: default:
return envdetect.Unknown, false return envdetect.Unknown, false
} }
} }
func filterPhasesForEnvironment(profile *envdetect.EnvironmentProfile, env envdetect.Environment) { func filterPhasesForEnvironment(profile *envdetect.EnvironmentProfile, env envdetect.Environment) {
if len(profile.Phases) == 0 { if len(profile.Phases) == 0 {
return return
} }
var keep []envdetect.SubnetPhase var keep []envdetect.SubnetPhase
for _, phase := range profile.Phases { for _, phase := range profile.Phases {
name := strings.ToLower(phase.Name) name := strings.ToLower(phase.Name)
switch env { switch env {
case envdetect.Native, envdetect.DockerHost, envdetect.LXCPrivileged: case envdetect.Native, envdetect.DockerHost, envdetect.LXCPrivileged:
if strings.Contains(name, "local") || strings.Contains(name, "host") { if strings.Contains(name, "local") || strings.Contains(name, "host") {
keep = append(keep, phase) keep = append(keep, phase)
} }
case envdetect.DockerBridge: case envdetect.DockerBridge:
if strings.Contains(name, "container") || strings.Contains(name, "inferred") || strings.Contains(name, "host") { if strings.Contains(name, "container") || strings.Contains(name, "inferred") || strings.Contains(name, "host") {
keep = append(keep, phase) keep = append(keep, phase)
} }
case envdetect.LXCUnprivileged: case envdetect.LXCUnprivileged:
if strings.Contains(name, "lxc") || strings.Contains(name, "container") || strings.Contains(name, "parent") { if strings.Contains(name, "lxc") || strings.Contains(name, "container") || strings.Contains(name, "parent") {
keep = append(keep, phase) keep = append(keep, phase)
} }
default: default:
keep = append(keep, phase) keep = append(keep, phase)
} }
} }
if len(keep) > 0 { if len(keep) > 0 {
profile.Phases = keep profile.Phases = keep
} }
} }

803
internal/monitoring/monitor.go
View file

@ -111,7 +111,7 @@ func mergeNVMeTempsIntoDisks(disks []models.PhysicalDisk, nodes []models.Node) [
} }
nvmeTempsByNode := make(map[string][]models.NVMeTemp) 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 { if node.Temperature == nil || !node.Temperature.Available || len(node.Temperature.NVMe) == 0 {
continue continue
} }
@ -371,7 +371,6 @@ type InstanceHealth struct {
DeadLetter InstanceDLQ `json:"deadLetter"` DeadLetter InstanceDLQ `json:"deadLetter"`
} }
func schedulerKey(instanceType InstanceType, name string) string { func schedulerKey(instanceType InstanceType, name string) string {
return string(instanceType) + "::" + name return string(instanceType) + "::" + name
} }
@ -386,60 +385,60 @@ func timePtr(t time.Time) *time.Time {
// Monitor handles all monitoring operations // Monitor handles all monitoring operations
type Monitor struct { type Monitor struct {
config *config.Config config *config.Config
state *models.State state *models.State
pveClients map[string]PVEClientInterface pveClients map[string]PVEClientInterface
pbsClients map[string]*pbs.Client pbsClients map[string]*pbs.Client
pmgClients map[string]*pmg.Client pmgClients map[string]*pmg.Client
pollMetrics *PollMetrics pollMetrics *PollMetrics
scheduler *AdaptiveScheduler scheduler *AdaptiveScheduler
stalenessTracker *StalenessTracker stalenessTracker *StalenessTracker
taskQueue *TaskQueue taskQueue *TaskQueue
circuitBreakers map[string]*circuitBreaker circuitBreakers map[string]*circuitBreaker
deadLetterQueue *TaskQueue deadLetterQueue *TaskQueue
failureCounts map[string]int failureCounts map[string]int
lastOutcome map[string]taskOutcome lastOutcome map[string]taskOutcome
backoffCfg backoffConfig backoffCfg backoffConfig
rng *rand.Rand rng *rand.Rand
maxRetryAttempts int maxRetryAttempts int
tempCollector *TemperatureCollector // SSH-based temperature collector tempCollector *TemperatureCollector // SSH-based temperature collector
mu sync.RWMutex mu sync.RWMutex
startTime time.Time startTime time.Time
rateTracker *RateTracker rateTracker *RateTracker
metricsHistory *MetricsHistory metricsHistory *MetricsHistory
alertManager *alerts.Manager alertManager *alerts.Manager
notificationMgr *notifications.NotificationManager notificationMgr *notifications.NotificationManager
configPersist *config.ConfigPersistence configPersist *config.ConfigPersistence
discoveryService *discovery.Service // Background discovery service discoveryService *discovery.Service // Background discovery service
activePollCount int32 // Number of active polling operations activePollCount int32 // Number of active polling operations
pollCounter int64 // Counter for polling cycles pollCounter int64 // Counter for polling cycles
authFailures map[string]int // Track consecutive auth failures per node authFailures map[string]int // Track consecutive auth failures per node
lastAuthAttempt map[string]time.Time // Track last auth attempt time lastAuthAttempt map[string]time.Time // Track last auth attempt time
lastClusterCheck map[string]time.Time // Track last cluster check for standalone nodes 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 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 lastPVEBackupPoll map[string]time.Time // Track last PVE backup poll per instance
lastPBSBackupPoll map[string]time.Time // Track last PBS 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 persistence *config.ConfigPersistence // Add persistence for saving updated configs
pbsBackupPollers map[string]bool // Track PBS backup polling goroutines per instance pbsBackupPollers map[string]bool // Track PBS backup polling goroutines per instance
runtimeCtx context.Context // Context used while monitor is running runtimeCtx context.Context // Context used while monitor is running
wsHub *websocket.Hub // Hub used for broadcasting state wsHub *websocket.Hub // Hub used for broadcasting state
diagMu sync.RWMutex // Protects diagnostic snapshot maps diagMu sync.RWMutex // Protects diagnostic snapshot maps
nodeSnapshots map[string]NodeMemorySnapshot nodeSnapshots map[string]NodeMemorySnapshot
guestSnapshots map[string]GuestMemorySnapshot guestSnapshots map[string]GuestMemorySnapshot
rrdCacheMu sync.RWMutex // Protects RRD memavailable cache rrdCacheMu sync.RWMutex // Protects RRD memavailable cache
nodeRRDMemCache map[string]rrdMemCacheEntry nodeRRDMemCache map[string]rrdMemCacheEntry
removedDockerHosts map[string]time.Time // Track deliberately removed Docker hosts (ID -> removal time) removedDockerHosts map[string]time.Time // Track deliberately removed Docker hosts (ID -> removal time)
dockerCommands map[string]*dockerHostCommand dockerCommands map[string]*dockerHostCommand
dockerCommandIndex map[string]string dockerCommandIndex map[string]string
guestMetadataMu sync.RWMutex guestMetadataMu sync.RWMutex
guestMetadataCache map[string]guestMetadataCacheEntry guestMetadataCache map[string]guestMetadataCacheEntry
executor PollExecutor executor PollExecutor
breakerBaseRetry time.Duration breakerBaseRetry time.Duration
breakerMaxDelay time.Duration breakerMaxDelay time.Duration
breakerHalfOpenWindow time.Duration breakerHalfOpenWindow time.Duration
instanceInfoCache map[string]*instanceInfo instanceInfoCache map[string]*instanceInfo
pollStatusMap map[string]*pollStatus pollStatusMap map[string]*pollStatus
dlqInsightMap map[string]*dlqInsight dlqInsightMap map[string]*dlqInsight
} }
type rrdMemCacheEntry struct { type rrdMemCacheEntry struct {
@ -530,10 +529,10 @@ type guestMetadataCacheEntry struct {
} }
type taskOutcome struct { type taskOutcome struct {
success bool success bool
transient bool transient bool
err error err error
recordedAt time.Time recordedAt time.Time
} }
func (m *Monitor) getNodeRRDMemAvailable(ctx context.Context, client PVEClientInterface, nodeName string) (uint64, error) { 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() taskQueue := NewTaskQueue()
deadLetterQueue := NewTaskQueue() deadLetterQueue := NewTaskQueue()
breakers := make(map[string]*circuitBreaker) breakers := make(map[string]*circuitBreaker)
failureCounts := make(map[string]int) failureCounts := make(map[string]int)
lastOutcome := make(map[string]taskOutcome) lastOutcome := make(map[string]taskOutcome)
backoff := backoffConfig{ backoff := backoffConfig{
Initial: 5 * time.Second, Initial: 5 * time.Second,
Multiplier: 2, Multiplier: 2,
@ -1893,7 +1892,6 @@ func (m *Monitor) buildInstanceInfoCache(cfg *config.Config) {
} }
} }
func (m *Monitor) getExecutor() PollExecutor { func (m *Monitor) getExecutor() PollExecutor {
m.mu.RLock() m.mu.RLock()
exec := m.executor exec := m.executor
@ -1999,12 +1997,12 @@ func (m *Monitor) Start(ctx context.Context, wsHub *websocket.Hub) {
// Create separate tickers for polling and broadcasting // Create separate tickers for polling and broadcasting
// Hardcoded to 10 seconds since Proxmox updates cluster/resources every 10 seconds // 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) workerCount := len(m.pveClients) + len(m.pbsClients) + len(m.pmgClients)
m.startTaskWorkers(ctx, workerCount) m.startTaskWorkers(ctx, workerCount)
pollTicker := time.NewTicker(pollingInterval) pollTicker := time.NewTicker(pollingInterval)
defer pollTicker.Stop() defer pollTicker.Stop()
broadcastTicker := time.NewTicker(pollingInterval) broadcastTicker := time.NewTicker(pollingInterval)
@ -2694,30 +2692,30 @@ func (m *Monitor) allowExecution(task ScheduledTask) bool {
} }
func (m *Monitor) ensureBreaker(key string) *circuitBreaker { func (m *Monitor) ensureBreaker(key string) *circuitBreaker {
m.mu.Lock() m.mu.Lock()
defer m.mu.Unlock() defer m.mu.Unlock()
if m.circuitBreakers == nil { if m.circuitBreakers == nil {
m.circuitBreakers = make(map[string]*circuitBreaker) m.circuitBreakers = make(map[string]*circuitBreaker)
} }
if breaker, ok := m.circuitBreakers[key]; ok { if breaker, ok := m.circuitBreakers[key]; ok {
return breaker return breaker
} }
baseRetry := m.breakerBaseRetry baseRetry := m.breakerBaseRetry
if baseRetry <= 0 { if baseRetry <= 0 {
baseRetry = 5 * time.Second baseRetry = 5 * time.Second
} }
maxDelay := m.breakerMaxDelay maxDelay := m.breakerMaxDelay
if maxDelay <= 0 { if maxDelay <= 0 {
maxDelay = 5 * time.Minute maxDelay = 5 * time.Minute
} }
halfOpen := m.breakerHalfOpenWindow halfOpen := m.breakerHalfOpenWindow
if halfOpen <= 0 { if halfOpen <= 0 {
halfOpen = 30 * time.Second halfOpen = 30 * time.Second
} }
breaker := newCircuitBreaker(3, baseRetry, maxDelay, halfOpen) breaker := newCircuitBreaker(3, baseRetry, maxDelay, halfOpen)
m.circuitBreakers[key] = breaker m.circuitBreakers[key] = breaker
return breaker return breaker
} }
func (m *Monitor) recordTaskResult(instanceType InstanceType, instance string, pollErr error) { 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. // SchedulerHealthResponse contains complete scheduler health data for API exposure.
type SchedulerHealthResponse struct { type SchedulerHealthResponse struct {
UpdatedAt time.Time `json:"updatedAt"` UpdatedAt time.Time `json:"updatedAt"`
Enabled bool `json:"enabled"` Enabled bool `json:"enabled"`
Queue QueueSnapshot `json:"queue"` Queue QueueSnapshot `json:"queue"`
DeadLetter DeadLetterSnapshot `json:"deadLetter"` DeadLetter DeadLetterSnapshot `json:"deadLetter"`
Breakers []BreakerSnapshot `json:"breakers,omitempty"` Breakers []BreakerSnapshot `json:"breakers,omitempty"`
Staleness []StalenessSnapshot `json:"staleness,omitempty"` Staleness []StalenessSnapshot `json:"staleness,omitempty"`
Instances []InstanceHealth `json:"instances"` Instances []InstanceHealth `json:"instances"`
} }
// DeadLetterSnapshot contains dead-letter queue data. // DeadLetterSnapshot contains dead-letter queue data.
@ -3667,52 +3665,58 @@ func (m *Monitor) pollPVEInstance(ctx context.Context, instanceName string, clie
Bool("isCluster", modelNode.IsClusterMember). Bool("isCluster", modelNode.IsClusterMember).
Int("endpointCount", len(instanceCfg.ClusterEndpoints)). Int("endpointCount", len(instanceCfg.ClusterEndpoints)).
Msg("Temperature collection failed - check SSH access") 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 { if m.pollMetrics != nil {
nodeNameLabel := strings.TrimSpace(node.Node) nodeNameLabel := strings.TrimSpace(node.Node)
if nodeNameLabel == "" { if nodeNameLabel == "" {
nodeNameLabel = strings.TrimSpace(modelNode.DisplayName) nodeNameLabel = strings.TrimSpace(modelNode.DisplayName)
} }
if nodeNameLabel == "" { if nodeNameLabel == "" {
nodeNameLabel = "unknown-node" nodeNameLabel = "unknown-node"
} }
success := true success := true
nodeErrReason := "" nodeErrReason := ""
health := strings.ToLower(strings.TrimSpace(modelNode.ConnectionHealth)) health := strings.ToLower(strings.TrimSpace(modelNode.ConnectionHealth))
if health != "" && health != "healthy" { if health != "" && health != "healthy" {
success = false success = false
nodeErrReason = fmt.Sprintf("connection health %s", health) nodeErrReason = fmt.Sprintf("connection health %s", health)
} }
status := strings.ToLower(strings.TrimSpace(modelNode.Status)) status := strings.ToLower(strings.TrimSpace(modelNode.Status))
if success && status != "" && status != "online" { if success && status != "" && status != "online" {
success = false success = false
nodeErrReason = fmt.Sprintf("status %s", status) nodeErrReason = fmt.Sprintf("status %s", status)
} }
var nodeErr error var nodeErr error
if !success { if !success {
if nodeErrReason == "" { if nodeErrReason == "" {
nodeErrReason = "unknown node error" nodeErrReason = "unknown node error"
} }
nodeErr = fmt.Errorf(nodeErrReason) nodeErr = stderrors.New(nodeErrReason)
} }
m.pollMetrics.RecordNodeResult(NodePollResult{ m.pollMetrics.RecordNodeResult(NodePollResult{
InstanceName: instanceName, InstanceName: instanceName,
InstanceType: "pve", InstanceType: "pve",
NodeName: nodeNameLabel, NodeName: nodeNameLabel,
Success: success, Success: success,
Error: nodeErr, Error: nodeErr,
StartTime: nodeStart, StartTime: nodeStart,
EndTime: time.Now(), EndTime: time.Now(),
}) })
} }
modelNodes = append(modelNodes, modelNode) modelNodes = append(modelNodes, modelNode)
} }
if len(modelNodes) == 0 && len(prevInstanceNodes) > 0 { if len(modelNodes) == 0 && len(prevInstanceNodes) > 0 {
@ -4138,19 +4142,19 @@ func (m *Monitor) pollPVEInstance(ctx context.Context, instanceName string, clie
m.mu.RUnlock() m.mu.RUnlock()
shouldPoll, reason, newLast := m.shouldRunBackupPoll(lastPoll, now) shouldPoll, reason, newLast := m.shouldRunBackupPoll(lastPoll, now)
if !shouldPoll { if !shouldPoll {
if reason != "" { if reason != "" {
log.Debug(). log.Debug().
Str("instance", instanceName). Str("instance", instanceName).
Str("reason", reason). Str("reason", reason).
Msg("Skipping PVE backup polling this cycle") Msg("Skipping PVE backup polling this cycle")
} }
} else { } else {
select { select {
case <-ctx.Done(): case <-ctx.Done():
pollErr = ctx.Err() pollErr = ctx.Err()
return return
default: default:
m.mu.Lock() m.mu.Lock()
m.lastPVEBackupPoll[instanceName] = newLast m.lastPVEBackupPoll[instanceName] = newLast
m.mu.Unlock() m.mu.Unlock()
@ -4939,8 +4943,8 @@ func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, clie
} }
defer m.recordTaskResult(InstanceTypePBS, instanceName, pollErr) defer m.recordTaskResult(InstanceTypePBS, instanceName, pollErr)
// Check if context is cancelled // Check if context is cancelled
select { select {
case <-ctx.Done(): case <-ctx.Done():
pollErr = ctx.Err() pollErr = ctx.Err()
if debugEnabled { 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") log.Debug().Str("instance", instanceName).Msg("Polling PBS instance")
} }
// Get instance config // Get instance config
var instanceCfg *config.PBSInstance var instanceCfg *config.PBSInstance
for _, cfg := range m.config.PBSInstances { for _, cfg := range m.config.PBSInstances {
if cfg.Name == instanceName { if cfg.Name == instanceName {
instanceCfg = &cfg instanceCfg = &cfg
if debugEnabled { if debugEnabled {
log.Debug(). log.Debug().
Str("instance", instanceName). Str("instance", instanceName).
Bool("monitorDatastores", cfg.MonitorDatastores). Bool("monitorDatastores", cfg.MonitorDatastores).
Msg("Found PBS instance config") Msg("Found PBS instance config")
} }
break break
} }
} }
if instanceCfg == nil { if instanceCfg == nil {
log.Error().Str("instance", instanceName).Msg("PBS instance config not found") log.Error().Str("instance", instanceName).Msg("PBS instance config not found")
return return
} }
// Initialize PBS instance with default values // Initialize PBS instance with default values
pbsInst := models.PBSInstance{ pbsInst := models.PBSInstance{
ID: "pbs-" + instanceName, ID: "pbs-" + instanceName,
Name: instanceName, Name: instanceName,
Host: instanceCfg.Host, Host: instanceCfg.Host,
Status: "offline", Status: "offline",
Version: "unknown", Version: "unknown",
ConnectionHealth: "unhealthy", ConnectionHealth: "unhealthy",
LastSeen: time.Now(), LastSeen: time.Now(),
} }
// Try to get version first // Try to get version first
version, versionErr := client.GetVersion(ctx) version, versionErr := client.GetVersion(ctx)
if versionErr == nil { if versionErr == nil {
pbsInst.Status = "online" pbsInst.Status = "online"
pbsInst.Version = version.Version pbsInst.Version = version.Version
pbsInst.ConnectionHealth = "healthy" pbsInst.ConnectionHealth = "healthy"
m.resetAuthFailures(instanceName, "pbs") m.resetAuthFailures(instanceName, "pbs")
m.state.SetConnectionHealth("pbs-"+instanceName, true) m.state.SetConnectionHealth("pbs-"+instanceName, true)
if debugEnabled { if debugEnabled {
log.Debug(). log.Debug().
Str("instance", instanceName). Str("instance", instanceName).
Str("version", version.Version). Str("version", version.Version).
Bool("monitorDatastores", instanceCfg.MonitorDatastores). Bool("monitorDatastores", instanceCfg.MonitorDatastores).
Msg("PBS version retrieved successfully") Msg("PBS version retrieved successfully")
} }
} else { } else {
if debugEnabled { if debugEnabled {
log.Debug().Err(versionErr).Str("instance", instanceName).Msg("Failed to get PBS version, trying fallback") log.Debug().Err(versionErr).Str("instance", instanceName).Msg("Failed to get PBS version, trying fallback")
} }
ctx2, cancel2 := context.WithTimeout(context.Background(), 10*time.Second) ctx2, cancel2 := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel2() defer cancel2()
_, datastoreErr := client.GetDatastores(ctx2) _, datastoreErr := client.GetDatastores(ctx2)
if datastoreErr == nil { if datastoreErr == nil {
pbsInst.Status = "online" pbsInst.Status = "online"
pbsInst.Version = "connected" pbsInst.Version = "connected"
pbsInst.ConnectionHealth = "healthy" pbsInst.ConnectionHealth = "healthy"
m.resetAuthFailures(instanceName, "pbs") m.resetAuthFailures(instanceName, "pbs")
m.state.SetConnectionHealth("pbs-"+instanceName, true) m.state.SetConnectionHealth("pbs-"+instanceName, true)
log.Info(). log.Info().
Str("instance", instanceName). Str("instance", instanceName).
Msg("PBS connected (version unavailable but datastores accessible)") Msg("PBS connected (version unavailable but datastores accessible)")
} else { } else {
pbsInst.Status = "offline" pbsInst.Status = "offline"
pbsInst.ConnectionHealth = "error" pbsInst.ConnectionHealth = "error"
monErr := errors.WrapConnectionError("get_pbs_version", instanceName, versionErr) monErr := errors.WrapConnectionError("get_pbs_version", instanceName, versionErr)
log.Error().Err(monErr).Str("instance", instanceName).Msg("Failed to connect to PBS") log.Error().Err(monErr).Str("instance", instanceName).Msg("Failed to connect to PBS")
m.state.SetConnectionHealth("pbs-"+instanceName, false) m.state.SetConnectionHealth("pbs-"+instanceName, false)
if errors.IsAuthError(versionErr) || errors.IsAuthError(datastoreErr) { if errors.IsAuthError(versionErr) || errors.IsAuthError(datastoreErr) {
m.recordAuthFailure(instanceName, "pbs") m.recordAuthFailure(instanceName, "pbs")
return return
} }
} }
} }
// Get node status (CPU, memory, etc.) // Get node status (CPU, memory, etc.)
nodeStatus, err := client.GetNodeStatus(ctx) nodeStatus, err := client.GetNodeStatus(ctx)
if err != nil { if err != nil {
if debugEnabled { if debugEnabled {
log.Debug().Err(err).Str("instance", instanceName).Msg("Could not get PBS node status (may need Sys.Audit permission)") log.Debug().Err(err).Str("instance", instanceName).Msg("Could not get PBS node status (may need Sys.Audit permission)")
} }
} else if nodeStatus != nil { } else if nodeStatus != nil {
pbsInst.CPU = nodeStatus.CPU pbsInst.CPU = nodeStatus.CPU
if nodeStatus.Memory.Total > 0 { if nodeStatus.Memory.Total > 0 {
pbsInst.Memory = float64(nodeStatus.Memory.Used) / float64(nodeStatus.Memory.Total) * 100 pbsInst.Memory = float64(nodeStatus.Memory.Used) / float64(nodeStatus.Memory.Total) * 100
pbsInst.MemoryUsed = nodeStatus.Memory.Used pbsInst.MemoryUsed = nodeStatus.Memory.Used
pbsInst.MemoryTotal = nodeStatus.Memory.Total pbsInst.MemoryTotal = nodeStatus.Memory.Total
} }
pbsInst.Uptime = nodeStatus.Uptime pbsInst.Uptime = nodeStatus.Uptime
log.Debug(). log.Debug().
Str("instance", instanceName). Str("instance", instanceName).
Float64("cpu", pbsInst.CPU). Float64("cpu", pbsInst.CPU).
Float64("memory", pbsInst.Memory). Float64("memory", pbsInst.Memory).
Int64("uptime", pbsInst.Uptime). Int64("uptime", pbsInst.Uptime).
Msg("PBS node status retrieved") Msg("PBS node status retrieved")
} }
// Poll datastores if enabled // Poll datastores if enabled
if instanceCfg.MonitorDatastores { if instanceCfg.MonitorDatastores {
datastores, err := client.GetDatastores(ctx) datastores, err := client.GetDatastores(ctx)
if err != nil { if err != nil {
monErr := errors.WrapAPIError("get_datastores", instanceName, err, 0) monErr := errors.WrapAPIError("get_datastores", instanceName, err, 0)
log.Error().Err(monErr).Str("instance", instanceName).Msg("Failed to get datastores") log.Error().Err(monErr).Str("instance", instanceName).Msg("Failed to get datastores")
} else { } else {
log.Info(). log.Info().
Str("instance", instanceName). Str("instance", instanceName).
Int("count", len(datastores)). Int("count", len(datastores)).
Msg("Got PBS datastores") Msg("Got PBS datastores")
for _, ds := range datastores { for _, ds := range datastores {
total := ds.Total total := ds.Total
if total == 0 && ds.TotalSpace > 0 { if total == 0 && ds.TotalSpace > 0 {
total = ds.TotalSpace total = ds.TotalSpace
} }
used := ds.Used used := ds.Used
if used == 0 && ds.UsedSpace > 0 { if used == 0 && ds.UsedSpace > 0 {
used = ds.UsedSpace used = ds.UsedSpace
} }
avail := ds.Avail avail := ds.Avail
if avail == 0 && ds.AvailSpace > 0 { if avail == 0 && ds.AvailSpace > 0 {
avail = ds.AvailSpace avail = ds.AvailSpace
} }
if total == 0 && used > 0 && avail > 0 { if total == 0 && used > 0 && avail > 0 {
total = used + avail total = used + avail
} }
log.Debug(). log.Debug().
Str("store", ds.Store). Str("store", ds.Store).
Int64("total", total). Int64("total", total).
Int64("used", used). Int64("used", used).
Int64("avail", avail). Int64("avail", avail).
Int64("orig_total", ds.Total). Int64("orig_total", ds.Total).
Int64("orig_total_space", ds.TotalSpace). Int64("orig_total_space", ds.TotalSpace).
Msg("PBS datastore details") Msg("PBS datastore details")
modelDS := models.PBSDatastore{ modelDS := models.PBSDatastore{
Name: ds.Store, Name: ds.Store,
Total: total, Total: total,
Used: used, Used: used,
Free: avail, Free: avail,
Usage: safePercentage(float64(used), float64(total)), Usage: safePercentage(float64(used), float64(total)),
Status: "available", Status: "available",
DeduplicationFactor: ds.DeduplicationFactor, DeduplicationFactor: ds.DeduplicationFactor,
} }
namespaces, err := client.ListNamespaces(ctx, ds.Store, "", 0) namespaces, err := client.ListNamespaces(ctx, ds.Store, "", 0)
if err != nil { if err != nil {
log.Warn().Err(err). log.Warn().Err(err).
Str("instance", instanceName). Str("instance", instanceName).
Str("datastore", ds.Store). Str("datastore", ds.Store).
Msg("Failed to list namespaces") Msg("Failed to list namespaces")
} else { } else {
for _, ns := range namespaces { for _, ns := range namespaces {
nsPath := ns.NS nsPath := ns.NS
if nsPath == "" { if nsPath == "" {
nsPath = ns.Path nsPath = ns.Path
} }
if nsPath == "" { if nsPath == "" {
nsPath = ns.Name nsPath = ns.Name
} }
modelNS := models.PBSNamespace{ modelNS := models.PBSNamespace{
Path: nsPath, Path: nsPath,
Parent: ns.Parent, Parent: ns.Parent,
Depth: strings.Count(nsPath, "/"), Depth: strings.Count(nsPath, "/"),
} }
modelDS.Namespaces = append(modelDS.Namespaces, modelNS) modelDS.Namespaces = append(modelDS.Namespaces, modelNS)
} }
hasRoot := false hasRoot := false
for _, ns := range modelDS.Namespaces { for _, ns := range modelDS.Namespaces {
if ns.Path == "" { if ns.Path == "" {
hasRoot = true hasRoot = true
break break
} }
} }
if !hasRoot { if !hasRoot {
modelDS.Namespaces = append([]models.PBSNamespace{{Path: "", Depth: 0}}, modelDS.Namespaces...) 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 // Update state and run alerts
m.state.UpdatePBSInstance(pbsInst) m.state.UpdatePBSInstance(pbsInst)
log.Info(). log.Info().
Str("instance", instanceName). Str("instance", instanceName).
Str("id", pbsInst.ID). Str("id", pbsInst.ID).
Int("datastores", len(pbsInst.Datastores)). Int("datastores", len(pbsInst.Datastores)).
Msg("PBS instance updated in state") Msg("PBS instance updated in state")
if m.alertManager != nil { if m.alertManager != nil {
m.alertManager.CheckPBS(pbsInst) m.alertManager.CheckPBS(pbsInst)
} }
// Poll backups if enabled // Poll backups if enabled
if instanceCfg.MonitorBackups { if instanceCfg.MonitorBackups {
if len(pbsInst.Datastores) == 0 { if len(pbsInst.Datastores) == 0 {
log.Debug(). log.Debug().
Str("instance", instanceName). Str("instance", instanceName).
Msg("No PBS datastores available for backup polling") Msg("No PBS datastores available for backup polling")
} else if !m.config.EnableBackupPolling { } else if !m.config.EnableBackupPolling {
log.Debug(). log.Debug().
Str("instance", instanceName). Str("instance", instanceName).
Msg("Skipping PBS backup polling - globally disabled") Msg("Skipping PBS backup polling - globally disabled")
} else { } else {
now := time.Now() now := time.Now()
m.mu.RLock() m.mu.RLock()
lastPoll := m.lastPBSBackupPoll[instanceName] lastPoll := m.lastPBSBackupPoll[instanceName]
inProgress := m.pbsBackupPollers[instanceName] inProgress := m.pbsBackupPollers[instanceName]
m.mu.RUnlock() m.mu.RUnlock()
shouldPoll, reason, newLast := m.shouldRunBackupPoll(lastPoll, now) shouldPoll, reason, newLast := m.shouldRunBackupPoll(lastPoll, now)
if !shouldPoll { if !shouldPoll {
if reason != "" { if reason != "" {
log.Debug(). log.Debug().
Str("instance", instanceName). Str("instance", instanceName).
Str("reason", reason). Str("reason", reason).
Msg("Skipping PBS backup polling this cycle") Msg("Skipping PBS backup polling this cycle")
} }
} else if inProgress { } else if inProgress {
log.Debug(). log.Debug().
Str("instance", instanceName). Str("instance", instanceName).
Msg("PBS backup polling already in progress") Msg("PBS backup polling already in progress")
} else { } else {
datastoreSnapshot := make([]models.PBSDatastore, len(pbsInst.Datastores)) datastoreSnapshot := make([]models.PBSDatastore, len(pbsInst.Datastores))
copy(datastoreSnapshot, pbsInst.Datastores) copy(datastoreSnapshot, pbsInst.Datastores)
m.mu.Lock() m.mu.Lock()
if m.pbsBackupPollers == nil { if m.pbsBackupPollers == nil {
m.pbsBackupPollers = make(map[string]bool) m.pbsBackupPollers = make(map[string]bool)
} }
if m.pbsBackupPollers[instanceName] { if m.pbsBackupPollers[instanceName] {
m.mu.Unlock() m.mu.Unlock()
} else { } else {
m.pbsBackupPollers[instanceName] = true m.pbsBackupPollers[instanceName] = true
m.lastPBSBackupPoll[instanceName] = newLast m.lastPBSBackupPoll[instanceName] = newLast
m.mu.Unlock() m.mu.Unlock()
go func(ds []models.PBSDatastore, inst string, start time.Time, pbsClient *pbs.Client) { go func(ds []models.PBSDatastore, inst string, start time.Time, pbsClient *pbs.Client) {
defer func() { defer func() {
m.mu.Lock() m.mu.Lock()
delete(m.pbsBackupPollers, inst) delete(m.pbsBackupPollers, inst)
m.lastPBSBackupPoll[inst] = time.Now() m.lastPBSBackupPoll[inst] = time.Now()
m.mu.Unlock() m.mu.Unlock()
}() }()
log.Info(). log.Info().
Str("instance", inst). Str("instance", inst).
Int("datastores", len(ds)). Int("datastores", len(ds)).
Msg("Starting background PBS backup polling") Msg("Starting background PBS backup polling")
backupCtx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) backupCtx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
defer cancel() defer cancel()
m.pollPBSBackups(backupCtx, inst, pbsClient, ds) m.pollPBSBackups(backupCtx, inst, pbsClient, ds)
log.Info(). log.Info().
Str("instance", inst). Str("instance", inst).
Dur("duration", time.Since(start)). Dur("duration", time.Since(start)).
Msg("Completed background PBS backup polling") Msg("Completed background PBS backup polling")
}(datastoreSnapshot, instanceName, now, client) }(datastoreSnapshot, instanceName, now, client)
} }
} }
} }
} else { } else {
log.Debug(). log.Debug().
Str("instance", instanceName). Str("instance", instanceName).
Msg("PBS backup monitoring disabled") Msg("PBS backup monitoring disabled")
} }
} }
// pollPMGInstance polls a single Proxmox Mail Gateway instance // pollPMGInstance polls a single Proxmox Mail Gateway instance
func (m *Monitor) pollPMGInstance(ctx context.Context, instanceName string, client *pmg.Client) { func (m *Monitor) pollPMGInstance(ctx context.Context, instanceName string, client *pmg.Client) {
start := time.Now() start := time.Now()