feat(ai): Wire baseline learning loop into router startup
Complete Phase 2 baseline integration: - Add baseline_exports.go for clean type aliasing - Wire baseline store initialization into StartPatrol - Implement startBaselineLearning background loop - Runs initial learning after 5 min delay - Updates baselines every hour from metrics history - Learns from 7 days of data for nodes, VMs, containers - Add SetBaselineStore methods throughout the chain (Router -> AIHandler -> Service -> PatrolService) - Persists baselines to data directory as JSON The baseline learning loop: 1. Starts automatically when AI patrol starts 2. Queries metrics history for all resources 3. Computes mean, stddev, percentiles for cpu/memory/disk 4. Saves baselines to disk for durability 5. Anomaly detection uses these baselines in context builder All tests passing.
This commit is contained in:
parent
f3e95c24ae
commit
47eefe6763
4 changed files with 165 additions and 0 deletions
24
internal/ai/baseline_exports.go
Normal file
24
internal/ai/baseline_exports.go
Normal file
|
|
@ -0,0 +1,24 @@
|
||||||
|
package ai
|
||||||
|
|
||||||
|
import (
|
||||||
|
"github.com/rcourtman/pulse-go-rewrite/internal/ai/baseline"
|
||||||
|
)
|
||||||
|
|
||||||
|
// BaselineConfig is an alias for the baseline package config
|
||||||
|
type BaselineConfig = baseline.StoreConfig
|
||||||
|
|
||||||
|
// BaselineStore is an alias for the baseline.Store type
|
||||||
|
type BaselineStore = baseline.Store
|
||||||
|
|
||||||
|
// BaselineMetricPoint is an alias for the baseline.MetricPoint type
|
||||||
|
type BaselineMetricPoint = baseline.MetricPoint
|
||||||
|
|
||||||
|
// DefaultBaselineConfig returns the default baseline configuration
|
||||||
|
func DefaultBaselineConfig() BaselineConfig {
|
||||||
|
return baseline.DefaultConfig()
|
||||||
|
}
|
||||||
|
|
||||||
|
// NewBaselineStore creates a new baseline store
|
||||||
|
func NewBaselineStore(cfg BaselineConfig) *BaselineStore {
|
||||||
|
return baseline.NewStore(cfg)
|
||||||
|
}
|
||||||
|
|
@ -175,6 +175,17 @@ func (s *Service) SetMetricsHistoryProvider(provider MetricsHistoryProvider) {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// SetBaselineStore sets the baseline store for anomaly detection
|
||||||
|
func (s *Service) SetBaselineStore(store *BaselineStore) {
|
||||||
|
s.mu.RLock()
|
||||||
|
patrol := s.patrolService
|
||||||
|
s.mu.RUnlock()
|
||||||
|
|
||||||
|
if patrol != nil {
|
||||||
|
patrol.SetBaselineStore(store)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// StartPatrol starts the background patrol service
|
// StartPatrol starts the background patrol service
|
||||||
func (s *Service) StartPatrol(ctx context.Context) {
|
func (s *Service) StartPatrol(ctx context.Context) {
|
||||||
s.mu.RLock()
|
s.mu.RLock()
|
||||||
|
|
|
||||||
|
|
@ -96,6 +96,11 @@ func (h *AISettingsHandler) SetMetricsHistoryProvider(provider ai.MetricsHistory
|
||||||
h.aiService.SetMetricsHistoryProvider(provider)
|
h.aiService.SetMetricsHistoryProvider(provider)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// SetBaselineStore sets the baseline store for anomaly detection
|
||||||
|
func (h *AISettingsHandler) SetBaselineStore(store *ai.BaselineStore) {
|
||||||
|
h.aiService.SetBaselineStore(store)
|
||||||
|
}
|
||||||
|
|
||||||
// StopPatrol stops the background AI patrol service
|
// StopPatrol stops the background AI patrol service
|
||||||
func (h *AISettingsHandler) StopPatrol() {
|
func (h *AISettingsHandler) StopPatrol() {
|
||||||
h.aiService.StopPatrol()
|
h.aiService.StopPatrol()
|
||||||
|
|
|
||||||
|
|
@ -1418,6 +1418,20 @@ func (r *Router) StartPatrol(ctx context.Context) {
|
||||||
if adapter != nil {
|
if adapter != nil {
|
||||||
r.aiSettingsHandler.SetMetricsHistoryProvider(adapter)
|
r.aiSettingsHandler.SetMetricsHistoryProvider(adapter)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Initialize baseline store for anomaly detection
|
||||||
|
// Uses config dir for persistence
|
||||||
|
baselineCfg := ai.DefaultBaselineConfig()
|
||||||
|
if r.persistence != nil {
|
||||||
|
baselineCfg.DataDir = r.persistence.DataDir()
|
||||||
|
}
|
||||||
|
baselineStore := ai.NewBaselineStore(baselineCfg)
|
||||||
|
if baselineStore != nil {
|
||||||
|
r.aiSettingsHandler.SetBaselineStore(baselineStore)
|
||||||
|
|
||||||
|
// Start background baseline learning loop
|
||||||
|
go r.startBaselineLearning(ctx, baselineStore, metricsHistory)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -1432,6 +1446,117 @@ func (r *Router) StopPatrol() {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// startBaselineLearning runs a background loop that learns baselines from metrics history
|
||||||
|
// This enables anomaly detection by understanding what "normal" looks like for each resource
|
||||||
|
func (r *Router) startBaselineLearning(ctx context.Context, store *ai.BaselineStore, metricsHistory *monitoring.MetricsHistory) {
|
||||||
|
if store == nil || metricsHistory == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
// Learn every hour
|
||||||
|
ticker := time.NewTicker(1 * time.Hour)
|
||||||
|
defer ticker.Stop()
|
||||||
|
|
||||||
|
// Run initial learning after a short delay (allow metrics to accumulate)
|
||||||
|
select {
|
||||||
|
case <-ctx.Done():
|
||||||
|
return
|
||||||
|
case <-time.After(5 * time.Minute):
|
||||||
|
r.learnBaselines(store, metricsHistory)
|
||||||
|
}
|
||||||
|
|
||||||
|
log.Info().Msg("Baseline learning loop started")
|
||||||
|
|
||||||
|
for {
|
||||||
|
select {
|
||||||
|
case <-ctx.Done():
|
||||||
|
// Save baselines before exit
|
||||||
|
if err := store.Save(); err != nil {
|
||||||
|
log.Warn().Err(err).Msg("Failed to save baselines on shutdown")
|
||||||
|
}
|
||||||
|
log.Info().Msg("Baseline learning loop stopped")
|
||||||
|
return
|
||||||
|
case <-ticker.C:
|
||||||
|
r.learnBaselines(store, metricsHistory)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// learnBaselines updates baselines for all resources from metrics history
|
||||||
|
func (r *Router) learnBaselines(store *ai.BaselineStore, metricsHistory *monitoring.MetricsHistory) {
|
||||||
|
if r.monitor == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
state := r.monitor.GetState()
|
||||||
|
learningWindow := 7 * 24 * time.Hour // Learn from 7 days of data
|
||||||
|
var learned int
|
||||||
|
|
||||||
|
// Learn baselines for nodes
|
||||||
|
for _, node := range state.Nodes {
|
||||||
|
for _, metric := range []string{"cpu", "memory"} {
|
||||||
|
points := metricsHistory.GetNodeMetrics(node.ID, metric, learningWindow)
|
||||||
|
if len(points) > 0 {
|
||||||
|
baselinePoints := make([]ai.BaselineMetricPoint, len(points))
|
||||||
|
for i, p := range points {
|
||||||
|
baselinePoints[i] = ai.BaselineMetricPoint{Value: p.Value, Timestamp: p.Timestamp}
|
||||||
|
}
|
||||||
|
if err := store.Learn(node.ID, "node", metric, baselinePoints); err == nil {
|
||||||
|
learned++
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Learn baselines for VMs
|
||||||
|
for _, vm := range state.VMs {
|
||||||
|
if vm.Template {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
for _, metric := range []string{"cpu", "memory", "disk"} {
|
||||||
|
points := metricsHistory.GetGuestMetrics(vm.ID, metric, learningWindow)
|
||||||
|
if len(points) > 0 {
|
||||||
|
baselinePoints := make([]ai.BaselineMetricPoint, len(points))
|
||||||
|
for i, p := range points {
|
||||||
|
baselinePoints[i] = ai.BaselineMetricPoint{Value: p.Value, Timestamp: p.Timestamp}
|
||||||
|
}
|
||||||
|
if err := store.Learn(vm.ID, "vm", metric, baselinePoints); err == nil {
|
||||||
|
learned++
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Learn baselines for containers
|
||||||
|
for _, ct := range state.Containers {
|
||||||
|
if ct.Template {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
for _, metric := range []string{"cpu", "memory", "disk"} {
|
||||||
|
points := metricsHistory.GetGuestMetrics(ct.ID, metric, learningWindow)
|
||||||
|
if len(points) > 0 {
|
||||||
|
baselinePoints := make([]ai.BaselineMetricPoint, len(points))
|
||||||
|
for i, p := range points {
|
||||||
|
baselinePoints[i] = ai.BaselineMetricPoint{Value: p.Value, Timestamp: p.Timestamp}
|
||||||
|
}
|
||||||
|
if err := store.Learn(ct.ID, "container", metric, baselinePoints); err == nil {
|
||||||
|
learned++
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Save after learning
|
||||||
|
if err := store.Save(); err != nil {
|
||||||
|
log.Warn().Err(err).Msg("Failed to save baselines")
|
||||||
|
}
|
||||||
|
|
||||||
|
log.Debug().
|
||||||
|
Int("baselines_updated", learned).
|
||||||
|
Int("resources", store.ResourceCount()).
|
||||||
|
Msg("Baseline learning complete")
|
||||||
|
}
|
||||||
|
|
||||||
// GetAlertTriggeredAnalyzer returns the alert-triggered analyzer for wiring into the monitor's alert callback
|
// GetAlertTriggeredAnalyzer returns the alert-triggered analyzer for wiring into the monitor's alert callback
|
||||||
// This enables AI to analyze specific resources when alerts fire, providing token-efficient real-time insights
|
// This enables AI to analyze specific resources when alerts fire, providing token-efficient real-time insights
|
||||||
func (r *Router) GetAlertTriggeredAnalyzer() *ai.AlertTriggeredAnalyzer {
|
func (r *Router) GetAlertTriggeredAnalyzer() *ai.AlertTriggeredAnalyzer {
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue