diff --git a/frontend-modern/src/utils/__tests__/resourceStateAdapters.test.ts b/frontend-modern/src/utils/__tests__/resourceStateAdapters.test.ts index 354e458d2..d1a0123d0 100644 --- a/frontend-modern/src/utils/__tests__/resourceStateAdapters.test.ts +++ b/frontend-modern/src/utils/__tests__/resourceStateAdapters.test.ts @@ -192,6 +192,22 @@ describe('resourceStateAdapters nodeFromResource', () => { expect(instance?.host).toBe('https://pmg-service.local:8006'); }); + it('uses PMG host and guest URLs from canonical platform data when available', () => { + const instance = pmgInstanceFromResource( + createServiceResource('pmg', { + pmg: { + hostname: 'pmg-service.local', + hostUrl: 'https://pmg-service.local:8443', + guestUrl: 'https://mail.example.com', + instanceId: 'pmg-instance-1', + }, + }), + ); + + expect(instance?.host).toBe('https://pmg-service.local:8443'); + expect(instance?.guestURL).toBe('https://mail.example.com'); + }); + it('keeps PMG operator labels on local identity when governed summaries exist', () => { const instance = pmgInstanceFromResource( createServiceResource( diff --git a/frontend-modern/src/utils/resourceStateAdapters.ts b/frontend-modern/src/utils/resourceStateAdapters.ts index cf2095ebc..507bb20c1 100644 --- a/frontend-modern/src/utils/resourceStateAdapters.ts +++ b/frontend-modern/src/utils/resourceStateAdapters.ts @@ -1224,7 +1224,7 @@ export const pmgInstanceFromResource = (resource: Resource): PMGInstance | null const platform = resourcePlatformData(resource); const pmg = asRecord(platform?.pmg); const hostName = getPreferredResourceHostname(resource) || resource.id; - const host = resource.platformId || `https://${hostName}:8006`; + const host = asString(pmg?.hostUrl) || resource.platformId || `https://${hostName}:8006`; const lastSeen = toISOTime(undefined, resource.lastSeen); const mailStats = mapPMGMailStats(pmg?.mailStats) || @@ -1240,6 +1240,7 @@ export const pmgInstanceFromResource = (resource: Resource): PMGInstance | null name: getPreferredInfrastructureDisplayName(resource), host, guestURL: + asString(pmg?.guestUrl) || asString((resource as unknown as Record).customURL) || asString((resource as unknown as Record).customUrl), status: resource.status || 'unknown', diff --git a/internal/ai/correlation/correlation_additional_test.go b/internal/ai/correlation/correlation_additional_test.go index 6f37d1a67..9c8c731d8 100644 --- a/internal/ai/correlation/correlation_additional_test.go +++ b/internal/ai/correlation/correlation_additional_test.go @@ -4,6 +4,7 @@ import ( "os" "path/filepath" "strings" + "sync" "testing" "time" ) @@ -155,6 +156,51 @@ func TestDetector_SaveLoadAndLargeFile(t *testing.T) { } } +func TestDetector_SaveToDiskSerializesConcurrentWrites(t *testing.T) { + dir := t.TempDir() + d := NewDetector(Config{ + DataDir: dir, + MinOccurrences: 1, + RetentionWindow: time.Hour, + }) + + for i := 0; i < 25; i++ { + d.events = append(d.events, Event{ + ID: intToStr(i), + ResourceID: "resource-" + intToStr(i), + EventType: EventAlert, + Timestamp: time.Now(), + }) + } + + start := make(chan struct{}) + errs := make(chan error, 25) + var wg sync.WaitGroup + for i := 0; i < 25; i++ { + wg.Add(1) + go func() { + defer wg.Done() + <-start + errs <- d.saveToDisk() + }() + } + + close(start) + wg.Wait() + close(errs) + + for err := range errs { + if err != nil { + t.Fatalf("concurrent save failed: %v", err) + } + } + + loaded := NewDetector(Config{DataDir: dir, MinOccurrences: 1}) + if len(loaded.events) != len(d.events) { + t.Fatalf("loaded events = %d, want %d", len(loaded.events), len(d.events)) + } +} + func TestDetector_LoadMissingAndInvalid(t *testing.T) { dir := t.TempDir() d := NewDetector(Config{DataDir: dir}) diff --git a/internal/ai/correlation/detector.go b/internal/ai/correlation/detector.go index 43e9f95c4..e8503a35a 100644 --- a/internal/ai/correlation/detector.go +++ b/internal/ai/correlation/detector.go @@ -59,6 +59,7 @@ type Correlation struct { // Detector tracks events and detects correlations between resources type Detector struct { mu sync.RWMutex + saveMu sync.Mutex events []Event correlations map[string]*Correlation // key: sourceID:targetID:pattern @@ -454,6 +455,9 @@ func (d *Detector) saveToDisk() error { return nil } + d.saveMu.Lock() + defer d.saveMu.Unlock() + d.mu.RLock() eventsSnapshot := make([]Event, len(d.events)) copy(eventsSnapshot, d.events) @@ -482,12 +486,37 @@ func (d *Detector) saveToDisk() error { } path := filepath.Join(d.dataDir, "ai_correlations.json") - tmpPath := path + ".tmp" - if err := os.WriteFile(tmpPath, jsonData, 0600); err != nil { + tmpFile, err := os.CreateTemp(d.dataDir, "ai_correlations-*.json.tmp") + if err != nil { return err } + tmpPath := tmpFile.Name() + cleanupTemp := true + defer func() { + if !cleanupTemp { + return + } + if err := os.Remove(tmpPath); err != nil && !os.IsNotExist(err) { + log.Warn().Err(err).Str("file", tmpPath).Msg("failed to remove temp correlation data file") + } + }() - return os.Rename(tmpPath, path) + if _, err := tmpFile.Write(jsonData); err != nil { + _ = tmpFile.Close() + return err + } + if err := tmpFile.Chmod(0600); err != nil { + _ = tmpFile.Close() + return err + } + if err := tmpFile.Close(); err != nil { + return err + } + if err := os.Rename(tmpPath, path); err != nil { + return err + } + cleanupTemp = false + return os.Chmod(path, 0600) } // loadFromDisk loads data diff --git a/internal/ai/parity_test.go b/internal/ai/parity_test.go index ec30be77a..7e949aa58 100644 --- a/internal/ai/parity_test.go +++ b/internal/ai/parity_test.go @@ -582,6 +582,7 @@ func TestParityPBSInstanceFields(t *testing.T) { func TestParityPMGInstanceFields(t *testing.T) { now := time.Now().UTC() queueUpdated := now.Add(-2 * time.Minute) + statsUpdated := now.Add(-5 * time.Minute) pmg := models.PMGInstance{ ID: "pmg-1", @@ -623,12 +624,24 @@ func TestParityPMGInstanceFields(t *testing.T) { }, }, MailStats: &models.PMGMailStats{ - Timeframe: "hour", - CountTotal: 123.0, - SpamIn: 7.0, - VirusIn: 2.0, - UpdatedAt: now.Add(-5 * time.Minute), + Timeframe: "hour", + CountTotal: 123.0, + CountIn: 100.0, + CountOut: 23.0, + SpamIn: 7.0, + VirusIn: 2.0, + JunkIn: 5.0, + PregreetRejects: 4.0, + UpdatedAt: statsUpdated, }, + MailCount: []models.PMGMailCountPoint{{ + Timestamp: statsUpdated, + Count: 123, + CountIn: 100, + CountOut: 23, + Timeframe: "hour", + Index: 1, + }}, ConnectionHealth: "connected", LastSeen: now, LastUpdated: now.Add(-1 * time.Minute), @@ -648,6 +661,7 @@ func TestParityPMGInstanceFields(t *testing.T) { require.Equal(t, pmg.Name, v.Name()) require.Equal(t, unifiedresources.StatusOnline, v.Status()) require.Equal(t, "pmg.example.test", v.Hostname()) + require.Equal(t, pmg.Host, v.HostURL()) require.Equal(t, pmg.Version, v.Version()) require.Equal(t, len(pmg.Nodes), v.NodeCount()) require.Equal(t, int64(2000), v.UptimeSeconds()) @@ -660,12 +674,24 @@ func TestParityPMGInstanceFields(t *testing.T) { require.Equal(t, pmg.MailStats.CountTotal, v.MailCountTotal()) require.Equal(t, pmg.MailStats.SpamIn, v.SpamIn()) require.Equal(t, pmg.MailStats.VirusIn, v.VirusIn()) + require.Len(t, v.Nodes(), 2) + require.NotNil(t, v.Nodes()[1].QueueStatus) + require.Equal(t, int64(300), v.Nodes()[1].QueueStatus.OldestAge) + require.True(t, v.Nodes()[1].QueueStatus.UpdatedAt.Equal(queueUpdated.Add(1*time.Minute))) + require.NotNil(t, v.MailStats()) + require.Equal(t, pmg.MailStats.CountIn, v.MailStats().CountIn) + require.Equal(t, pmg.MailStats.JunkIn, v.MailStats().JunkIn) + require.Equal(t, pmg.MailStats.PregreetRejects, v.MailStats().PregreetRejects) + require.True(t, v.MailStats().UpdatedAt.Equal(statsUpdated)) + require.Len(t, v.MailCount(), 1) + require.Equal(t, pmg.MailCount[0].Count, v.MailCount()[0].Count) require.Equal(t, pmg.ConnectionHealth, v.ConnectionHealth()) require.Equal(t, 0.0, v.CPUPercent()) require.Equal(t, 0.0, v.MemoryPercent()) require.Equal(t, 0.0, v.DiskPercent()) require.True(t, v.LastSeen().Equal(pmg.LastSeen)) + require.Equal(t, pmg.GuestURL, v.GuestURL()) require.Equal(t, pmg.GuestURL, v.CustomURL()) } diff --git a/internal/ai/service.go b/internal/ai/service.go index 7f3176482..5c9d9925c 100644 --- a/internal/ai/service.go +++ b/internal/ai/service.go @@ -562,9 +562,11 @@ func (s *Service) initDiscoveryServiceLocked() { // Create the discovery service with config-driven settings discoveryCfg := servicediscovery.DefaultConfig() commandScanningEnabled := false + backgroundScanningEnabled := false if s.cfg != nil { discoveryCfg.Interval = s.cfg.GetDiscoveryInterval() commandScanningEnabled = s.cfg.Enabled && s.cfg.IsDiscoveryEnabled() && s.provider != nil + backgroundScanningEnabled = commandScanningEnabled && !BackgroundAutomationDisabledForDev() discoveryCfg.CommandScanning = commandScanningEnabled } @@ -577,11 +579,15 @@ func (s *Service) initDiscoveryServiceLocked() { s.discoveryService.SetReadState(s.readState) // Start background discovery if enabled and interval is set - if commandScanningEnabled && s.cfg.GetDiscoveryInterval() > 0 { + if backgroundScanningEnabled && s.cfg.GetDiscoveryInterval() > 0 { s.discoveryService.Start(context.Background()) log.Info(). Dur("interval", s.cfg.GetDiscoveryInterval()). Msg("AI-powered deep discovery service started with automatic scanning") + } else if commandScanningEnabled && BackgroundAutomationDisabledForDev() { + log.Info(). + Str("env", DevDisableBackgroundAIEnv). + Msg("AI-powered deep discovery service initialized; automatic scanning disabled for dev") } else if commandScanningEnabled { log.Info().Msg("AI-powered deep discovery service initialized (manual command scanning mode)") } else { @@ -849,9 +855,10 @@ func (s *Service) updateInfraDiscoverySettings(cfg *config.AIConfig) { } enabled := s.IsEnabled() && cfg.IsDiscoveryEnabled() + backgroundEnabled := enabled && !BackgroundAutomationDisabledForDev() interval := cfg.GetDiscoveryInterval() - if enabled && interval > 0 { + if backgroundEnabled && interval > 0 { s.infraDiscoveryService.SetInterval(interval) s.infraDiscoveryService.Start(context.Background()) log.Info(). @@ -874,10 +881,11 @@ func (s *Service) updateDiscoverySettings(cfg *config.AIConfig) { } enabled := s.IsEnabled() && cfg.IsDiscoveryEnabled() + backgroundEnabled := enabled && !BackgroundAutomationDisabledForDev() interval := cfg.GetDiscoveryInterval() s.discoveryService.SetCommandScanningEnabled(enabled) - if enabled && interval > 0 { + if backgroundEnabled && interval > 0 { // Update interval and ensure service is running s.discoveryService.SetInterval(interval) s.discoveryService.Start(context.Background()) @@ -1021,6 +1029,28 @@ const ( FeatureAIAutoFix = pkglicensing.FeatureAIAutoFix ) +const DevDisableBackgroundAIEnv = "PULSE_DEV_DISABLE_BACKGROUND_AI" + +// BackgroundAutomationDisabledForDev returns true when the local development +// runtime should avoid automatic AI work that can burn provider quota while a +// developer is only trying to run the app. +func BackgroundAutomationDisabledForDev() bool { + if os.Getenv("PULSE_DEV") != "true" && !strings.EqualFold(os.Getenv("NODE_ENV"), "development") { + return false + } + + value := strings.TrimSpace(os.Getenv(DevDisableBackgroundAIEnv)) + if value == "" { + return false + } + + disabled, err := strconv.ParseBool(value) + if err != nil { + return false + } + return disabled +} + // StartPatrol starts the background patrol service func (s *Service) StartPatrol(ctx context.Context) { s.mu.RLock() @@ -1049,6 +1079,21 @@ func (s *Service) StartPatrol(ctx context.Context) { patrolCfg := s.patrolConfigFromAIConfig(cfg) patrol.SetConfig(patrolCfg) patrol.SetProactiveMode(cfg.UseProactiveThresholds) + + if BackgroundAutomationDisabledForDev() { + patrol.SetEventTriggerConfig(PatrolEventTriggerConfig{ + AlertTriggersEnabled: false, + AnomalyTriggersEnabled: false, + }) + if alertAnalyzer != nil { + alertAnalyzer.SetEnabled(false) + } + log.Info(). + Str("env", DevDisableBackgroundAIEnv). + Msg("Pulse dev background AI disabled; Patrol scheduler and alert-triggered AI are not started") + return + } + patrol.SetEventTriggerConfig(PatrolEventTriggerConfig{ AlertTriggersEnabled: cfg.IsPatrolAlertTriggersEnabled(), AnomalyTriggersEnabled: cfg.IsPatrolAnomalyTriggersEnabled(), @@ -1150,6 +1195,21 @@ func (s *Service) ReconfigurePatrol() { // Update proactive threshold mode patrol.SetProactiveMode(cfg.UseProactiveThresholds) + if BackgroundAutomationDisabledForDev() { + patrol.SetEventTriggerConfig(PatrolEventTriggerConfig{ + AlertTriggersEnabled: false, + AnomalyTriggersEnabled: false, + }) + patrol.Stop() + if alertAnalyzer != nil { + alertAnalyzer.SetEnabled(false) + } + log.Info(). + Str("env", DevDisableBackgroundAIEnv). + Msg("Pulse dev background AI disabled; Patrol background automation stopped") + return + } + // Update event-driven patrol trigger gate patrol.SetEventTriggerConfig(PatrolEventTriggerConfig{ AlertTriggersEnabled: cfg.IsPatrolAlertTriggersEnabled(), diff --git a/internal/ai/service_extended_test.go b/internal/ai/service_extended_test.go index 30f5e5294..3faedc278 100644 --- a/internal/ai/service_extended_test.go +++ b/internal/ai/service_extended_test.go @@ -854,6 +854,42 @@ func TestService_PatrolManagement(t *testing.T) { } } +func TestService_StartPatrolHonorsDevBackgroundAIGuard(t *testing.T) { + t.Setenv("PULSE_DEV", "true") + t.Setenv(DevDisableBackgroundAIEnv, "true") + + svc := NewService(nil, nil) + patrol := NewPatrolService(svc, nil) + svc.patrolService = patrol + + alertAnalyzer := &mockAlertAnalyzer{enabled: true} + svc.alertTriggeredAnalyzer = alertAnalyzer + svc.licenseChecker = &mockLicenseStore{ + features: map[string]bool{ + FeatureAIPatrol: true, + FeatureAIAlerts: true, + }, + } + svc.cfg = &config.AIConfig{ + Enabled: true, + PatrolEnabled: true, + PatrolIntervalMinutes: 30, + AlertTriggeredAnalysis: true, + PatrolEventTriggersEnabled: true, + PatrolAlertTriggersEnabled: true, + PatrolAnomalyTriggersEnabled: true, + } + + svc.StartPatrol(context.Background()) + + if alertAnalyzer.IsEnabled() { + t.Fatal("expected dev background AI guard to keep alert analyzer disabled") + } + if status := patrol.GetStatus(); status.NextPatrolAt != nil { + t.Fatalf("expected dev background AI guard to avoid scheduling patrol, got %v", status.NextPatrolAt) + } +} + func TestService_LookupNodeForVMID_Extended(t *testing.T) { svc := NewService(nil, nil) diff --git a/internal/ai/service_test.go b/internal/ai/service_test.go index 016c78dfa..7aaba3d9e 100644 --- a/internal/ai/service_test.go +++ b/internal/ai/service_test.go @@ -34,6 +34,36 @@ func (m *mockAlertAnalyzer) IsEnabled() bool { return m.en func (m *mockAlertAnalyzer) Start() {} func (m *mockAlertAnalyzer) Stop() {} +func TestBackgroundAutomationDisabledForDev(t *testing.T) { + t.Run("disabled in Pulse dev", func(t *testing.T) { + t.Setenv("PULSE_DEV", "true") + t.Setenv(DevDisableBackgroundAIEnv, "true") + + if !BackgroundAutomationDisabledForDev() { + t.Fatal("expected dev background AI automation to be disabled") + } + }) + + t.Run("explicit false allows dev automation", func(t *testing.T) { + t.Setenv("PULSE_DEV", "true") + t.Setenv(DevDisableBackgroundAIEnv, "false") + + if BackgroundAutomationDisabledForDev() { + t.Fatal("expected explicit false to allow dev background AI automation") + } + }) + + t.Run("ignored outside dev", func(t *testing.T) { + t.Setenv("PULSE_DEV", "") + t.Setenv("NODE_ENV", "production") + t.Setenv(DevDisableBackgroundAIEnv, "true") + + if BackgroundAutomationDisabledForDev() { + t.Fatal("expected background AI automation guard to be dev-only") + } + }) +} + func TestNewService(t *testing.T) { svc := NewService(nil, nil) diff --git a/internal/ai/tools/pmg_test.go b/internal/ai/tools/pmg_test.go index 4cc184797..2dccddb70 100644 --- a/internal/ai/tools/pmg_test.go +++ b/internal/ai/tools/pmg_test.go @@ -391,4 +391,5 @@ func TestExecuteGetMailQueuesUsesFirstInstance(t *testing.T) { require.NoError(t, json.Unmarshal([]byte(result.Content[0].Text), &resp)) assert.Equal(t, "gateway-1", resp.Instance) require.Len(t, resp.Queues, 1) + assert.Equal(t, int64(5), resp.Queues[0].OldestAgeSeconds) } diff --git a/internal/ai/tools/tools_pmg.go b/internal/ai/tools/tools_pmg.go index 6ce2914ef..ed7dc080b 100644 --- a/internal/ai/tools/tools_pmg.go +++ b/internal/ai/tools/tools_pmg.go @@ -251,7 +251,7 @@ func (e *PulseToolExecutor) executeGetMailQueues(_ context.Context, args map[str Hold: node.QueueStatus.Hold, Incoming: node.QueueStatus.Incoming, Total: node.QueueStatus.Total, - OldestAgeSeconds: 0, // Age not provided in unified resource + OldestAgeSeconds: node.QueueStatus.OldestAge, }) } } diff --git a/internal/alerts/active_cleanup.go b/internal/alerts/active_cleanup.go index a096ed513..16772a8ca 100644 --- a/internal/alerts/active_cleanup.go +++ b/internal/alerts/active_cleanup.go @@ -5,6 +5,7 @@ import ( "time" alertspecs "github.com/rcourtman/pulse-go-rewrite/internal/alerts/specs" + "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" "github.com/rs/zerolog/log" ) @@ -220,7 +221,61 @@ func (m *Manager) Cleanup(maxAge time.Duration) { } } -// CleanupAlertsForNodes removes alerts for nodes that no longer exist. +func alertMetadataResourceType(alert *Alert) string { + if alert == nil || alert.Metadata == nil { + return "" + } + if value, ok := alert.Metadata["resourceType"].(string); ok { + return strings.ToLower(strings.TrimSpace(value)) + } + return "" +} + +func shouldPreserveAlertOutsideNodeCleanup(alertID string, alert *Alert) bool { + if alert == nil { + return true + } + if strings.HasPrefix(alertID, "docker-") || strings.HasPrefix(alert.ResourceID, "docker:") { + return true + } + if strings.HasPrefix(alertID, "pbs-") || alert.Type == "pbs-offline" { + return true + } + if strings.HasPrefix(alert.ResourceID, "agent:") { + return true + } + if alert.CanonicalKind == string(alertspecs.AlertSpecKindProviderIncident) { + return true + } + + switch alertMetadataResourceType(alert) { + case string(unifiedresources.ResourceTypeAgent), + string(unifiedresources.ResourceTypeAppContainer), + string(unifiedresources.ResourceTypeDockerService), + string(unifiedresources.ResourceTypeK8sCluster), + string(unifiedresources.ResourceTypeK8sNode), + string(unifiedresources.ResourceTypePod), + string(unifiedresources.ResourceTypeK8sDeployment), + string(unifiedresources.ResourceTypeStorage), + string(unifiedresources.ResourceTypePBS), + string(unifiedresources.ResourceTypePMG), + string(unifiedresources.ResourceTypeCeph), + string(unifiedresources.ResourceTypePhysicalDisk), + string(unifiedresources.ResourceTypeNetworkShare), + string(unifiedresources.ResourceTypeNetworkEndpoint), + "docker", + "host", + "kubernetes", + "k8s", + "truenas", + "vmware": + return true + default: + return false + } +} + +// CleanupAlertsForNodes removes stale Proxmox node-scoped alerts. func (m *Manager) CleanupAlertsForNodes(existingNodes map[string]bool) { m.mu.Lock() defer m.mu.Unlock() @@ -238,33 +293,10 @@ func (m *Manager) CleanupAlertsForNodes(existingNodes map[string]bool) { continue } - if strings.HasPrefix(alertID, "docker-") || strings.HasPrefix(alert.ResourceID, "docker:") { - continue - } - if strings.HasPrefix(alertID, "pbs-") || alert.Type == "pbs-offline" { - continue - } - if alert.Metadata != nil { - if resourceType, _ := alert.Metadata["resourceType"].(string); resourceType == "pbs" { - continue - } - } if alert.CanonicalKind == string(alertspecs.AlertSpecKindConnectivity) && strings.HasPrefix(alert.ResourceID, "pbs") { continue } - // Agent-sourced resources (Unraid, standalone Linux hosts, TrueNAS, - // any platform reached via Pulse Agent) do not appear in the - // existingNodes map — that map is built only from Proxmox nodes - // and PBS instances. Without this carve-out, every cleanup cycle - // removes the agent alert because alert.Node is empty or unmapped, - // then the next poll re-creates the alert as "new", calls - // AddAlert, and appends a fresh history row. Result observed in - // the wild: 3,980 history entries in 7 days for what were really - // a small number of persistent issues (e.g. an Unraid array - // running without parity protection generating one entry per - // 30-second poll). Agent resource IDs are namespaced as - // "agent:...", so this carve-out matches that pattern. - if strings.HasPrefix(alert.ResourceID, "agent:") { + if shouldPreserveAlertOutsideNodeCleanup(alertID, alert) { continue } @@ -289,7 +321,7 @@ func (m *Manager) CleanupAlertsForNodes(existingNodes map[string]bool) { } }() } else { - log.Info().Msg("no alerts needed cleanup") + log.Debug().Msg("no alerts needed cleanup") } } diff --git a/internal/alerts/alerts_test.go b/internal/alerts/alerts_test.go index 9559acf6e..8e098e4bb 100644 --- a/internal/alerts/alerts_test.go +++ b/internal/alerts/alerts_test.go @@ -11372,6 +11372,84 @@ func TestCleanupAlertsForNodes(t *testing.T) { t.Error("non-agent alert with stale node must still be removed (control case)") } }) + + t.Run("preserves non-node canonical infrastructure alerts", func(t *testing.T) { + m := newTestManager(t) + + storageProviderID := buildCanonicalStateID("storage-1", "alertspec:provider-incident:test") + pmgQueueID := buildCanonicalStateID("pmg-main", "pmg-main-oldest-message") + + m.mu.Lock() + m.activeAlerts[storageProviderID] = &Alert{ + ID: storageProviderID, + ResourceID: "storage-1", + Type: "storage-incident", + CanonicalKind: string(alertspecs.AlertSpecKindProviderIncident), + Node: "", + } + m.activeAlerts[pmgQueueID] = &Alert{ + ID: pmgQueueID, + ResourceID: "pmg-main", + Type: "message-age", + Node: "", + Metadata: map[string]interface{}{"resourceType": "pmg"}, + } + m.activeAlerts["proxmox-empty-node"] = &Alert{ + ID: "proxmox-empty-node", + Node: "", + } + m.mu.Unlock() + + m.CleanupAlertsForNodes(map[string]bool{"pve1": true}) + time.Sleep(50 * time.Millisecond) + + m.mu.RLock() + _, storageStays := testLookupActiveAlert(t, m, storageProviderID) + _, pmgStays := testLookupActiveAlert(t, m, pmgQueueID) + _, proxmoxStays := testLookupActiveAlert(t, m, "proxmox-empty-node") + m.mu.RUnlock() + + if !storageStays { + t.Error("canonical provider incident must survive node cleanup") + } + if !pmgStays { + t.Error("PMG alert must survive node cleanup") + } + if proxmoxStays { + t.Error("plain node-scoped alert with empty node should still be removed") + } + }) + + t.Run("preserves PMG alerts created by canonical stateful evaluator", func(t *testing.T) { + m := newTestManager(t) + pmgState := buildCanonicalStateID("pmg1", "pmg1-oldest-message") + + m.checkPMGOldestMessage(models.PMGInstance{ + ID: "pmg1", + Name: "PMG 1", + Host: "https://pmg.mock.lan:8006", + Nodes: []models.PMGNodeStatus{ + {Name: "node1", QueueStatus: &models.PMGQueueStatus{OldestAge: 2400}}, + }, + }, PMGThresholdConfig{ + OldestMessageWarnMins: 30, + OldestMessageCritMins: 60, + }) + + m.CleanupAlertsForNodes(map[string]bool{"pve1": true}) + time.Sleep(50 * time.Millisecond) + + m.mu.RLock() + alert, pmgStays := testLookupActiveAlert(t, m, pmgState) + m.mu.RUnlock() + + if !pmgStays { + t.Fatal("PMG alert created by canonical stateful evaluator must survive node cleanup") + } + if got := alert.Metadata["resourceType"]; got != string(unifiedresources.ResourceTypePMG) { + t.Fatalf("resourceType = %v, want pmg", got) + } + }) } func TestCheckZFSPoolHealth(t *testing.T) { diff --git a/internal/alerts/canonical_lifecycle.go b/internal/alerts/canonical_lifecycle.go index 16722115d..3afc52431 100644 --- a/internal/alerts/canonical_lifecycle.go +++ b/internal/alerts/canonical_lifecycle.go @@ -386,6 +386,9 @@ func (m *Manager) evaluateCanonicalLifecycleAlert(params canonicalLifecycleAlert if alert.Metadata == nil { alert.Metadata = make(map[string]interface{}, 2) } + if _, ok := alert.Metadata["resourceType"]; !ok && params.Spec.ResourceType != "" { + alert.Metadata["resourceType"] = string(params.Spec.ResourceType) + } applyCanonicalIdentity(alert, params.Spec.ID, string(params.Spec.Kind)) m.preserveAlertState(storageKey, alert) @@ -508,6 +511,9 @@ func (m *Manager) evaluateCanonicalStatefulAlert(params canonicalStatefulAlertPa if alert.Metadata == nil { alert.Metadata = make(map[string]interface{}, 2) } + if _, ok := alert.Metadata["resourceType"]; !ok && params.Spec.ResourceType != "" { + alert.Metadata["resourceType"] = string(params.Spec.ResourceType) + } applyCanonicalIdentity(alert, params.Spec.ID, string(params.Spec.Kind)) m.preserveAlertState(storageKey, alert) diff --git a/internal/alerts/unified_eval_test.go b/internal/alerts/unified_eval_test.go index ca0bc6260..ac2c3d271 100644 --- a/internal/alerts/unified_eval_test.go +++ b/internal/alerts/unified_eval_test.go @@ -7,6 +7,7 @@ import ( alertspecs "github.com/rcourtman/pulse-go-rewrite/internal/alerts/specs" "github.com/rcourtman/pulse-go-rewrite/internal/models" + "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" ) func unifiedEvalBaseConfig() AlertConfig { @@ -807,6 +808,9 @@ func TestCheckPMGOldestMessageAnnotatesCanonicalSpecMetadata(t *testing.T) { if got := alert.Metadata["canonicalSpecID"]; got != "pmg-1-oldest-message" { t.Fatalf("canonicalSpecID = %v, want pmg-1-oldest-message", got) } + if got := alert.Metadata["resourceType"]; got != string(unifiedresources.ResourceTypePMG) { + t.Fatalf("resourceType = %v, want pmg", got) + } } func TestCheckPMGNodeQueueAnnotatesCanonicalSpecMetadata(t *testing.T) { diff --git a/internal/api/contract_test.go b/internal/api/contract_test.go index 97e7c6178..1908ebf27 100644 --- a/internal/api/contract_test.go +++ b/internal/api/contract_test.go @@ -60,6 +60,74 @@ type resourceContractSnapshot struct { Type string } +func TestContract_PMGInstancesEndpointUsesUnifiedReadStatePayload(t *testing.T) { + now := time.Now().UTC() + source := models.PMGInstance{ + ID: "pmg-contract-1", + Name: "gateway-contract", + Host: "https://pmg-contract.example.com:8006", + GuestURL: "https://pmg-contract.example.com/quarantine", + Status: "online", + Version: "8.2", + Nodes: []models.PMGNodeStatus{{ + Name: "pmg-node-contract", + Status: "online", + QueueStatus: &models.PMGQueueStatus{ + Incoming: 4, + Total: 10, + OldestAge: 600, + UpdatedAt: now, + }, + }}, + MailStats: &models.PMGMailStats{ + CountTotal: 1000, + JunkIn: 10, + PregreetRejects: 13, + UpdatedAt: now, + }, + MailCount: []models.PMGMailCountPoint{{Timestamp: now, Count: 1000, Timeframe: "hour", Index: 1}}, + DomainStats: []models.PMGDomainStat{{Domain: "example.com", MailCount: 100, Bytes: 2048}}, + DomainStatsAsOf: now, + ConnectionHealth: "connected", + LastSeen: now, + LastUpdated: now, + } + + registry := unifiedresources.NewRegistry(nil) + registry.IngestSnapshot(models.StateSnapshot{PMGInstances: []models.PMGInstance{source}}) + monitor := &monitoring.Monitor{} + setUnexportedField(t, monitor, "resourceStore", unifiedresources.NewMonitorAdapter(registry)) + + router := &Router{monitor: monitor} + req := httptest.NewRequest(http.MethodGet, "/api/pmg/instances?id=pmg-contract-1", nil) + rec := httptest.NewRecorder() + router.handleListPMGInstances(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, body=%s", rec.Code, rec.Body.String()) + } + var resp PMGInstancesResponse + if err := json.NewDecoder(rec.Body).Decode(&resp); err != nil { + t.Fatalf("decode PMG contract response: %v", err) + } + if resp.Meta.Total != 1 || len(resp.Data) != 1 { + t.Fatalf("expected one PMG instance, got total=%d len=%d", resp.Meta.Total, len(resp.Data)) + } + got := resp.Data[0] + if got.ID != source.ID || got.Host != source.Host || got.GuestURL != source.GuestURL { + t.Fatalf("PMG identity and URL contract not preserved: %+v", got) + } + if len(got.Nodes) != 1 || got.Nodes[0].QueueStatus == nil || got.Nodes[0].QueueStatus.OldestAge != 600 { + t.Fatalf("PMG queue contract not preserved: %+v", got.Nodes) + } + if got.MailStats == nil || got.MailStats.CountTotal != 1000 || got.MailStats.JunkIn != 10 || got.MailStats.PregreetRejects != 13 { + t.Fatalf("PMG mail stats contract not preserved: %+v", got.MailStats) + } + if len(got.MailCount) != 1 || got.MailCount[0].Index != 1 || len(got.DomainStats) != 1 || got.DomainStats[0].Bytes != 2048 { + t.Fatalf("PMG historical/domain contract not preserved: mail=%+v domains=%+v", got.MailCount, got.DomainStats) + } +} + func TestContract_MockAvailabilityTargetsUseSavedTargetPayloads(t *testing.T) { previousMock := mock.IsMockEnabled() if err := mock.SetEnabled(true); err != nil { diff --git a/internal/api/pmg.go b/internal/api/pmg.go index b79a9df4d..611650393 100644 --- a/internal/api/pmg.go +++ b/internal/api/pmg.go @@ -8,6 +8,7 @@ import ( "github.com/rcourtman/pulse-go-rewrite/internal/config" "github.com/rcourtman/pulse-go-rewrite/internal/models" + "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" "github.com/rcourtman/pulse-go-rewrite/internal/utils" ) @@ -42,8 +43,15 @@ func (r *Router) handleListPMGInstances(w http.ResponseWriter, req *http.Request return } - state := monitor.GetState() - instances := filterPMGInstances(state.PMGInstances, req) + readState := monitor.GetUnifiedReadStateOrSnapshot() + views := readState.PMGInstances() + + mappedInstances := make([]models.PMGInstance, len(views)) + for i, v := range views { + mappedInstances[i] = mapPMGInstanceViewToModel(v) + } + + instances := filterPMGInstances(mappedInstances, req) response := PMGInstancesResponse{ Data: instances, @@ -57,6 +65,143 @@ func (r *Router) handleListPMGInstances(w http.ResponseWriter, req *http.Request } } +// mapPMGInstanceViewToModel projects the canonical PMG read-state view onto +// the legacy API payload consumed by existing PMG surfaces. +func mapPMGInstanceViewToModel(v *unifiedresources.PMGInstanceView) models.PMGInstance { + if v == nil { + return models.PMGInstance{} + } + + nodes := v.Nodes() + modelNodes := make([]models.PMGNodeStatus, len(nodes)) + for i, n := range nodes { + modelNodes[i] = models.PMGNodeStatus{ + Name: n.Name, + Status: n.Status, + Role: n.Role, + Uptime: n.Uptime, + LoadAvg: n.LoadAvg, + } + if n.QueueStatus != nil { + modelNodes[i].QueueStatus = &models.PMGQueueStatus{ + Active: n.QueueStatus.Active, + Deferred: n.QueueStatus.Deferred, + Hold: n.QueueStatus.Hold, + Incoming: n.QueueStatus.Incoming, + Total: n.QueueStatus.Total, + OldestAge: n.QueueStatus.OldestAge, + UpdatedAt: n.QueueStatus.UpdatedAt, + } + } + } + + var modelMailStats *models.PMGMailStats + if stats := v.MailStats(); stats != nil { + countTotal := stats.CountTotal + if countTotal == 0 { + countTotal = v.MailCountTotal() + } + updatedAt := stats.UpdatedAt + if updatedAt.IsZero() { + updatedAt = v.LastUpdated() + } + modelMailStats = &models.PMGMailStats{ + Timeframe: stats.Timeframe, + CountTotal: countTotal, + CountIn: stats.CountIn, + CountOut: stats.CountOut, + SpamIn: stats.SpamIn, + SpamOut: stats.SpamOut, + VirusIn: stats.VirusIn, + VirusOut: stats.VirusOut, + BouncesIn: stats.BouncesIn, + BouncesOut: stats.BouncesOut, + BytesIn: stats.BytesIn, + BytesOut: stats.BytesOut, + GreylistCount: stats.GreylistCount, + JunkIn: stats.JunkIn, + AverageProcessTimeMs: stats.AverageProcessTimeMs, + RBLRejects: stats.RBLRejects, + PregreetRejects: stats.PregreetRejects, + UpdatedAt: updatedAt, + } + } + + var modelQuarantine *models.PMGQuarantineTotals + if q := v.Quarantine(); q != nil { + modelQuarantine = &models.PMGQuarantineTotals{ + Spam: q.Spam, + Virus: q.Virus, + Attachment: q.Attachment, + Blacklisted: q.Blacklisted, + } + } + + spamDist := v.SpamDistribution() + modelSpamDist := make([]models.PMGSpamBucket, len(spamDist)) + for i, b := range spamDist { + modelSpamDist[i] = models.PMGSpamBucket{ + Score: b.Bucket, + Count: b.Count, + } + } + + relayDomains := v.RelayDomains() + modelRelayDomains := make([]models.PMGRelayDomain, len(relayDomains)) + for i, d := range relayDomains { + modelRelayDomains[i] = models.PMGRelayDomain{ + Domain: d.Domain, + Comment: d.Comment, + } + } + + domainStats := v.DomainStats() + modelDomainStats := make([]models.PMGDomainStat, len(domainStats)) + for i, s := range domainStats { + modelDomainStats[i] = models.PMGDomainStat{ + Domain: s.Domain, + MailCount: s.MailCount, + SpamCount: s.SpamCount, + VirusCount: s.VirusCount, + Bytes: s.Bytes, + } + } + + host := v.HostURL() + if strings.TrimSpace(host) == "" { + host = v.Hostname() + } + guestURL := v.GuestURL() + if strings.TrimSpace(guestURL) == "" { + guestURL = v.CustomURL() + } + instanceID := v.InstanceID() + if strings.TrimSpace(instanceID) == "" { + instanceID = v.ID() + } + + inst := models.PMGInstance{ + ID: instanceID, + Name: v.Name(), + Host: host, + GuestURL: guestURL, + Status: string(v.Status()), + Version: v.Version(), + Nodes: modelNodes, + MailStats: modelMailStats, + MailCount: v.MailCount(), + SpamDistribution: modelSpamDist, + Quarantine: modelQuarantine, + RelayDomains: modelRelayDomains, + DomainStats: modelDomainStats, + DomainStatsAsOf: v.DomainStatsAsOf(), + ConnectionHealth: v.ConnectionHealth(), + LastSeen: v.LastSeen(), + LastUpdated: v.LastUpdated(), + } + return inst.NormalizeCollections() +} + // filterPMGInstances narrows the snapshot by optional `id=` or // `name=` query parameters. With neither, returns the full list. func filterPMGInstances(instances []models.PMGInstance, req *http.Request) []models.PMGInstance { diff --git a/internal/api/pmg_test.go b/internal/api/pmg_test.go new file mode 100644 index 000000000..320e43fd5 --- /dev/null +++ b/internal/api/pmg_test.go @@ -0,0 +1,143 @@ +package api + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/models" + "github.com/rcourtman/pulse-go-rewrite/internal/monitoring" + "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" +) + +func TestHandleListPMGInstancesUsesUnifiedReadStateWithFullPayload(t *testing.T) { + now := time.Now().UTC() + queueUpdated := now.Add(-2 * time.Minute) + statsUpdated := now.Add(-1 * time.Minute) + domainStatsAsOf := now.Add(-30 * time.Second) + + source := models.PMGInstance{ + ID: "pmg-1", + Name: "gateway-main", + Host: "https://pmg.example.com:8006", + GuestURL: "https://guest.example.com/pmg", + Status: "online", + Version: "8.2", + Nodes: []models.PMGNodeStatus{{ + Name: "pmg-node-1", + Status: "online", + Role: "master", + Uptime: 3600, + LoadAvg: "0.10", + QueueStatus: &models.PMGQueueStatus{ + Active: 1, + Deferred: 2, + Hold: 3, + Incoming: 4, + Total: 10, + OldestAge: 600, + UpdatedAt: queueUpdated, + }, + }}, + MailStats: &models.PMGMailStats{ + Timeframe: "day", + CountTotal: 1000, + CountIn: 700, + CountOut: 300, + SpamIn: 11, + SpamOut: 2, + VirusIn: 3, + VirusOut: 4, + BouncesIn: 5, + BouncesOut: 6, + BytesIn: 7000, + BytesOut: 8000, + GreylistCount: 9, + JunkIn: 10, + AverageProcessTimeMs: 42, + RBLRejects: 12, + PregreetRejects: 13, + UpdatedAt: statsUpdated, + }, + MailCount: []models.PMGMailCountPoint{{ + Timestamp: statsUpdated, + Count: 1000, + CountIn: 700, + CountOut: 300, + Timeframe: "hour", + Index: 1, + }}, + SpamDistribution: []models.PMGSpamBucket{{Score: "5", Count: 6}}, + Quarantine: &models.PMGQuarantineTotals{Spam: 1, Virus: 2, Attachment: 3, Blacklisted: 4}, + RelayDomains: []models.PMGRelayDomain{{Domain: "example.com", Comment: "primary relay"}}, + DomainStats: []models.PMGDomainStat{{Domain: "example.com", MailCount: 100, SpamCount: 5, VirusCount: 1, Bytes: 2048}}, + DomainStatsAsOf: domainStatsAsOf, + ConnectionHealth: "connected", + LastSeen: now, + LastUpdated: now, + } + + registry := unifiedresources.NewRegistry(nil) + registry.IngestSnapshot(models.StateSnapshot{PMGInstances: []models.PMGInstance{source}}) + monitor := &monitoring.Monitor{} + setUnexportedField(t, monitor, "resourceStore", unifiedresources.NewMonitorAdapter(registry)) + + router := &Router{monitor: monitor} + req := httptest.NewRequest(http.MethodGet, "/api/pmg/instances?id=pmg-1", nil) + rec := httptest.NewRecorder() + + router.handleListPMGInstances(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, body=%s", rec.Code, rec.Body.String()) + } + + var resp PMGInstancesResponse + if err := json.NewDecoder(rec.Body).Decode(&resp); err != nil { + t.Fatalf("decode response: %v", err) + } + if resp.Meta.Total != 1 || len(resp.Data) != 1 { + t.Fatalf("expected one PMG instance, got total=%d len=%d", resp.Meta.Total, len(resp.Data)) + } + + got := resp.Data[0] + if got.ID != source.ID || got.Name != source.Name || got.GuestURL != source.GuestURL || got.Status != source.Status || got.Version != source.Version { + t.Fatalf("basic PMG identity/status fields not preserved: %+v", got) + } + if got.Host != source.Host { + t.Fatalf("expected canonical host %q, got %q", source.Host, got.Host) + } + if len(got.Nodes) != 1 || got.Nodes[0].QueueStatus == nil { + t.Fatalf("expected node queue payload, got %+v", got.Nodes) + } + queue := got.Nodes[0].QueueStatus + if queue.OldestAge != 600 || !queue.UpdatedAt.Equal(queueUpdated) || queue.Incoming != 4 || queue.Total != 10 { + t.Fatalf("queue payload not preserved: %+v", queue) + } + if got.MailStats == nil { + t.Fatal("expected mail stats payload") + } + if got.MailStats.CountTotal != 1000 || got.MailStats.JunkIn != 10 || got.MailStats.PregreetRejects != 13 || !got.MailStats.UpdatedAt.Equal(statsUpdated) { + t.Fatalf("mail stats payload not preserved: %+v", got.MailStats) + } + if len(got.MailCount) != 1 || got.MailCount[0].Count != 1000 || got.MailCount[0].Index != 1 { + t.Fatalf("mail count payload not preserved: %+v", got.MailCount) + } + if len(got.SpamDistribution) != 1 || got.SpamDistribution[0].Score != "5" { + t.Fatalf("spam distribution not preserved: %+v", got.SpamDistribution) + } + if got.Quarantine == nil || got.Quarantine.Blacklisted != 4 { + t.Fatalf("quarantine payload not preserved: %+v", got.Quarantine) + } + if len(got.RelayDomains) != 1 || got.RelayDomains[0].Domain != "example.com" { + t.Fatalf("relay domains not preserved: %+v", got.RelayDomains) + } + if len(got.DomainStats) != 1 || got.DomainStats[0].Bytes != 2048 || !got.DomainStatsAsOf.Equal(domainStatsAsOf) { + t.Fatalf("domain stats not preserved: stats=%+v asOf=%v", got.DomainStats, got.DomainStatsAsOf) + } + if got.ConnectionHealth != "connected" || !got.LastSeen.Equal(now) || !got.LastUpdated.Equal(now) { + t.Fatalf("health/timestamps not preserved: %+v", got) + } +} diff --git a/internal/api/router.go b/internal/api/router.go index c1af9bf17..d9fb18386 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -2160,7 +2160,12 @@ func (r *Router) initializeAIIntelligenceServices(ctx context.Context, orgID, da // Set patrol trigger function (triggers mini-patrol on alert events) patrol := aiService.GetPatrolService() - if patrol != nil { + if patrol != nil && ai.BackgroundAutomationDisabledForDev() { + log.Info(). + Str("env", ai.DevDisableBackgroundAIEnv). + Str("org_id", orgID). + Msg("Pulse dev background AI disabled; alert bridge patrol trigger not wired") + } else if patrol != nil { alertBridge.SetPatrolTrigger(func(resourceID, resourceType, reason, alertType string) { scope := ai.PatrolScope{ ResourceIDs: []string{resourceID}, @@ -3579,6 +3584,13 @@ func (r *Router) WireAlertTriggeredAI() { return } + if ai.BackgroundAutomationDisabledForDev() { + log.Info(). + Str("env", ai.DevDisableBackgroundAIEnv). + Msg("Pulse dev background AI disabled; alert-triggered AI analyzer not wired") + return + } + // 3. Connect alert-fired events to the dedicated alert-triggered analyzer. // Patrol's event-triggered runs are owned by the canonical alert bridge / // trigger-manager path, so this callback should not enqueue Patrol directly. diff --git a/internal/config/watcher.go b/internal/config/watcher.go index 154414bd6..7bd298324 100644 --- a/internal/config/watcher.go +++ b/internal/config/watcher.go @@ -26,7 +26,7 @@ var ( debounceAPITokensWrite = 250 * time.Millisecond ) -// ConfigWatcher monitors the .env file for changes and updates runtime config +// ConfigWatcher monitors runtime config files for changes and updates runtime config. type ConfigWatcher struct { config *Config envPath string @@ -147,38 +147,80 @@ func (cw *ConfigWatcher) SetAPITokenReloadCallback(callback func()) { cw.onAPITokenReload = callback } -// Start begins watching the config file +// Start begins watching the config files. func (cw *ConfigWatcher) Start() error { - // Watch the directory for .env - dir := filepath.Dir(cw.envPath) - err := cw.watcher.Add(dir) - if err != nil { - log.Warn().Err(err).Str("path", dir).Msg("Failed to watch config directory") + watched := 0 + for _, path := range []string{cw.envPath, cw.apiTokensPath} { + ok, err := cw.addFileWatchIfPresent(path) + if err != nil { + log.Warn().Err(err).Str("path", path).Msg("Failed to watch config file") + continue + } + if ok { + watched++ + } } - if err != nil { - log.Warn(). - Err(err). - Str("env_path", cw.envPath). - Dur("poll_interval", cw.pollInterval). - Msg("Falling back to polling for config changes") + if watched > 0 { cw.wg.Add(1) go func() { defer cw.wg.Done() - cw.pollForChanges() + cw.watchForChanges() }() - return nil + log.Info(). + Str("env_path", cw.envPath). + Str("api_tokens_path", cw.apiTokensPath). + Dur("poll_interval", cw.pollInterval). + Msg("Started watching config files for changes") + } else { + log.Info(). + Str("env_path", cw.envPath). + Str("api_tokens_path", cw.apiTokensPath). + Dur("poll_interval", cw.pollInterval). + Msg("Config files not present yet; polling for config changes") } cw.wg.Add(1) go func() { defer cw.wg.Done() - cw.watchForChanges() + cw.pollForChanges() }() - log.Info().Str("env_path", cw.envPath).Str("api_tokens_path", cw.apiTokensPath).Msg("Started watching config files for changes") return nil } +func (cw *ConfigWatcher) addFileWatchIfPresent(path string) (bool, error) { + if cw == nil || cw.watcher == nil || strings.TrimSpace(path) == "" { + return false, nil + } + stat, err := watcherOsStat(path) + if err != nil { + if os.IsNotExist(err) { + return false, nil + } + return false, err + } + if stat.IsDir() { + return false, nil + } + if err := cw.watcher.Add(path); err != nil { + return false, err + } + return true, nil +} + +func (cw *ConfigWatcher) rewatchFileIfPresent(path string) { + if cw == nil || cw.watcher == nil || strings.TrimSpace(path) == "" { + return + } + if _, err := watcherOsStat(path); err != nil { + return + } + _ = cw.watcher.Remove(path) + if err := cw.watcher.Add(path); err != nil { + log.Debug().Err(err).Str("path", path).Msg("Failed to re-arm config file watch") + } +} + // Stop stops the config watcher func (cw *ConfigWatcher) Stop() { // Use sync.Once to prevent double-close panic @@ -211,13 +253,14 @@ func (cw *ConfigWatcher) handleEvents(events <-chan fsnotify.Event, errors <-cha } // Check if the event is for our .env file - if filepath.Base(event.Name) == ".env" || event.Name == cw.envPath { + if configWatcherEventMatchesPath(event, cw.envPath) { // Debounce - wait a bit for write to complete if !cw.waitForDuration(debounceEnvWrite) { return } - if event.Op&(fsnotify.Write|fsnotify.Create) != 0 { + if event.Op&(fsnotify.Write|fsnotify.Create|fsnotify.Rename|fsnotify.Remove) != 0 { + cw.rewatchFileIfPresent(cw.envPath) // Check if content actually changed to prevent restart loops on touch newHash, err := cw.calculateFileHash(cw.envPath) if err == nil { @@ -226,6 +269,9 @@ func (cw *ConfigWatcher) handleEvents(events <-chan fsnotify.Event, errors <-cha continue } cw.lastEnvHash = newHash + if stat, statErr := watcherOsStat(cw.envPath); statErr == nil { + cw.lastModTime = stat.ModTime() + } } log.Info().Str("event", event.Op.String()).Msg("Detected .env file change") @@ -233,14 +279,18 @@ func (cw *ConfigWatcher) handleEvents(events <-chan fsnotify.Event, errors <-cha } } - if cw.apiTokensPath != "" && (filepath.Base(event.Name) == filepath.Base(cw.apiTokensPath) || event.Name == cw.apiTokensPath) { + if configWatcherEventMatchesPath(event, cw.apiTokensPath) { // Debounce - wait longer for atomic file operations to complete // (write to .tmp, rename to final file) if !cw.waitForDuration(debounceAPITokensWrite) { return } - if event.Op&(fsnotify.Write|fsnotify.Create) != 0 { + if event.Op&(fsnotify.Write|fsnotify.Create|fsnotify.Rename|fsnotify.Remove) != 0 { + cw.rewatchFileIfPresent(cw.apiTokensPath) + if stat, statErr := watcherOsStat(cw.apiTokensPath); statErr == nil { + cw.apiTokensLastModTime = stat.ModTime() + } log.Info().Str("event", event.Op.String()).Msg("Detected API token file change") cw.reloadAPITokens() } @@ -262,6 +312,13 @@ func (cw *ConfigWatcher) handleEvents(events <-chan fsnotify.Event, errors <-cha } } +func configWatcherEventMatchesPath(event fsnotify.Event, path string) bool { + if strings.TrimSpace(path) == "" { + return false + } + return event.Name == path || filepath.Base(event.Name) == filepath.Base(path) +} + // waitForDuration blocks for d unless shutdown is requested first. func (cw *ConfigWatcher) waitForDuration(d time.Duration) bool { if d <= 0 { diff --git a/internal/config/watcher_test.go b/internal/config/watcher_test.go index 97f86419c..401e2c269 100644 --- a/internal/config/watcher_test.go +++ b/internal/config/watcher_test.go @@ -107,6 +107,29 @@ func TestConfigWatcher_StopWaitsForBackgroundWorker(t *testing.T) { } } +func TestConfigWatcher_StartWatchesFilesWithoutDirectoryFanout(t *testing.T) { + tempDir := t.TempDir() + envPath := filepath.Join(tempDir, ".env") + apiTokensPath := filepath.Join(tempDir, "api_tokens.json") + require.NoError(t, os.WriteFile(envPath, []byte("PULSE_AUTH_USER=admin\n"), 0644)) + require.NoError(t, os.WriteFile(apiTokensPath, []byte("[]"), 0644)) + require.NoError(t, os.WriteFile(filepath.Join(tempDir, ".env.tmp.123"), []byte("stale"), 0644)) + + t.Setenv("PULSE_AUTH_CONFIG_DIR", tempDir) + cw, err := NewConfigWatcher(&Config{ConfigPath: tempDir}) + require.NoError(t, err) + cw.pollInterval = time.Hour + + require.NoError(t, cw.Start()) + defer cw.Stop() + + watched := cw.watcher.WatchList() + assert.Contains(t, watched, envPath) + assert.Contains(t, watched, apiTokensPath) + assert.NotContains(t, watched, tempDir) + assert.NotContains(t, watched, filepath.Join(tempDir, ".env.tmp.123")) +} + func TestConfigWatcher_ReloadConfig(t *testing.T) { // Setup tempDir := t.TempDir() diff --git a/internal/monitoring/canonical_guardrails_test.go b/internal/monitoring/canonical_guardrails_test.go index f8392e517..4264c7bf7 100644 --- a/internal/monitoring/canonical_guardrails_test.go +++ b/internal/monitoring/canonical_guardrails_test.go @@ -138,6 +138,21 @@ func TestGetStateRefreshesLiveAlertSnapshots(t *testing.T) { } } +func TestAlertStateSyncDoesNotLogNormalResolvedStateAtInfo(t *testing.T) { + data, err := os.ReadFile("monitor_alert_sync.go") + if err != nil { + t.Fatalf("failed to read monitor_alert_sync.go: %v", err) + } + source := string(data) + + if !strings.Contains(source, `log.Debug().Int("count", len(recentlyResolved)).Msg("syncing recently resolved alerts")`) { + t.Fatal("recently resolved alert sync must stay at debug level to avoid info-log churn") + } + if strings.Contains(source, `log.Info().Int("count", len(recentlyResolved)).Msg("syncing recently resolved alerts")`) { + t.Fatal("recently resolved alert sync must not log normal polling state at info level") + } +} + func TestStateBroadcastTreatsTypedNilHubAsAbsent(t *testing.T) { data, err := os.ReadFile("monitor.go") if err != nil { diff --git a/internal/monitoring/monitor_alert_sync.go b/internal/monitoring/monitor_alert_sync.go index 5f8f44243..74b03234b 100644 --- a/internal/monitoring/monitor_alert_sync.go +++ b/internal/monitoring/monitor_alert_sync.go @@ -28,8 +28,8 @@ func (m *Monitor) syncAlertsToState() { m.state.UpdateActiveAlerts(modelAlerts) recentlyResolved := m.alertManager.GetRecentlyResolved() - if len(recentlyResolved) > 0 { - log.Info().Int("count", len(recentlyResolved)).Msg("syncing recently resolved alerts") + if len(recentlyResolved) > 0 && logging.IsLevelEnabled(zerolog.DebugLevel) { + log.Debug().Int("count", len(recentlyResolved)).Msg("syncing recently resolved alerts") } m.state.UpdateRecentlyResolved(recentlyResolved) } diff --git a/internal/monitoring/monitor_alerts.go b/internal/monitoring/monitor_alerts.go index ec3228f92..9bb708116 100644 --- a/internal/monitoring/monitor_alerts.go +++ b/internal/monitoring/monitor_alerts.go @@ -292,16 +292,16 @@ func (m *Monitor) broadcastStateUpdate() { func (m *Monitor) checkMockAlerts() { defer recoverFromPanic("checkMockAlerts") - log.Info().Bool("mockEnabled", mock.IsMockEnabled()).Msg("checkMockAlerts called") + log.Debug().Bool("mockEnabled", mock.IsMockEnabled()).Msg("checkMockAlerts called") if !mock.IsMockEnabled() { - log.Info().Msg("mock mode not enabled, skipping mock alert check") + log.Debug().Msg("mock mode not enabled, skipping mock alert check") return } // Get mock state state := mock.CurrentFixtureGraph().State - log.Info(). + log.Debug(). Int("vms", len(state.VMs)). Int("containers", len(state.Containers)). Int("nodes", len(state.Nodes)). @@ -322,7 +322,7 @@ func (m *Monitor) checkMockAlerts() { existingNodes[pbsInst.Host] = true } } - log.Info(). + log.Debug(). Int("trackedNodes", len(existingNodes)). Msg("Collecting resources for alert cleanup in mock mode") m.alertManager.CleanupAlertsForNodes(existingNodes) @@ -367,7 +367,7 @@ func (m *Monitor) checkMockAlerts() { } // Check alerts for storage - log.Info().Int("storageCount", len(state.Storage)).Msg("checking storage alerts") + log.Debug().Int("storageCount", len(state.Storage)).Msg("checking storage alerts") for _, storage := range state.Storage { log.Debug(). Str("name", storage.Name). @@ -377,13 +377,13 @@ func (m *Monitor) checkMockAlerts() { } // Check alerts for PBS instances - log.Info().Int("pbsCount", len(state.PBSInstances)).Msg("checking PBS alerts") + log.Debug().Int("pbsCount", len(state.PBSInstances)).Msg("checking PBS alerts") for _, pbsInst := range state.PBSInstances { m.alertManager.CheckPBS(pbsInst) } // Check alerts for PMG instances - log.Info().Int("pmgCount", len(state.PMGInstances)).Msg("checking PMG alerts") + log.Debug().Int("pmgCount", len(state.PMGInstances)).Msg("checking PMG alerts") for _, pmgInst := range state.PMGInstances { m.alertManager.CheckPMG(pmgInst) } diff --git a/internal/unifiedresources/adapters.go b/internal/unifiedresources/adapters.go index 9549fac77..5785bf0dd 100644 --- a/internal/unifiedresources/adapters.go +++ b/internal/unifiedresources/adapters.go @@ -1328,6 +1328,8 @@ func resourceFromPMGInstance(instance models.PMGInstance) (Resource, ResourceIde PMG: &PMGData{ InstanceID: instance.ID, Hostname: extractHostname(instance.Host), + HostURL: strings.TrimSpace(instance.Host), + GuestURL: strings.TrimSpace(instance.GuestURL), Version: instance.Version, NodeCount: len(instance.Nodes), UptimeSeconds: uptime, @@ -1360,11 +1362,13 @@ func resourceFromPMGInstance(instance models.PMGInstance) (Resource, ResourceIde } if n.QueueStatus != nil { nodes[i].QueueStatus = &PMGQueueMeta{ - Active: n.QueueStatus.Active, - Deferred: n.QueueStatus.Deferred, - Hold: n.QueueStatus.Hold, - Incoming: n.QueueStatus.Incoming, - Total: n.QueueStatus.Total, + Active: n.QueueStatus.Active, + Deferred: n.QueueStatus.Deferred, + Hold: n.QueueStatus.Hold, + Incoming: n.QueueStatus.Incoming, + Total: n.QueueStatus.Total, + OldestAge: n.QueueStatus.OldestAge, + UpdatedAt: n.QueueStatus.UpdatedAt, } } } @@ -1375,6 +1379,7 @@ func resourceFromPMGInstance(instance models.PMGInstance) (Resource, ResourceIde if instance.MailStats != nil { resource.PMG.MailStats = &PMGMailStatsMeta{ Timeframe: instance.MailStats.Timeframe, + CountTotal: instance.MailStats.CountTotal, CountIn: instance.MailStats.CountIn, CountOut: instance.MailStats.CountOut, SpamIn: instance.MailStats.SpamIn, @@ -1386,11 +1391,18 @@ func resourceFromPMGInstance(instance models.PMGInstance) (Resource, ResourceIde BytesIn: instance.MailStats.BytesIn, BytesOut: instance.MailStats.BytesOut, GreylistCount: instance.MailStats.GreylistCount, + JunkIn: instance.MailStats.JunkIn, RBLRejects: instance.MailStats.RBLRejects, AverageProcessTimeMs: instance.MailStats.AverageProcessTimeMs, + PregreetRejects: instance.MailStats.PregreetRejects, + UpdatedAt: instance.MailStats.UpdatedAt, } } + if len(instance.MailCount) > 0 { + resource.PMG.MailCount = append([]models.PMGMailCountPoint(nil), instance.MailCount...) + } + // Populate quarantine if instance.Quarantine != nil { resource.PMG.Quarantine = &PMGQuarantineMeta{ diff --git a/internal/unifiedresources/canonical_identity_test.go b/internal/unifiedresources/canonical_identity_test.go index b91af844e..c7a38926f 100644 --- a/internal/unifiedresources/canonical_identity_test.go +++ b/internal/unifiedresources/canonical_identity_test.go @@ -103,6 +103,33 @@ func TestRefreshCanonicalIdentityFallsBackWithoutTargets(t *testing.T) { } } +func TestRefreshCanonicalIdentityUsesPMGInstanceIdentity(t *testing.T) { + resource := Resource{ + ID: "pmg-resource-1", + Type: ResourceTypePMG, + Name: "", + PMG: &PMGData{ + InstanceID: "pmg-main", + Hostname: "mail.example", + }, + } + + RefreshCanonicalIdentity(&resource) + + if resource.Canonical == nil { + t.Fatalf("expected canonical identity") + } + if got := resource.Canonical.DisplayName; got != "mail.example" { + t.Fatalf("displayName = %q, want mail.example", got) + } + if got := resource.Canonical.Hostname; got != "mail.example" { + t.Fatalf("hostname = %q, want mail.example", got) + } + if got := resource.Canonical.PrimaryID; got != "pmg:pmg-main" { + t.Fatalf("primaryId = %q, want pmg:pmg-main", got) + } +} + func TestRefreshCanonicalIdentityUsesAvailabilityTargetIdentity(t *testing.T) { resource := Resource{ ID: "availability:energy-meter", diff --git a/internal/unifiedresources/clone.go b/internal/unifiedresources/clone.go index 15d73c6d8..5fc2b006a 100644 --- a/internal/unifiedresources/clone.go +++ b/internal/unifiedresources/clone.go @@ -241,6 +241,7 @@ func clonePMGData(in *PMGData) *PMGData { out := *in out.Nodes = clonePMGNodeMetaSlice(in.Nodes) out.MailStats = clonePMGMailStatsMeta(in.MailStats) + out.MailCount = append([]models.PMGMailCountPoint(nil), in.MailCount...) out.Quarantine = clonePMGQuarantineMeta(in.Quarantine) out.SpamDistribution = clonePMGSpamBucketMetaSlice(in.SpamDistribution) out.RelayDomains = clonePMGRelayDomainMetaSlice(in.RelayDomains) diff --git a/internal/unifiedresources/monitored_system_projection.go b/internal/unifiedresources/monitored_system_projection.go index 554e90b64..9fa5359cf 100644 --- a/internal/unifiedresources/monitored_system_projection.go +++ b/internal/unifiedresources/monitored_system_projection.go @@ -792,6 +792,7 @@ func monitoredSystemCandidateResource(candidate MonitoredSystemCandidate) *Resou resource.PMG = &PMGData{ InstanceID: strings.TrimSpace(candidate.ResourceID), Hostname: host, + HostURL: strings.TrimSpace(candidate.HostURL), } case SourceK8s: clusterName := firstTrimmed(candidate.Name, host, candidate.ResourceID) diff --git a/internal/unifiedresources/pbs_pmg_registry_test.go b/internal/unifiedresources/pbs_pmg_registry_test.go index dff8907a6..475648339 100644 --- a/internal/unifiedresources/pbs_pmg_registry_test.go +++ b/internal/unifiedresources/pbs_pmg_registry_test.go @@ -55,18 +55,23 @@ func TestIngestSnapshotIncludesPBSAndPMGInstances(t *testing.T) { ID: "pmg-1", Name: "pmg-main", Host: "https://pmg.example.com:8006", + GuestURL: "https://pmg.example.com/quarantine", Status: "online", Version: "8.2", ConnectionHealth: "connected", LastSeen: now, LastUpdated: now, MailStats: &models.PMGMailStats{ - CountTotal: 1900, - BytesIn: 5_000_000, - BytesOut: 4_000_000, - SpamIn: 125, - VirusIn: 4, + CountTotal: 1900, + BytesIn: 5_000_000, + BytesOut: 4_000_000, + SpamIn: 125, + VirusIn: 4, + JunkIn: 32, + PregreetRejects: 7, + UpdatedAt: now, }, + MailCount: []models.PMGMailCountPoint{{Timestamp: now, Count: 1900, CountIn: 1200, CountOut: 700, Timeframe: "hour", Index: 1}}, Nodes: []models.PMGNodeStatus{ { Name: "pmg-node-1", @@ -78,6 +83,7 @@ func TestIngestSnapshotIncludesPBSAndPMGInstances(t *testing.T) { Hold: 2, Incoming: 3, Total: 22, + OldestAge: 1800, UpdatedAt: now, }, }, @@ -172,4 +178,16 @@ func TestIngestSnapshotIncludesPBSAndPMGInstances(t *testing.T) { if pmgResource.PMG == nil || pmgResource.PMG.QueueTotal != 22 { t.Fatalf("expected PMG payload with queue totals, got %+v", pmgResource.PMG) } + if pmgResource.PMG.HostURL != "https://pmg.example.com:8006" || pmgResource.PMG.GuestURL != "https://pmg.example.com/quarantine" { + t.Fatalf("expected PMG URL payloads, got %+v", pmgResource.PMG) + } + if len(pmgResource.PMG.Nodes) != 1 || pmgResource.PMG.Nodes[0].QueueStatus == nil || pmgResource.PMG.Nodes[0].QueueStatus.OldestAge != 1800 { + t.Fatalf("expected PMG queue oldest age, got %+v", pmgResource.PMG.Nodes) + } + if pmgResource.PMG.MailStats == nil || pmgResource.PMG.MailStats.CountTotal != 1900 || pmgResource.PMG.MailStats.JunkIn != 32 || pmgResource.PMG.MailStats.PregreetRejects != 7 { + t.Fatalf("expected full PMG mail stats payload, got %+v", pmgResource.PMG.MailStats) + } + if len(pmgResource.PMG.MailCount) != 1 || pmgResource.PMG.MailCount[0].Index != 1 { + t.Fatalf("expected PMG mail count points, got %+v", pmgResource.PMG.MailCount) + } } diff --git a/internal/unifiedresources/types.go b/internal/unifiedresources/types.go index fa061414c..02e1f7cb2 100644 --- a/internal/unifiedresources/types.go +++ b/internal/unifiedresources/types.go @@ -901,29 +901,35 @@ type PMGNodeMeta struct { // PMGQueueMeta describes a PMG node's postfix queue status. type PMGQueueMeta struct { - Active int `json:"active"` - Deferred int `json:"deferred"` - Hold int `json:"hold"` - Incoming int `json:"incoming"` - Total int `json:"total"` + Active int `json:"active"` + Deferred int `json:"deferred"` + Hold int `json:"hold"` + Incoming int `json:"incoming"` + Total int `json:"total"` + OldestAge int64 `json:"oldestAge,omitempty"` + UpdatedAt time.Time `json:"updatedAt,omitempty"` } // PMGMailStatsMeta describes PMG mail statistics. type PMGMailStatsMeta struct { - Timeframe string `json:"timeframe"` - CountIn float64 `json:"countIn"` - CountOut float64 `json:"countOut"` - SpamIn float64 `json:"spamIn"` - SpamOut float64 `json:"spamOut"` - VirusIn float64 `json:"virusIn"` - VirusOut float64 `json:"virusOut"` - BouncesIn float64 `json:"bouncesIn"` - BouncesOut float64 `json:"bouncesOut"` - BytesIn float64 `json:"bytesIn"` - BytesOut float64 `json:"bytesOut"` - GreylistCount float64 `json:"greylistCount"` - RBLRejects float64 `json:"rblRejects"` - AverageProcessTimeMs float64 `json:"averageProcessTimeMs"` + Timeframe string `json:"timeframe"` + CountTotal float64 `json:"countTotal"` + CountIn float64 `json:"countIn"` + CountOut float64 `json:"countOut"` + SpamIn float64 `json:"spamIn"` + SpamOut float64 `json:"spamOut"` + VirusIn float64 `json:"virusIn"` + VirusOut float64 `json:"virusOut"` + BouncesIn float64 `json:"bouncesIn"` + BouncesOut float64 `json:"bouncesOut"` + BytesIn float64 `json:"bytesIn"` + BytesOut float64 `json:"bytesOut"` + GreylistCount float64 `json:"greylistCount"` + JunkIn float64 `json:"junkIn"` + RBLRejects float64 `json:"rblRejects"` + AverageProcessTimeMs float64 `json:"averageProcessTimeMs"` + PregreetRejects float64 `json:"pregreetRejects"` + UpdatedAt time.Time `json:"updatedAt,omitempty"` } // PMGQuarantineMeta describes PMG quarantine totals. @@ -960,6 +966,8 @@ type PMGDomainStatMeta struct { type PMGData struct { InstanceID string `json:"instanceId,omitempty"` Hostname string `json:"hostname,omitempty"` + HostURL string `json:"hostUrl,omitempty"` + GuestURL string `json:"guestUrl,omitempty"` Version string `json:"version,omitempty"` NodeCount int `json:"nodeCount,omitempty"` UptimeSeconds int64 `json:"uptimeSeconds,omitempty"` @@ -974,13 +982,14 @@ type PMGData struct { ConnectionHealth string `json:"connectionHealth,omitempty"` LastUpdated time.Time `json:"lastUpdated,omitempty"` - Nodes []PMGNodeMeta `json:"nodes,omitempty"` - MailStats *PMGMailStatsMeta `json:"mailStats,omitempty"` - Quarantine *PMGQuarantineMeta `json:"quarantine,omitempty"` - SpamDistribution []PMGSpamBucketMeta `json:"spamDistribution,omitempty"` - RelayDomains []PMGRelayDomainMeta `json:"relayDomains,omitempty"` - DomainStats []PMGDomainStatMeta `json:"domainStats,omitempty"` - DomainStatsAsOf time.Time `json:"domainStatsAsOf,omitempty"` + Nodes []PMGNodeMeta `json:"nodes,omitempty"` + MailStats *PMGMailStatsMeta `json:"mailStats,omitempty"` + MailCount []models.PMGMailCountPoint `json:"mailCount,omitempty"` + Quarantine *PMGQuarantineMeta `json:"quarantine,omitempty"` + SpamDistribution []PMGSpamBucketMeta `json:"spamDistribution,omitempty"` + RelayDomains []PMGRelayDomainMeta `json:"relayDomains,omitempty"` + DomainStats []PMGDomainStatMeta `json:"domainStats,omitempty"` + DomainStatsAsOf time.Time `json:"domainStatsAsOf,omitempty"` } // VMwareData contains VMware vSphere metadata for canonical agent, vm, and diff --git a/internal/unifiedresources/views.go b/internal/unifiedresources/views.go index 971548709..e70aa9fd3 100644 --- a/internal/unifiedresources/views.go +++ b/internal/unifiedresources/views.go @@ -2629,7 +2629,21 @@ func (v PMGInstanceView) Hostname() string { if v.r == nil || v.r.PMG == nil { return "" } - return v.r.PMG.Hostname + return strings.TrimSpace(v.r.PMG.Hostname) +} + +func (v PMGInstanceView) HostURL() string { + if v.r == nil || v.r.PMG == nil { + return "" + } + return strings.TrimSpace(v.r.PMG.HostURL) +} + +func (v PMGInstanceView) GuestURL() string { + if v.r == nil || v.r.PMG == nil { + return "" + } + return strings.TrimSpace(v.r.PMG.GuestURL) } func (v PMGInstanceView) Version() string { @@ -2674,6 +2688,13 @@ func (v PMGInstanceView) QueueHold() int { return v.r.PMG.QueueHold } +func (v PMGInstanceView) QueueIncoming() int { + if v.r == nil || v.r.PMG == nil { + return 0 + } + return v.r.PMG.QueueIncoming +} + func (v PMGInstanceView) QueueTotal() int { if v.r == nil || v.r.PMG == nil { return 0 @@ -2765,6 +2786,13 @@ func (v PMGInstanceView) MailStats() *PMGMailStatsMeta { return clonePMGMailStatsMeta(v.r.PMG.MailStats) } +func (v PMGInstanceView) MailCount() []models.PMGMailCountPoint { + if v.r == nil || v.r.PMG == nil { + return nil + } + return append([]models.PMGMailCountPoint(nil), v.r.PMG.MailCount...) +} + func (v PMGInstanceView) Quarantine() *PMGQuarantineMeta { if v.r == nil || v.r.PMG == nil { return nil @@ -2779,6 +2807,34 @@ func (v PMGInstanceView) SpamDistribution() []PMGSpamBucketMeta { return clonePMGSpamBucketMetaSlice(v.r.PMG.SpamDistribution) } +func (v PMGInstanceView) LastUpdated() time.Time { + if v.r == nil || v.r.PMG == nil { + return time.Time{} + } + return v.r.PMG.LastUpdated +} + +func (v PMGInstanceView) RelayDomains() []PMGRelayDomainMeta { + if v.r == nil || v.r.PMG == nil { + return nil + } + return clonePMGRelayDomainMetaSlice(v.r.PMG.RelayDomains) +} + +func (v PMGInstanceView) DomainStats() []PMGDomainStatMeta { + if v.r == nil || v.r.PMG == nil { + return nil + } + return clonePMGDomainStatMetaSlice(v.r.PMG.DomainStats) +} + +func (v PMGInstanceView) DomainStatsAsOf() time.Time { + if v.r == nil || v.r.PMG == nil { + return time.Time{} + } + return v.r.PMG.DomainStatsAsOf +} + // K8sClusterView wraps a Kubernetes cluster resource. type K8sClusterView struct{ r *Resource } diff --git a/internal/unifiedresources/views_test.go b/internal/unifiedresources/views_test.go index 0cc593768..04483bd22 100644 --- a/internal/unifiedresources/views_test.go +++ b/internal/unifiedresources/views_test.go @@ -1329,6 +1329,9 @@ func TestView_PBSAndPMGInstanceViewAccessors(t *testing.T) { }) t.Run("PMGInstanceView", func(t *testing.T) { + queueUpdated := now.Add(-2 * time.Minute) + statsUpdated := now.Add(-1 * time.Minute) + domainStatsAsOf := now.Add(-30 * time.Second) r := &Resource{ ID: "pmg-1", Type: ResourceTypePMG, @@ -1339,17 +1342,67 @@ func TestView_PBSAndPMGInstanceViewAccessors(t *testing.T) { CustomURL: "https://pmg.example/ui", PMG: &PMGData{ Hostname: "pmg.example", + HostURL: "https://pmg.example:8006", + GuestURL: "https://pmg-guest.example:8006", Version: "8.0", NodeCount: 2, UptimeSeconds: 200, QueueActive: 10, QueueDeferred: 20, QueueHold: 15, + QueueIncoming: 4, QueueTotal: 30, MailCountTotal: 1000, SpamIn: 11, VirusIn: 22, ConnectionHealth: "online", + LastUpdated: now, + Nodes: []PMGNodeMeta{{ + Name: "pmg-node-1", + Status: "online", + Role: "master", + Uptime: 123, + QueueStatus: &PMGQueueMeta{ + Active: 1, + Deferred: 2, + Hold: 3, + Incoming: 4, + Total: 10, + OldestAge: 55, + UpdatedAt: queueUpdated, + }, + }}, + MailStats: &PMGMailStatsMeta{ + Timeframe: "day", + CountTotal: 1000, + CountIn: 700, + CountOut: 300, + SpamIn: 11, + SpamOut: 2, + VirusIn: 22, + VirusOut: 1, + BouncesIn: 3, + BouncesOut: 4, + BytesIn: 500, + BytesOut: 600, + GreylistCount: 7, + JunkIn: 8, + RBLRejects: 9, + AverageProcessTimeMs: 10, + PregreetRejects: 12, + UpdatedAt: statsUpdated, + }, + MailCount: []models.PMGMailCountPoint{{ + Timestamp: statsUpdated, + Count: 1000, + Timeframe: "hour", + Index: 1, + }}, + Quarantine: &PMGQuarantineMeta{Spam: 1, Virus: 2, Attachment: 3, Blacklisted: 4}, + SpamDistribution: []PMGSpamBucketMeta{{Bucket: "5", Count: 6}}, + RelayDomains: []PMGRelayDomainMeta{{Domain: "example.com", Comment: "primary"}}, + DomainStats: []PMGDomainStatMeta{{Domain: "example.com", MailCount: 100, SpamCount: 5, VirusCount: 1, Bytes: 2048}}, + DomainStatsAsOf: domainStatsAsOf, }, Metrics: &ResourceMetrics{ CPU: &MetricValue{Percent: 4}, @@ -1361,15 +1414,39 @@ func TestView_PBSAndPMGInstanceViewAccessors(t *testing.T) { if v.ID() != "pmg-1" || v.Name() != "pmg-a" || v.Status() != StatusWarning { t.Fatalf("expected basic accessors to match, got id=%q name=%q status=%q", v.ID(), v.Name(), v.Status()) } - if v.Hostname() != "pmg.example" || v.Version() != "8.0" || v.NodeCount() != 2 || v.UptimeSeconds() != 200 { - t.Fatalf("expected pmg fields to match, got host=%q ver=%q nodes=%d uptime=%d", v.Hostname(), v.Version(), v.NodeCount(), v.UptimeSeconds()) + if v.Hostname() != "pmg.example" || v.HostURL() != "https://pmg.example:8006" || v.GuestURL() != "https://pmg-guest.example:8006" || v.Version() != "8.0" || v.NodeCount() != 2 || v.UptimeSeconds() != 200 { + t.Fatalf("expected pmg fields to match, got host=%q hostURL=%q guestURL=%q ver=%q nodes=%d uptime=%d", v.Hostname(), v.HostURL(), v.GuestURL(), v.Version(), v.NodeCount(), v.UptimeSeconds()) } - if v.QueueActive() != 10 || v.QueueDeferred() != 20 || v.QueueHold() != 15 || v.QueueTotal() != 30 { - t.Fatalf("expected queue active/deferred/hold/total 10/20/15/30, got %d/%d/%d/%d", v.QueueActive(), v.QueueDeferred(), v.QueueHold(), v.QueueTotal()) + if v.QueueActive() != 10 || v.QueueDeferred() != 20 || v.QueueHold() != 15 || v.QueueIncoming() != 4 || v.QueueTotal() != 30 { + t.Fatalf("expected queue active/deferred/hold/incoming/total 10/20/15/4/30, got %d/%d/%d/%d/%d", v.QueueActive(), v.QueueDeferred(), v.QueueHold(), v.QueueIncoming(), v.QueueTotal()) } if v.MailCountTotal() != 1000 || v.SpamIn() != 11 || v.VirusIn() != 22 { t.Fatalf("expected mail stats total/spam/virus 1000/11/22, got %v/%v/%v", v.MailCountTotal(), v.SpamIn(), v.VirusIn()) } + if nodes := v.Nodes(); len(nodes) != 1 || nodes[0].QueueStatus == nil || nodes[0].QueueStatus.OldestAge != 55 || !nodes[0].QueueStatus.UpdatedAt.Equal(queueUpdated) { + t.Fatalf("expected node queue details to match, got %+v", nodes) + } + if stats := v.MailStats(); stats == nil || stats.CountTotal != 1000 || stats.JunkIn != 8 || stats.PregreetRejects != 12 || !stats.UpdatedAt.Equal(statsUpdated) { + t.Fatalf("expected full mail stats, got %+v", stats) + } + if points := v.MailCount(); len(points) != 1 || points[0].Count != 1000 || points[0].Index != 1 { + t.Fatalf("expected mail count points, got %+v", points) + } + if q := v.Quarantine(); q == nil || q.Spam != 1 || q.Blacklisted != 4 { + t.Fatalf("expected quarantine details, got %+v", q) + } + if buckets := v.SpamDistribution(); len(buckets) != 1 || buckets[0].Bucket != "5" || buckets[0].Count != 6 { + t.Fatalf("expected spam distribution, got %+v", buckets) + } + if domains := v.RelayDomains(); len(domains) != 1 || domains[0].Domain != "example.com" || domains[0].Comment != "primary" { + t.Fatalf("expected relay domains, got %+v", domains) + } + if stats := v.DomainStats(); len(stats) != 1 || stats[0].Domain != "example.com" || stats[0].Bytes != 2048 { + t.Fatalf("expected domain stats, got %+v", stats) + } + if !v.LastUpdated().Equal(now) || !v.DomainStatsAsOf().Equal(domainStatsAsOf) { + t.Fatalf("expected PMG timestamps to match, got lastUpdated=%v domainStatsAsOf=%v", v.LastUpdated(), v.DomainStatsAsOf()) + } if v.ConnectionHealth() != "online" { t.Fatalf("expected connection health %q, got %q", "online", v.ConnectionHealth()) } @@ -1386,7 +1463,7 @@ func TestView_PBSAndPMGInstanceViewAccessors(t *testing.T) { t.Run("NilResourceAndNilPMGAreSafe", func(t *testing.T) { var zero PMGInstanceView - if zero.QueueActive() != 0 || zero.QueueDeferred() != 0 || zero.QueueHold() != 0 || zero.QueueTotal() != 0 { + if zero.HostURL() != "" || zero.GuestURL() != "" || zero.QueueActive() != 0 || zero.QueueDeferred() != 0 || zero.QueueHold() != 0 || zero.QueueIncoming() != 0 || zero.QueueTotal() != 0 || zero.MailCount() != nil || !zero.LastUpdated().IsZero() || zero.RelayDomains() != nil || zero.DomainStats() != nil || !zero.DomainStatsAsOf().IsZero() { t.Fatalf("expected zero queue values for nil resource, got active=%d deferred=%d hold=%d total=%d", zero.QueueActive(), zero.QueueDeferred(), zero.QueueHold(), zero.QueueTotal()) } @@ -1394,7 +1471,7 @@ func TestView_PBSAndPMGInstanceViewAccessors(t *testing.T) { rNoPMG := testResource(ResourceTypePMG) rNoPMG.PMG = nil vNoPMG := NewPMGInstanceView(rNoPMG) - if vNoPMG.QueueActive() != 0 || vNoPMG.QueueDeferred() != 0 || vNoPMG.QueueHold() != 0 || vNoPMG.QueueTotal() != 0 { + if vNoPMG.HostURL() != "" || vNoPMG.GuestURL() != "" || vNoPMG.QueueActive() != 0 || vNoPMG.QueueDeferred() != 0 || vNoPMG.QueueHold() != 0 || vNoPMG.QueueIncoming() != 0 || vNoPMG.QueueTotal() != 0 || vNoPMG.MailCount() != nil || !vNoPMG.LastUpdated().IsZero() || vNoPMG.RelayDomains() != nil || vNoPMG.DomainStats() != nil || !vNoPMG.DomainStatsAsOf().IsZero() { t.Fatalf("expected zero queue values when PMG is nil, got active=%d deferred=%d hold=%d total=%d", vNoPMG.QueueActive(), vNoPMG.QueueDeferred(), vNoPMG.QueueHold(), vNoPMG.QueueTotal()) } diff --git a/scripts/hot-dev.sh b/scripts/hot-dev.sh index ce6ef8f20..a34ff308f 100755 --- a/scripts/hot-dev.sh +++ b/scripts/hot-dev.sh @@ -17,6 +17,9 @@ # PULSE_DATA_DIR=/path Override data directory # PULSE_DEV_API_PORT=7655 Backend API port (default: 7655) # FRONTEND_DEV_PORT=5173 Frontend dev server port (default: 5173) +# LOG_LEVEL=debug Opt into verbose backend logs (default: info) +# PULSE_DEV_DISABLE_BACKGROUND_AI=false +# Allow automatic Patrol/discovery/alert AI in dev # HOT_DEV_BACKEND_HEALTH_STARTUP_GRACE_SECONDS=180 # Backend /api/health grace after starts/restarts # HOT_DEV_BACKEND_UNHEALTHY_THRESHOLD=2 @@ -342,12 +345,14 @@ FRONTEND_PORT=${PULSE_DEV_API_PORT} PORT=${PULSE_DEV_API_PORT} PULSE_DEV=true # Enable development mode features (needed for admin bypass etc) ALLOW_ADMIN_BYPASS=1 # Allow X-Admin-Bypass header in dev mode +LOG_LEVEL="${LOG_LEVEL:-info}" +PULSE_DEV_DISABLE_BACKGROUND_AI="${PULSE_DEV_DISABLE_BACKGROUND_AI:-true}" # Managed dev credentials default to admin/adminadminadmin unless # HOT_DEV_AUTH_USER / HOT_DEV_AUTH_PASS explicitly override them. PULSE_AUTH_USER="$(hot_dev_resolve_auth_user)" PULSE_AUTH_PASS="$(hot_dev_resolve_auth_pass)" -export FRONTEND_PORT PULSE_DEV_API_PORT PORT PULSE_DEV ALLOW_ADMIN_BYPASS PULSE_AUTH_USER PULSE_AUTH_PASS +export FRONTEND_PORT PULSE_DEV_API_PORT PORT PULSE_DEV ALLOW_ADMIN_BYPASS PULSE_AUTH_USER PULSE_AUTH_PASS LOG_LEVEL PULSE_DEV_DISABLE_BACKGROUND_AI # Data Directory Setup # Keep one persistent data directory in both mock and real modes so real @@ -453,7 +458,8 @@ mark_backend_startup_grace() { start_backend_process() { mark_backend_startup_grace - LOG_LEVEL="${LOG_LEVEL:-debug}" \ + LOG_LEVEL="${LOG_LEVEL:-info}" \ + PULSE_DEV_DISABLE_BACKGROUND_AI="${PULSE_DEV_DISABLE_BACKGROUND_AI:-true}" \ FRONTEND_PORT="${PULSE_DEV_API_PORT:-7655}" \ PORT="${PULSE_DEV_API_PORT:-7655}" \ PULSE_DATA_DIR="${PULSE_DATA_DIR:-}" \ diff --git a/scripts/lib/hot-dev-auth.sh b/scripts/lib/hot-dev-auth.sh index 3a028b7a2..6f213fbc8 100644 --- a/scripts/lib/hot-dev-auth.sh +++ b/scripts/lib/hot-dev-auth.sh @@ -42,6 +42,19 @@ hot_dev_single_quote() { printf '%s' "$1" | sed "s/'/'\\\\''/g" } +hot_dev_cleanup_env_temp_files() { + local runtime_env=$1 + local env_dir + local env_base + + [[ -n "${runtime_env}" ]] || return 0 + env_dir="$(dirname "${runtime_env}")" + env_base="$(basename "${runtime_env}")" + [[ -d "${env_dir}" ]] || return 0 + + find "${env_dir}" -maxdepth 1 -type f \( -name "${env_base}.tmp.*" -o -name "${env_base}.audit.*" \) -exec rm -f {} + 2>/dev/null || true +} + hot_dev_sync_auth_env_file() { local runtime_env=$1 local auth_user=$2 @@ -49,6 +62,7 @@ hot_dev_sync_auth_env_file() { local tmp_file mkdir -p "$(dirname "${runtime_env}")" + hot_dev_cleanup_env_temp_files "${runtime_env}" tmp_file="${runtime_env}.tmp.$$" { diff --git a/scripts/lib/hot-dev-runtime.sh b/scripts/lib/hot-dev-runtime.sh index a1a75e2c3..f5355b834 100644 --- a/scripts/lib/hot-dev-runtime.sh +++ b/scripts/lib/hot-dev-runtime.sh @@ -76,6 +76,9 @@ hot_dev_configure_network_defaults() { ALLOWED_ORIGINS="http://${PULSE_DEV_API_HOST:-127.0.0.1}:${FRONTEND_DEV_PORT:-7655}" ALLOWED_ORIGINS="${ALLOWED_ORIGINS},http://localhost:${FRONTEND_DEV_PORT:-7655},http://127.0.0.1:${FRONTEND_DEV_PORT:-7655}" ALLOWED_ORIGINS="${ALLOWED_ORIGINS},http://localhost:5173,http://127.0.0.1:5173" + if [[ "${FRONTEND_DEV_HOST:-}" == "0.0.0.0" ]]; then + ALLOWED_ORIGINS="${ALLOWED_ORIGINS},http://0.0.0.0:${FRONTEND_DEV_PORT:-7655},http://0.0.0.0:5173" + fi local ip for ip in ${ALL_IPS:-}; do diff --git a/scripts/tests/test-hot-dev-auth.sh b/scripts/tests/test-hot-dev-auth.sh index d6ede3dc2..5d03aace7 100755 --- a/scripts/tests/test-hot-dev-auth.sh +++ b/scripts/tests/test-hot-dev-auth.sh @@ -104,6 +104,30 @@ EOF assert_contains "sync preserves mock settings" "${output}" "PULSE_MOCK_MODE=false" } +test_sync_auth_env_file_removes_stale_temp_files() { + local state_dir runtime_env output + state_dir="$(make_temp_dir)" + runtime_env="${state_dir}/.env" + + touch "${runtime_env}.tmp.111" "${runtime_env}.audit.222" + + output="$( + HOT_DEV_AUTH_LIB="${HOT_DEV_AUTH_LIB}" \ + RUNTIME_ENV_PATH="${runtime_env}" \ + bash -lc ' + source "${HOT_DEV_AUTH_LIB}" + hot_dev_sync_auth_env_file "${RUNTIME_ENV_PATH}" "admin" "${HOT_DEV_DEFAULT_AUTH_HASH}" + if compgen -G "${RUNTIME_ENV_PATH}.tmp.*" >/dev/null || compgen -G "${RUNTIME_ENV_PATH}.audit.*" >/dev/null; then + printf "stale_temp_remaining=yes\n" + else + printf "stale_temp_remaining=no\n" + fi + ' + )" + + assert_contains "sync removes stale managed temp files" "${output}" "stale_temp_remaining=no" +} + test_sync_auth_env_file_handles_managed_only_env_under_errexit() { local state_dir runtime_env output state_dir="$(make_temp_dir)" @@ -175,6 +199,7 @@ source "${HOT_DEV_AUTH_LIB}" test_default_auth_contract test_custom_auth_banner_contract test_sync_auth_env_file_preserves_non_auth_settings +test_sync_auth_env_file_removes_stale_temp_files test_sync_auth_env_file_handles_managed_only_env_under_errexit test_sync_audit_signing_env_file_restores_missing_key diff --git a/scripts/tests/test-hot-dev-runtime.sh b/scripts/tests/test-hot-dev-runtime.sh index 12fa0d36c..fb0f5ad76 100755 --- a/scripts/tests/test-hot-dev-runtime.sh +++ b/scripts/tests/test-hot-dev-runtime.sh @@ -123,6 +123,8 @@ test_hot_dev_keeps_backend_launch_errors_in_debug_log() { output="$(sed -n '1,760p' "${HOT_DEV}")" assert_contains "hot-dev exposes backend debug log override" "${output}" "BACKEND_DEBUG_LOG=\"\${PULSE_BACKEND_LOG_FILE:-/tmp/pulse-debug.log}\"" + assert_contains "hot-dev defaults backend log level to info" "${output}" "LOG_LEVEL=\"\${LOG_LEVEL:-info}\"" + assert_contains "hot-dev disables background AI automation by default" "${output}" "PULSE_DEV_DISABLE_BACKGROUND_AI=\"\${PULSE_DEV_DISABLE_BACKGROUND_AI:-true}\"" assert_contains "backend launch stderr is appended to the debug log" "${output}" ">> \"\${BACKEND_DEBUG_LOG}\" 2>&1" assert_contains "backend LOG_FILE follows debug log override" "${output}" "LOG_FILE=\"\${BACKEND_DEBUG_LOG}\"" } @@ -165,6 +167,7 @@ test_hot_dev_network_defaults_are_lan_capable() { assert_contains "network defaults derive the websocket URL" "${output}" "ws_url=ws://192.168.50.10:7655" assert_contains "network defaults allow the detected LAN frontend origin" "${output}" "http://192.168.50.10:5173" assert_contains "network defaults retain loopback frontend origins" "${output}" "http://127.0.0.1:5173" + assert_contains "network defaults allow the wildcard dev origin Electron may report" "${output}" "http://0.0.0.0:5173" } test_hot_dev_browser_urls_distinguish_bind_and_browser_hosts() {