From f6d31d166a7792b93531bda213a36da04f5604c2 Mon Sep 17 00:00:00 2001 From: rcourtman Date: Fri, 12 Dec 2025 14:26:10 +0000 Subject: [PATCH] feat(ai): Add multi-resource correlation detection (Phase 6) Create internal/ai/correlation package: 1. Correlation Detector (detector.go): - Tracks events across resources - Detects when events on one resource follow events on another - Calculates average delay between correlated events - Confidence scoring based on occurrence count - Persists to ai_correlations.json 2. Features: - GetCorrelations() - All detected relationships - GetCorrelationsForResource() - Relationships for one resource - GetDependencies() - What resources depend on this one - GetDependsOn() - What this resource depends on - PredictCascade() - Predict what will be affected - FormatForContext() - AI-consumable summary 3. Integration: - Wire to alert history in router startup - Map alert types to correlation event types - Add correlation context to enriched AI context Example AI context now includes: 'When local-zfs experiences high usage, database often follows within 5 minutes' This enables the AI to understand infrastructure dependencies and predict cascade failures. All tests passing. --- internal/ai/correlation/detector.go | 501 +++++++++++++++++++++++ internal/ai/correlation/detector_test.go | 246 +++++++++++ internal/ai/correlation_exports.go | 44 ++ internal/ai/patrol.go | 47 ++- internal/ai/service.go | 11 + internal/api/ai_handlers.go | 5 + internal/api/router.go | 39 ++ 7 files changed, 882 insertions(+), 11 deletions(-) create mode 100644 internal/ai/correlation/detector.go create mode 100644 internal/ai/correlation/detector_test.go create mode 100644 internal/ai/correlation_exports.go diff --git a/internal/ai/correlation/detector.go b/internal/ai/correlation/detector.go new file mode 100644 index 0000000..219d680 --- /dev/null +++ b/internal/ai/correlation/detector.go @@ -0,0 +1,501 @@ +// Package correlation detects relationships between resources. +// It tracks when events on one resource are followed by events on another, +// enabling the AI to understand dependencies and predict cascade failures. +package correlation + +import ( + "encoding/json" + "os" + "path/filepath" + "sort" + "sync" + "time" + + "github.com/rs/zerolog/log" +) + +// EventType represents the type of event being correlated +type EventType string + +const ( + EventAlert EventType = "alert" // Alert triggered + EventRestart EventType = "restart" // Resource restarted + EventHighCPU EventType = "high_cpu" // CPU spike + EventHighMem EventType = "high_mem" // Memory spike + EventDiskFull EventType = "disk_full" // Disk space critical + EventOffline EventType = "offline" // Resource went offline + EventMigration EventType = "migration" // Resource migrated +) + +// Event represents a tracked event for correlation analysis +type Event struct { + ID string `json:"id"` + ResourceID string `json:"resource_id"` + ResourceName string `json:"resource_name,omitempty"` + ResourceType string `json:"resource_type,omitempty"` // node, vm, container, storage + EventType EventType `json:"event_type"` + Timestamp time.Time `json:"timestamp"` + Value float64 `json:"value,omitempty"` // Metric value if applicable +} + +// Correlation represents a detected relationship between two resources +type Correlation struct { + SourceID string `json:"source_id"` // Resource that triggers + SourceName string `json:"source_name"` + SourceType string `json:"source_type"` + TargetID string `json:"target_id"` // Resource that follows + TargetName string `json:"target_name"` + TargetType string `json:"target_type"` + EventPattern string `json:"event_pattern"` // e.g., "high_mem -> restart" + Occurrences int `json:"occurrences"` // Number of times observed + AvgDelay time.Duration `json:"avg_delay"` // Average time between events + Confidence float64 `json:"confidence"` // 0-1 confidence level + LastSeen time.Time `json:"last_seen"` + Description string `json:"description"` +} + +// Detector tracks events and detects correlations between resources +type Detector struct { + mu sync.RWMutex + events []Event + correlations map[string]*Correlation // key: sourceID:targetID:pattern + + // Configuration + maxEvents int + correlationWindow time.Duration // How long after source event to look for target + minOccurrences int // Minimum co-occurrences to form correlation + retentionWindow time.Duration // How long to keep events + + // Persistence + dataDir string +} + +// Config configures the correlation detector +type Config struct { + MaxEvents int + CorrelationWindow time.Duration // Default: 10 minutes + MinOccurrences int // Default: 3 + RetentionWindow time.Duration // Default: 30 days + DataDir string +} + +// DefaultConfig returns default configuration +func DefaultConfig() Config { + return Config{ + MaxEvents: 10000, + CorrelationWindow: 10 * time.Minute, + MinOccurrences: 3, + RetentionWindow: 30 * 24 * time.Hour, + } +} + +// NewDetector creates a new correlation detector +func NewDetector(cfg Config) *Detector { + if cfg.MaxEvents <= 0 { + cfg.MaxEvents = 10000 + } + if cfg.CorrelationWindow <= 0 { + cfg.CorrelationWindow = 10 * time.Minute + } + if cfg.MinOccurrences <= 0 { + cfg.MinOccurrences = 3 + } + if cfg.RetentionWindow <= 0 { + cfg.RetentionWindow = 30 * 24 * time.Hour + } + + d := &Detector{ + events: make([]Event, 0), + correlations: make(map[string]*Correlation), + maxEvents: cfg.MaxEvents, + correlationWindow: cfg.CorrelationWindow, + minOccurrences: cfg.MinOccurrences, + retentionWindow: cfg.RetentionWindow, + dataDir: cfg.DataDir, + } + + // Load existing data + if cfg.DataDir != "" { + if err := d.loadFromDisk(); err != nil { + log.Warn().Err(err).Msg("Failed to load correlation data from disk") + } else if len(d.events) > 0 { + log.Info().Int("events", len(d.events)).Int("correlations", len(d.correlations)). + Msg("Loaded correlation data from disk") + } + } + + return d +} + +// RecordEvent records a new event for correlation analysis +func (d *Detector) RecordEvent(event Event) { + d.mu.Lock() + defer d.mu.Unlock() + + if event.ID == "" { + event.ID = generateEventID() + } + if event.Timestamp.IsZero() { + event.Timestamp = time.Now() + } + + d.events = append(d.events, event) + d.trimEvents() + + // Check for correlations with recent events on OTHER resources + d.detectCorrelations(event) + + // Persist asynchronously + go func() { + if err := d.saveToDisk(); err != nil { + log.Warn().Err(err).Msg("Failed to save correlation data") + } + }() +} + +// detectCorrelations looks for patterns where this event follows a recent event on another resource +func (d *Detector) detectCorrelations(newEvent Event) { + cutoff := newEvent.Timestamp.Add(-d.correlationWindow) + + for _, oldEvent := range d.events { + // Skip same resource + if oldEvent.ResourceID == newEvent.ResourceID { + continue + } + // Skip events outside the correlation window + if oldEvent.Timestamp.Before(cutoff) || oldEvent.Timestamp.After(newEvent.Timestamp) { + continue + } + + // Found a potential correlation: oldEvent -> newEvent + key := correlationKey(oldEvent.ResourceID, newEvent.ResourceID, oldEvent.EventType, newEvent.EventType) + pattern := string(oldEvent.EventType) + " -> " + string(newEvent.EventType) + delay := newEvent.Timestamp.Sub(oldEvent.Timestamp) + + if existing, ok := d.correlations[key]; ok { + // Update existing correlation + existing.Occurrences++ + existing.AvgDelay = (existing.AvgDelay*time.Duration(existing.Occurrences-1) + delay) / time.Duration(existing.Occurrences) + existing.LastSeen = newEvent.Timestamp + existing.Confidence = d.calculateConfidence(existing.Occurrences) + existing.Description = d.formatCorrelationDescription(existing) + } else { + // Create new correlation + d.correlations[key] = &Correlation{ + SourceID: oldEvent.ResourceID, + SourceName: oldEvent.ResourceName, + SourceType: oldEvent.ResourceType, + TargetID: newEvent.ResourceID, + TargetName: newEvent.ResourceName, + TargetType: newEvent.ResourceType, + EventPattern: pattern, + Occurrences: 1, + AvgDelay: delay, + Confidence: 0.1, // Low initial confidence + LastSeen: newEvent.Timestamp, + } + } + } +} + +// calculateConfidence returns confidence based on occurrence count +func (d *Detector) calculateConfidence(occurrences int) float64 { + // Confidence grows with occurrences, capped at 0.95 + if occurrences < d.minOccurrences { + return float64(occurrences) * 0.1 + } + // Logarithmic growth after threshold + confidence := 0.3 + 0.1*float64(occurrences-d.minOccurrences) + if confidence > 0.95 { + confidence = 0.95 + } + return confidence +} + +// formatCorrelationDescription creates a human-readable description +func (d *Detector) formatCorrelationDescription(c *Correlation) string { + sourceName := c.SourceName + if sourceName == "" { + sourceName = c.SourceID + } + targetName := c.TargetName + if targetName == "" { + targetName = c.TargetID + } + + delayStr := formatDuration(c.AvgDelay) + return "When " + sourceName + " experiences " + string(c.EventPattern[:len(c.EventPattern)/2]) + + ", " + targetName + " often follows within " + delayStr +} + +// GetCorrelations returns all detected correlations above minimum confidence +func (d *Detector) GetCorrelations() []*Correlation { + d.mu.RLock() + defer d.mu.RUnlock() + + var result []*Correlation + for _, c := range d.correlations { + if c.Occurrences >= d.minOccurrences && c.Confidence >= 0.3 { + result = append(result, c) + } + } + + // Sort by confidence descending + sort.Slice(result, func(i, j int) bool { + return result[i].Confidence > result[j].Confidence + }) + + return result +} + +// GetCorrelationsForResource returns correlations involving a specific resource +func (d *Detector) GetCorrelationsForResource(resourceID string) []*Correlation { + d.mu.RLock() + defer d.mu.RUnlock() + + var result []*Correlation + for _, c := range d.correlations { + if (c.SourceID == resourceID || c.TargetID == resourceID) && c.Occurrences >= d.minOccurrences { + result = append(result, c) + } + } + + return result +} + +// GetDependencies returns resources that depend on the given resource +// (resources that experience events after this resource has an event) +func (d *Detector) GetDependencies(resourceID string) []string { + d.mu.RLock() + defer d.mu.RUnlock() + + deps := make(map[string]bool) + for _, c := range d.correlations { + if c.SourceID == resourceID && c.Occurrences >= d.minOccurrences { + deps[c.TargetID] = true + } + } + + result := make([]string, 0, len(deps)) + for dep := range deps { + result = append(result, dep) + } + return result +} + +// GetDependsOn returns resources that the given resource depends on +// (resources whose events precede events on this resource) +func (d *Detector) GetDependsOn(resourceID string) []string { + d.mu.RLock() + defer d.mu.RUnlock() + + deps := make(map[string]bool) + for _, c := range d.correlations { + if c.TargetID == resourceID && c.Occurrences >= d.minOccurrences { + deps[c.SourceID] = true + } + } + + result := make([]string, 0, len(deps)) + for dep := range deps { + result = append(result, dep) + } + return result +} + +// PredictCascade predicts what resources might be affected if the given resource has an issue +func (d *Detector) PredictCascade(resourceID string, eventType EventType) []CascadePrediction { + d.mu.RLock() + defer d.mu.RUnlock() + + var predictions []CascadePrediction + + for _, c := range d.correlations { + if c.SourceID == resourceID && c.Occurrences >= d.minOccurrences { + // Check if the event pattern starts with the given event type + if len(c.EventPattern) > 0 && EventType(c.EventPattern[:len(string(eventType))]) == eventType { + predictions = append(predictions, CascadePrediction{ + ResourceID: c.TargetID, + ResourceName: c.TargetName, + EventPattern: c.EventPattern, + ExpectedIn: c.AvgDelay, + Confidence: c.Confidence, + }) + } + } + } + + // Sort by confidence + sort.Slice(predictions, func(i, j int) bool { + return predictions[i].Confidence > predictions[j].Confidence + }) + + return predictions +} + +// CascadePrediction represents a predicted downstream effect +type CascadePrediction struct { + ResourceID string `json:"resource_id"` + ResourceName string `json:"resource_name"` + EventPattern string `json:"event_pattern"` + ExpectedIn time.Duration `json:"expected_in"` + Confidence float64 `json:"confidence"` +} + +// FormatForContext formats correlations for AI consumption +func (d *Detector) FormatForContext(resourceID string) string { + var correlations []*Correlation + if resourceID != "" { + correlations = d.GetCorrelationsForResource(resourceID) + } else { + correlations = d.GetCorrelations() + } + + if len(correlations) == 0 { + return "" + } + + var result string + result = "\n## 🔗 Resource Correlations\n" + result += "Observed relationships between resources:\n" + + for i, c := range correlations { + if i >= 10 { // Limit to 10 correlations + result += "\n... and more\n" + break + } + if c.Description != "" { + result += "- " + c.Description + "\n" + } else { + result += "- " + c.EventPattern + " (" + formatConfidence(c.Confidence) + " confidence)\n" + } + } + + return result +} + +// trimEvents removes old events +func (d *Detector) trimEvents() { + // Remove events beyond maxEvents + if len(d.events) > d.maxEvents { + d.events = d.events[len(d.events)-d.maxEvents:] + } + + // Remove events older than retention window + cutoff := time.Now().Add(-d.retentionWindow) + kept := make([]Event, 0, len(d.events)) + for _, e := range d.events { + if e.Timestamp.After(cutoff) { + kept = append(kept, e) + } + } + d.events = kept +} + +// saveToDisk persists data +func (d *Detector) saveToDisk() error { + if d.dataDir == "" { + return nil + } + + d.mu.RLock() + data := struct { + Events []Event `json:"events"` + Correlations map[string]*Correlation `json:"correlations"` + }{ + Events: d.events, + Correlations: d.correlations, + } + d.mu.RUnlock() + + jsonData, err := json.MarshalIndent(data, "", " ") + if err != nil { + return err + } + + path := filepath.Join(d.dataDir, "ai_correlations.json") + tmpPath := path + ".tmp" + if err := os.WriteFile(tmpPath, jsonData, 0600); err != nil { + return err + } + + return os.Rename(tmpPath, path) +} + +// loadFromDisk loads data +func (d *Detector) loadFromDisk() error { + if d.dataDir == "" { + return nil + } + + path := filepath.Join(d.dataDir, "ai_correlations.json") + jsonData, err := os.ReadFile(path) + if err != nil { + if os.IsNotExist(err) { + return nil + } + return err + } + + var data struct { + Events []Event `json:"events"` + Correlations map[string]*Correlation `json:"correlations"` + } + + if err := json.Unmarshal(jsonData, &data); err != nil { + return err + } + + d.events = data.Events + d.correlations = data.Correlations + + return nil +} + +// Helper functions + +var eventCounter int64 + +func generateEventID() string { + eventCounter++ + return time.Now().Format("20060102150405") + "-" + intToStr(int(eventCounter%1000)) +} + +func intToStr(n int) string { + if n == 0 { + return "0" + } + var result string + for n > 0 { + result = string(rune('0'+n%10)) + result + n /= 10 + } + return result +} + +func correlationKey(sourceID, targetID string, sourceType, targetType EventType) string { + return sourceID + ":" + targetID + ":" + string(sourceType) + ":" + string(targetType) +} + +func formatDuration(d time.Duration) string { + if d < time.Minute { + return "seconds" + } + if d < time.Hour { + mins := int(d.Minutes()) + if mins == 1 { + return "1 minute" + } + return intToStr(mins) + " minutes" + } + hours := int(d.Hours()) + if hours == 1 { + return "1 hour" + } + return intToStr(hours) + " hours" +} + +func formatConfidence(c float64) string { + pct := int(c * 100) + return intToStr(pct) + "%" +} diff --git a/internal/ai/correlation/detector_test.go b/internal/ai/correlation/detector_test.go new file mode 100644 index 0000000..f99f2ca --- /dev/null +++ b/internal/ai/correlation/detector_test.go @@ -0,0 +1,246 @@ +package correlation + +import ( + "testing" + "time" +) + +func TestDetector_RecordEvent(t *testing.T) { + d := NewDetector(DefaultConfig()) + + d.RecordEvent(Event{ + ResourceID: "vm-100", + ResourceName: "web-server", + EventType: EventHighMem, + Timestamp: time.Now(), + }) + + if len(d.events) != 1 { + t.Errorf("Expected 1 event, got %d", len(d.events)) + } +} + +func TestDetector_CorrelationDetection(t *testing.T) { + d := NewDetector(Config{ + MaxEvents: 1000, + CorrelationWindow: 10 * time.Minute, + MinOccurrences: 2, // Lower threshold for testing + RetentionWindow: 30 * 24 * time.Hour, + }) + + now := time.Now() + + // Simulate pattern: when storage has high usage, database VM restarts + // Occurrence 1 + d.RecordEvent(Event{ + ResourceID: "storage-1", + ResourceName: "local-zfs", + ResourceType: "storage", + EventType: EventDiskFull, + Timestamp: now.Add(-2 * time.Hour), + }) + d.RecordEvent(Event{ + ResourceID: "vm-100", + ResourceName: "database", + ResourceType: "vm", + EventType: EventRestart, + Timestamp: now.Add(-2*time.Hour + 5*time.Minute), + }) + + // Occurrence 2 + d.RecordEvent(Event{ + ResourceID: "storage-1", + ResourceName: "local-zfs", + ResourceType: "storage", + EventType: EventDiskFull, + Timestamp: now.Add(-1 * time.Hour), + }) + d.RecordEvent(Event{ + ResourceID: "vm-100", + ResourceName: "database", + ResourceType: "vm", + EventType: EventRestart, + Timestamp: now.Add(-1*time.Hour + 5*time.Minute), + }) + + // Occurrence 3 + d.RecordEvent(Event{ + ResourceID: "storage-1", + ResourceName: "local-zfs", + ResourceType: "storage", + EventType: EventDiskFull, + Timestamp: now, + }) + d.RecordEvent(Event{ + ResourceID: "vm-100", + ResourceName: "database", + ResourceType: "vm", + EventType: EventRestart, + Timestamp: now.Add(5 * time.Minute), + }) + + // Check correlations + correlations := d.GetCorrelations() + + found := false + for _, c := range correlations { + if c.SourceID == "storage-1" && c.TargetID == "vm-100" { + found = true + if c.Occurrences < 2 { + t.Errorf("Expected at least 2 occurrences, got %d", c.Occurrences) + } + break + } + } + + if !found { + t.Error("Expected correlation between storage-1 and vm-100") + } +} + +func TestDetector_GetCorrelationsForResource(t *testing.T) { + d := NewDetector(Config{ + MaxEvents: 1000, + CorrelationWindow: 10 * time.Minute, + MinOccurrences: 2, + RetentionWindow: 30 * 24 * time.Hour, + }) + + now := time.Now() + + // Create correlation pattern + for i := 0; i < 3; i++ { + offset := time.Duration(-i) * time.Hour + d.RecordEvent(Event{ + ResourceID: "node-1", + EventType: EventHighCPU, + Timestamp: now.Add(offset), + }) + d.RecordEvent(Event{ + ResourceID: "vm-100", + EventType: EventHighMem, + Timestamp: now.Add(offset + 2*time.Minute), + }) + } + + // Get correlations for node-1 + correlations := d.GetCorrelationsForResource("node-1") + if len(correlations) == 0 { + t.Error("Expected correlations for node-1") + } +} + +func TestDetector_GetDependencies(t *testing.T) { + d := NewDetector(Config{ + MaxEvents: 1000, + CorrelationWindow: 10 * time.Minute, + MinOccurrences: 3, + RetentionWindow: 30 * 24 * time.Hour, + }) + + now := time.Now() + + // When storage is full, both VM and container have issues + for i := 0; i < 4; i++ { + offset := time.Duration(-i) * time.Hour + d.RecordEvent(Event{ + ResourceID: "storage-1", + EventType: EventDiskFull, + Timestamp: now.Add(offset), + }) + d.RecordEvent(Event{ + ResourceID: "vm-100", + EventType: EventRestart, + Timestamp: now.Add(offset + 3*time.Minute), + }) + d.RecordEvent(Event{ + ResourceID: "ct-200", + EventType: EventRestart, + Timestamp: now.Add(offset + 5*time.Minute), + }) + } + + deps := d.GetDependencies("storage-1") + if len(deps) < 2 { + t.Errorf("Expected at least 2 dependencies, got %d", len(deps)) + } +} + +func TestDetector_PredictCascade(t *testing.T) { + d := NewDetector(Config{ + MaxEvents: 1000, + CorrelationWindow: 10 * time.Minute, + MinOccurrences: 3, + RetentionWindow: 30 * 24 * time.Hour, + }) + + now := time.Now() + + // Create strong correlation + for i := 0; i < 5; i++ { + offset := time.Duration(-i) * time.Hour + d.RecordEvent(Event{ + ResourceID: "node-1", + ResourceName: "pve-main", + EventType: EventHighMem, + Timestamp: now.Add(offset), + }) + d.RecordEvent(Event{ + ResourceID: "vm-100", + ResourceName: "database", + EventType: EventRestart, + Timestamp: now.Add(offset + 5*time.Minute), + }) + } + + predictions := d.PredictCascade("node-1", EventHighMem) + if len(predictions) == 0 { + t.Error("Expected cascade predictions") + } +} + +func TestDetector_FormatForContext(t *testing.T) { + d := NewDetector(Config{ + MaxEvents: 1000, + CorrelationWindow: 10 * time.Minute, + MinOccurrences: 2, + RetentionWindow: 30 * 24 * time.Hour, + }) + + now := time.Now() + + // Create correlations + for i := 0; i < 3; i++ { + offset := time.Duration(-i) * time.Hour + d.RecordEvent(Event{ + ResourceID: "storage-1", + ResourceName: "local-zfs", + EventType: EventDiskFull, + Timestamp: now.Add(offset), + }) + d.RecordEvent(Event{ + ResourceID: "vm-100", + ResourceName: "database", + EventType: EventRestart, + Timestamp: now.Add(offset + 5*time.Minute), + }) + } + + context := d.FormatForContext("") + if context == "" { + t.Error("Expected non-empty context") + } + + if !contains(context, "Correlation") { + t.Errorf("Expected context to mention correlations: %s", context) + } +} + +func contains(s, substr string) bool { + for i := 0; i <= len(s)-len(substr); i++ { + if s[i:i+len(substr)] == substr { + return true + } + } + return false +} diff --git a/internal/ai/correlation_exports.go b/internal/ai/correlation_exports.go new file mode 100644 index 0000000..96fcdc4 --- /dev/null +++ b/internal/ai/correlation_exports.go @@ -0,0 +1,44 @@ +package ai + +import ( + "github.com/rcourtman/pulse-go-rewrite/internal/ai/correlation" +) + +// CorrelationDetector is an alias for correlation.Detector +type CorrelationDetector = correlation.Detector + +// CorrelationConfig is an alias for correlation.Config +type CorrelationConfig = correlation.Config + +// CorrelationEvent is an alias for correlation.Event +type CorrelationEvent = correlation.Event + +// Correlation is an alias for correlation.Correlation +type Correlation = correlation.Correlation + +// CascadePrediction is an alias for correlation.CascadePrediction +type CascadePrediction = correlation.CascadePrediction + +// CorrelationEventType is an alias for correlation.EventType +type CorrelationEventType = correlation.EventType + +// Event type constants +const ( + CorrelationEventAlert = correlation.EventAlert + CorrelationEventRestart = correlation.EventRestart + CorrelationEventHighCPU = correlation.EventHighCPU + CorrelationEventHighMem = correlation.EventHighMem + CorrelationEventDiskFull = correlation.EventDiskFull + CorrelationEventOffline = correlation.EventOffline + CorrelationEventMigration = correlation.EventMigration +) + +// NewCorrelationDetector creates a new correlation detector +func NewCorrelationDetector(cfg CorrelationConfig) *CorrelationDetector { + return correlation.NewDetector(cfg) +} + +// DefaultCorrelationConfig returns default correlation detector configuration +func DefaultCorrelationConfig() CorrelationConfig { + return correlation.DefaultConfig() +} diff --git a/internal/ai/patrol.go b/internal/ai/patrol.go index 751f514..31a3ae5 100644 --- a/internal/ai/patrol.go +++ b/internal/ai/patrol.go @@ -204,17 +204,18 @@ const MaxPatrolRunHistory = 100 type PatrolService struct { mu sync.RWMutex - aiService *Service - stateProvider StateProvider - thresholdProvider ThresholdProvider - config PatrolConfig - findings *FindingsStore - knowledgeStore *knowledge.Store // For per-resource notes in patrol context - metricsHistory MetricsHistoryProvider // For trend analysis and predictions - baselineStore *baseline.Store // For anomaly detection via learned baselines - changeDetector *ChangeDetector // For tracking infrastructure changes - remediationLog *RemediationLog // For tracking remediation actions - patternDetector *PatternDetector // For failure prediction from historical patterns + aiService *Service + stateProvider StateProvider + thresholdProvider ThresholdProvider + config PatrolConfig + findings *FindingsStore + knowledgeStore *knowledge.Store // For per-resource notes in patrol context + metricsHistory MetricsHistoryProvider // For trend analysis and predictions + baselineStore *baseline.Store // For anomaly detection via learned baselines + changeDetector *ChangeDetector // For tracking infrastructure changes + remediationLog *RemediationLog // For tracking remediation actions + patternDetector *PatternDetector // For failure prediction from historical patterns + correlationDetector *CorrelationDetector // For multi-resource correlation // Cached thresholds (recalculated when thresholdProvider changes) thresholds PatrolThresholds @@ -399,6 +400,21 @@ func (p *PatrolService) GetPatternDetector() *PatternDetector { return p.patternDetector } +// SetCorrelationDetector sets the correlation detector for multi-resource correlation +func (p *PatrolService) SetCorrelationDetector(detector *CorrelationDetector) { + p.mu.Lock() + defer p.mu.Unlock() + p.correlationDetector = detector + log.Info().Msg("AI Patrol: Correlation detector set for multi-resource analysis") +} + +// GetCorrelationDetector returns the correlation detector +func (p *PatrolService) GetCorrelationDetector() *CorrelationDetector { + p.mu.RLock() + defer p.mu.RUnlock() + return p.correlationDetector +} + // GetConfig returns the current patrol configuration func (p *PatrolService) GetConfig() PatrolConfig { p.mu.RLock() @@ -1802,6 +1818,7 @@ func (p *PatrolService) buildEnrichedContext(state models.StateSnapshot) string // Append failure predictions if pattern detector is available p.mu.RLock() patternDetector := p.patternDetector + correlationDetector := p.correlationDetector p.mu.RUnlock() if patternDetector != nil { @@ -1810,6 +1827,14 @@ func (p *PatrolService) buildEnrichedContext(state models.StateSnapshot) string formatted += predictionsContext } } + + // Append resource correlations if correlation detector is available + if correlationDetector != nil { + correlationsContext := correlationDetector.FormatForContext("") + if correlationsContext != "" { + formatted += correlationsContext + } + } log.Debug(). Int("resources", infraCtx.TotalResources). diff --git a/internal/ai/service.go b/internal/ai/service.go index 1c164b0..7a9388c 100644 --- a/internal/ai/service.go +++ b/internal/ai/service.go @@ -252,6 +252,17 @@ func (s *Service) SetPatternDetector(detector *PatternDetector) { } } +// SetCorrelationDetector sets the correlation detector for multi-resource correlation +func (s *Service) SetCorrelationDetector(detector *CorrelationDetector) { + s.mu.RLock() + patrol := s.patrolService + s.mu.RUnlock() + + if patrol != nil { + patrol.SetCorrelationDetector(detector) + } +} + // StartPatrol starts the background patrol service func (s *Service) StartPatrol(ctx context.Context) { s.mu.RLock() diff --git a/internal/api/ai_handlers.go b/internal/api/ai_handlers.go index f54c2cc..febeae4 100644 --- a/internal/api/ai_handlers.go +++ b/internal/api/ai_handlers.go @@ -120,6 +120,11 @@ func (h *AISettingsHandler) SetPatternDetector(detector *ai.PatternDetector) { h.aiService.SetPatternDetector(detector) } +// SetCorrelationDetector sets the correlation detector for multi-resource correlation +func (h *AISettingsHandler) SetCorrelationDetector(detector *ai.CorrelationDetector) { + h.aiService.SetCorrelationDetector(detector) +} + // StopPatrol stops the background AI patrol service func (h *AISettingsHandler) StopPatrol() { h.aiService.StopPatrol() diff --git a/internal/api/router.go b/internal/api/router.go index d39c79c..c7df749 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -1480,6 +1480,45 @@ func (r *Router) StartPatrol(ctx context.Context) { log.Info().Msg("AI Pattern Detector: Wired to alert history for failure prediction") } } + + // Initialize correlation detector for multi-resource relationships + correlationDetector := ai.NewCorrelationDetector(ai.CorrelationConfig{ + MaxEvents: 10000, + CorrelationWindow: 10 * time.Minute, + MinOccurrences: 3, + RetentionWindow: 30 * 24 * time.Hour, + DataDir: dataDir, + }) + if correlationDetector != nil { + r.aiSettingsHandler.SetCorrelationDetector(correlationDetector) + + // Wire alert history to correlation detector + if alertManager := r.monitor.GetAlertManager(); alertManager != nil { + alertManager.OnAlertHistory(func(alert alerts.Alert) { + // Record as correlation event + eventType := ai.CorrelationEventType(ai.CorrelationEventAlert) + switch alert.Type { + case "cpu": + eventType = ai.CorrelationEventHighCPU + case "memory": + eventType = ai.CorrelationEventHighMem + case "disk": + eventType = ai.CorrelationEventDiskFull + case "offline", "connectivity": + eventType = ai.CorrelationEventOffline + } + correlationDetector.RecordEvent(ai.CorrelationEvent{ + ResourceID: alert.ResourceID, + ResourceName: alert.ResourceName, + ResourceType: alert.Type, + EventType: eventType, + Timestamp: alert.StartTime, + Value: alert.Value, + }) + }) + log.Info().Msg("AI Correlation Detector: Wired to alert history for multi-resource analysis") + } + } r.aiSettingsHandler.StartPatrol(ctx) }