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.
This commit is contained in:
rcourtman
parent
8b3bfb60d2
commit
f6d31d166a
7 changed files with 882 additions and 11 deletions
501
internal/ai/correlation/detector.go
Normal file
501
internal/ai/correlation/detector.go
Normal file
|
|
@ -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) + "%"
|
||||
}
|
||||
246
internal/ai/correlation/detector_test.go
Normal file
246
internal/ai/correlation/detector_test.go
Normal file
|
|
@ -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
|
||||
}
|
||||
44
internal/ai/correlation_exports.go
Normal file
44
internal/ai/correlation_exports.go
Normal file
|
|
@ -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()
|
||||
}
|
||||
|
|
@ -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).
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue