[AAD-6] Evict inactive observer series

comp/anomalydetection/AGENTS.md

@@ -214,6 +214,8 @@ Keys are declared in the config schema (`pkg/config/schema/yaml/`).
 | `anomaly_detection.storage.max_series` | `50000` | Storage series cap |
 | `anomaly_detection.storage.eviction_floor_ratio` | `0.5` | Fraction below the cap to drain during series eviction |
 | `anomaly_detection.storage.point_retention` | `120s` | Per-series point retention |
+| `anomaly_detection.storage.inactive_series_ttl` | `20m` | Evict non-telemetry series inactive for this long; `0` disables inactivity eviction |
+| `anomaly_detection.storage.inactive_series_check_interval` | `5m` | Advance-time interval between inactivity scans; `0` disables inactivity eviction |
 
 Per-source log rate limits and min severity live under
 `anomaly_detection.logs.{internal,kubelet,containers}.*`.

comp/anomalydetection/observer/impl/component_test.go

@@ -144,3 +144,60 @@ anomaly_detection:
 
 	requireNoObserverMetricFamilies(t, telComp)
 }
+
+func TestNewComponentReadsInactiveSeriesEvictionStorageConfig(t *testing.T) {
+	tt := []struct {
+		name          string
+		storageConfig string
+		wantTTL       int64
+		wantInterval  int64
+	}{
+		{
+			name: "configured",
+			storageConfig: `
+    inactive_series_ttl: 30m
+    inactive_series_check_interval: 10m`,
+			wantTTL:      30 * 60,
+			wantInterval: 10 * 60,
+		},
+		{
+			name: "disabled with zero",
+			storageConfig: `
+    inactive_series_ttl: 0s
+    inactive_series_check_interval: 0s`,
+			wantTTL:      0,
+			wantInterval: 0,
+		},
+		{
+			name: "negative values retain defaults",
+			storageConfig: `
+    inactive_series_ttl: -1s
+    inactive_series_check_interval: -1s`,
+			wantTTL:      storageInactiveSeriesTTLSeconds,
+			wantInterval: storageInactiveSeriesCheckIntervalSeconds,
+		},
+	}
+
+	for _, tc := range tt {
+		t.Run(tc.name, func(t *testing.T) {
+			cfg := configmock.NewFromYAML(t, `
+anomaly_detection:
+  reporting:
+    events:
+      enabled: true
+  storage:
+`+tc.storageConfig)
+
+			provides, err := NewComponent(Requires{
+				Lifecycle: &testLifecycle{},
+				Config:    cfg,
+				Log:       noopLogComponent{},
+			})
+			require.NoError(t, err)
+			obs, ok := provides.Comp.(*observerImpl)
+			require.True(t, ok)
+			require.Equal(t, tc.wantTTL, obs.engine.storage.cfg.InactiveSeriesTTLSeconds)
+			require.Equal(t, tc.wantInterval, obs.engine.storage.cfg.InactiveSeriesCheckIntervalSeconds)
+		})
+	}
+}

comp/anomalydetection/observer/impl/engine.go

@@ -63,6 +63,11 @@ type engine struct {
 	// latestDataTime is the latest data timestamp seen across all ingested observations.
 	latestDataTime int64
 
+	// inactiveSeriesEvictionChecked tracks the advance timestamp of the last
+	// inactivity scan. It is engine-goroutine owned, like storage mutation.
+	inactiveSeriesEvictionChecked   bool
+	lastInactiveSeriesEvictionCheck int64
+
 	// Raw anomaly tracking (for telemetry and testbench display).
 	rawAnomalies         []observerdef.Anomaly
 	rawAnomalyIndex      map[anomalyDedupKey]int // O(1) dedup lookup
@@ -503,6 +508,10 @@ func (e *engine) advanceWithReason(upToSec int64, reason advanceReason) advanceR
 	if e.logCounts != nil {
 		e.logCounts.flush(e.storage, upToSec)
 	}
+	// Inactivity eviction happens after materialized log-count buckets have
+	// restored their real last-observation activity time, and before detectors
+	// can recreate state for series that are no longer relevant.
+	e.evictInactiveSeries(upToSec)
 
 	result := e.runDetectorsAndCorrelatorsSnapshot(upToSec, detectors, correlators)
 
@@ -535,6 +544,30 @@ func (e *engine) advanceWithReason(upToSec int64, reason advanceReason) advanceR
 	return result
 }
 
+func (e *engine) evictInactiveSeries(upToSec int64) {
+	cfg := e.storage.cfg
+	if cfg.InactiveSeriesTTLSeconds <= 0 || cfg.InactiveSeriesCheckIntervalSeconds <= 0 {
+		return
+	}
+	if e.inactiveSeriesEvictionChecked && upToSec-e.lastInactiveSeriesEvictionCheck < cfg.InactiveSeriesCheckIntervalSeconds {
+		return
+	}
+	e.inactiveSeriesEvictionChecked = true
+	e.lastInactiveSeriesEvictionCheck = upToSec
+
+	freed := e.storage.EvictInactiveBefore(upToSec - cfg.InactiveSeriesTTLSeconds)
+	if len(freed) == 0 {
+		return
+	}
+	if e.logCounts != nil {
+		e.logCounts.removeSeriesByRefs(freed)
+	}
+	if e.onStorageSeriesEvicted != nil {
+		e.onStorageSeriesEvicted("inactive", len(freed))
+	}
+	e.fanOutSeriesRemoval(freed)
+}
+
 // runDetectorsAndCorrelatorsSnapshot runs the given detectors and correlators.
 // Uses explicit slices so the caller can snapshot them under a lock.
 //
@@ -939,6 +972,8 @@ func (e *engine) Reset() {
 
 	e.lastAnalyzedDataTime = 0
 	e.latestDataTime = 0
+	e.inactiveSeriesEvictionChecked = false
+	e.lastInactiveSeriesEvictionCheck = 0
 
 	for _, detector := range e.detectors {
 		if resetter, ok := detector.(interface{ Reset() }); ok {
@@ -1000,6 +1035,8 @@ func (e *engine) resetAnalysisState() {
 	e.mu.Lock()
 	e.lastAnalyzedDataTime = 0
 	e.latestDataTime = 0
+	e.inactiveSeriesEvictionChecked = false
+	e.lastInactiveSeriesEvictionCheck = 0
 	e.mu.Unlock()
 
 	for _, detector := range e.detectors {

comp/anomalydetection/observer/impl/engine_stepadvance_test.go

@@ -18,6 +18,23 @@ type fixedDetector struct {
 	fired     bool
 }
 
+type inactiveEvictionDetector struct {
+	seenSeries int
+	removed    []observer.SeriesRef
+}
+
+func (*inactiveEvictionDetector) Name() string { return "inactive_eviction" }
+func (*inactiveEvictionDetector) Ready() bool  { return true }
+
+func (d *inactiveEvictionDetector) Detect(storage observer.StorageReader, _ int64) observer.DetectionResult {
+	d.seenSeries = len(storage.ListSeries(observer.WorkloadSeriesFilter()))
+	return observer.DetectionResult{}
+}
+
+func (d *inactiveEvictionDetector) RemoveSeries(refs []observer.SeriesRef) {
+	d.removed = append(d.removed, refs...)
+}
+
 func (d *fixedDetector) Name() string { return "fixed" }
 func (*fixedDetector) Ready() bool    { return true }
 
@@ -165,3 +182,58 @@ func TestStepAdvance_SingleGroupWithinWindow(t *testing.T) {
 	t.Logf("Accumulated: %d correlations", len(accumulated))
 	assert.NotEmpty(t, accumulated, "cluster within window should always be accumulated")
 }
+
+func TestEngine_EvictsInactiveSeriesBeforeDetectionAtConfiguredCadence(t *testing.T) {
+	storageCfg := DefaultStorageConfig()
+	storageCfg.PointRetentionSecs = 0
+	storageCfg.MaxSeries = 0
+	storageCfg.InactiveSeriesTTLSeconds = 1_200
+	storageCfg.InactiveSeriesCheckIntervalSeconds = 300
+	storage := newTimeSeriesStorageWith(storageCfg)
+	old := storage.Add("ns", "old", 1, 100, nil).Ref
+	active := storage.Add("ns", "active", 1, 101, nil).Ref
+	detector := &inactiveEvictionDetector{}
+	e := newEngine(engineConfig{storage: storage, detectors: []observer.Detector{detector}})
+
+	var evictionReasons []string
+	e.onStorageSeriesEvicted = func(reason string, _ int) {
+		evictionReasons = append(evictionReasons, reason)
+	}
+
+	// The first advance always scans. At this exact cutoff old is stale while
+	// active remains; the detector must only see active.
+	e.Advance(1_300)
+	assert.Nil(t, storage.GetSeriesMeta(old))
+	assert.NotNil(t, storage.GetSeriesMeta(active))
+	assert.Equal(t, 1, detector.seenSeries)
+	assert.Equal(t, []observer.SeriesRef{old}, detector.removed)
+	assert.Equal(t, []string{"inactive"}, evictionReasons)
+
+	// Add an already-stale series after the first scan. It is retained until
+	// the full 5-minute advance interval elapses.
+	pending := storage.Add("ns", "pending", 1, 200, nil).Ref
+	e.Advance(1_599)
+	assert.NotNil(t, storage.GetSeriesMeta(pending))
+	assert.Len(t, detector.removed, 1)
+
+	// At the exact interval boundary the next scan removes both remaining
+	// stale series and notifies the detector.
+	e.Advance(1_600)
+	assert.Nil(t, storage.GetSeriesMeta(active))
+	assert.Nil(t, storage.GetSeriesMeta(pending))
+	assert.ElementsMatch(t, []observer.SeriesRef{old, active, pending}, detector.removed)
+	assert.Equal(t, []string{"inactive", "inactive"}, evictionReasons)
+}
+
+func TestEngine_InactiveSeriesEvictionDisabled(t *testing.T) {
+	storageCfg := DefaultStorageConfig()
+	storageCfg.PointRetentionSecs = 0
+	storageCfg.InactiveSeriesTTLSeconds = 0
+	storage := newTimeSeriesStorageWith(storageCfg)
+	ref := storage.Add("ns", "old", 1, 0, nil).Ref
+	e := newEngine(engineConfig{storage: storage})
+
+	e.Advance(10_000)
+
+	assert.NotNil(t, storage.GetSeriesMeta(ref))
+}

comp/anomalydetection/observer/impl/observer.go

@@ -274,6 +274,22 @@ func NewComponent(deps Requires) (Provides, error) {
 				storageCfg.PointRetentionSecs = int64(d.Seconds())
 			}
 		}
+		if cfg.IsConfigured("anomaly_detection.storage.inactive_series_ttl") {
+			d := cfg.GetDuration("anomaly_detection.storage.inactive_series_ttl")
+			if d < 0 {
+				pkglog.Warnf("anomaly_detection.storage.inactive_series_ttl must be >= 0, got %s — using default", d)
+			} else {
+				storageCfg.InactiveSeriesTTLSeconds = int64(d.Seconds())
+			}
+		}
+		if cfg.IsConfigured("anomaly_detection.storage.inactive_series_check_interval") {
+			d := cfg.GetDuration("anomaly_detection.storage.inactive_series_check_interval")
+			if d < 0 {
+				pkglog.Warnf("anomaly_detection.storage.inactive_series_check_interval must be >= 0, got %s — using default", d)
+			} else {
+				storageCfg.InactiveSeriesCheckIntervalSeconds = int64(d.Seconds())
+			}
+		}
 	}
 
 	compiledMetricFilter, err := loadMetricFilter(cfg)

comp/anomalydetection/observer/impl/storage.go

@@ -33,6 +33,14 @@ type StorageConfig struct {
 	// on each Add. 0 disables trimming.
 	PointRetentionSecs int64
 
+	// InactiveSeriesTTLSeconds is how long a non-telemetry series may remain
+	// inactive before an engine advance evicts it. 0 disables inactivity eviction.
+	InactiveSeriesTTLSeconds int64
+
+	// InactiveSeriesCheckIntervalSeconds is the minimum advance-time interval
+	// between inactivity scans. 0 disables inactivity eviction.
+	InactiveSeriesCheckIntervalSeconds int64
+
 	// MaxCorrelations caps how many unique correlation patterns are retained in
 	// the engine's accumulated-correlations map. 0 uses the built-in default
 	// (500). -1 disables the cap entirely (suitable for testbench replay where
@@ -52,9 +60,11 @@ type StorageConfig struct {
 // DefaultStorageConfig returns the hard-coded production defaults.
 func DefaultStorageConfig() StorageConfig {
 	return StorageConfig{
-		MaxSeries:          storageMaxSeries,
-		EvictionFloorRatio: storageEvictionBandRatio,
-		PointRetentionSecs: storagePointRetentionSecs,
+		MaxSeries:                          storageMaxSeries,
+		EvictionFloorRatio:                 storageEvictionBandRatio,
+		PointRetentionSecs:                 storagePointRetentionSecs,
+		InactiveSeriesTTLSeconds:           storageInactiveSeriesTTLSeconds,
+		InactiveSeriesCheckIntervalSeconds: storageInactiveSeriesCheckIntervalSeconds,
 		// TrackCorrelationHistory defaults to false: live agent incurs no overhead.
 	}
 }
@@ -70,6 +80,14 @@ const (
 	// storagePointRetentionSecs is the default point retention window.
 	// Points older than (latest_ts - 120s) are trimmed on each Add.
 	storagePointRetentionSecs = 120
+
+	// storageInactiveSeriesTTLSeconds is the default inactivity lifetime for
+	// non-telemetry series. Inactivity is evaluated against advance timestamps.
+	storageInactiveSeriesTTLSeconds = 20 * 60
+
+	// storageInactiveSeriesCheckIntervalSeconds bounds the work done by
+	// inactivity scans while keeping eviction deterministic under replay.
+	storageInactiveSeriesCheckIntervalSeconds = 5 * 60
 )
 
 // timeSeriesStorage is an internal storage for time series data.
@@ -1244,6 +1262,28 @@ func (s *timeSeriesStorage) EvictToCapacity(seriesLimit, target int) []observer.
 	return freed
 }
 
+// EvictInactiveBefore removes non-telemetry series whose last activity is at
+// or before cutoff. The caller supplies a data-time cutoff so eviction is
+// deterministic in both live operation and replay.
+func (s *timeSeriesStorage) EvictInactiveBefore(cutoff int64) []observer.SeriesRef {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	var freed []observer.SeriesRef
+	for _, stats := range s.seriesIDStats {
+		if stats == nil || stats.Namespace == observer.TelemetryNamespace || stats.lastActivityTimestamp > cutoff {
+			continue
+		}
+		if s.removeSeries(stats) {
+			freed = append(freed, stats.ref)
+		}
+	}
+	if len(freed) > 0 {
+		s.seriesGen++
+	}
+	return freed
+}
+
 // EvictDefault evicts to capacity using the storage's own config.
 // The eviction target is MaxSeries*(1-EvictionFloorRatio).
 func (s *timeSeriesStorage) EvictDefault() []observer.SeriesRef {

comp/anomalydetection/observer/impl/storage_test.go

@@ -33,6 +33,12 @@ func TestTimeSeriesStorage_Add(t *testing.T) {
 	assert.Equal(t, 10.0, series.Points[0].Value)
 }
 
+func TestDefaultStorageConfigIncludesInactiveSeriesEviction(t *testing.T) {
+	cfg := DefaultStorageConfig()
+	assert.Equal(t, int64(20*60), cfg.InactiveSeriesTTLSeconds)
+	assert.Equal(t, int64(5*60), cfg.InactiveSeriesCheckIntervalSeconds)
+}
+
 func TestTimeSeriesStorage_AddSameBucket_Average(t *testing.T) {
 	s := newTimeSeriesStorage()
 
@@ -684,6 +690,25 @@ func TestTimeSeriesStorage_TotalSeriesCountTracksCapacityEviction(t *testing.T)
 	require.Equal(t, 1, s.TotalSeriesCount())
 }
 
+func TestTimeSeriesStorage_EvictInactiveBefore(t *testing.T) {
+	s := newTimeSeriesStorage()
+	old := s.Add("workload", "old", 1, 100, []string{"env:test"}).Ref
+	exact := s.Add("workload", "exact", 1, 300, nil).Ref
+	newer := s.Add("workload", "newer", 1, 301, nil).Ref
+	telemetry := s.Add(observer.TelemetryNamespace, "internal", 1, 100, nil).Ref
+	genBefore := s.SeriesGeneration()
+
+	freed := s.EvictInactiveBefore(300)
+
+	require.ElementsMatch(t, []observer.SeriesRef{old, exact}, freed)
+	assert.Nil(t, s.GetSeriesMeta(old))
+	assert.Nil(t, s.GetSeriesMeta(exact))
+	assert.NotNil(t, s.GetSeriesMeta(newer))
+	assert.NotNil(t, s.GetSeriesMeta(telemetry), "telemetry series must not be evicted by inactivity")
+	assert.Equal(t, 1, s.TotalSeriesCount(), "only the newer non-telemetry series remains live")
+	assert.Greater(t, s.SeriesGeneration(), genBefore)
+}
+
 func TestTimeSeriesStorage_FindRefsByHashes(t *testing.T) {
 	s := newTimeSeriesStorage()
 

internal/qbranch/anomalydetection-testbench/bench/bench.go

@@ -666,21 +666,25 @@ func (m *parquetMetricView) GetTimestampUnix() int64 { return m.timestamp }
 func (m *parquetMetricView) GetSampleRate() float64  { return 1.0 }
 
 // unboundedStorageCfg returns a StorageConfig for testbench replay:
-// no point-retention window (pre-loaded data stays in memory) and full
+// no point-retention or inactivity-eviction window (pre-loaded data stays in memory) and full
 // correlation history accumulation enabled (disabled in live mode to avoid
 // per-Advance overhead that production reporters never read).
 func unboundedStorageCfg() observerimpl.StorageConfig {
 	cfg := observerimpl.DefaultStorageConfig()
 	cfg.PointRetentionSecs = 0
+	cfg.InactiveSeriesTTLSeconds = 0
+	cfg.InactiveSeriesCheckIntervalSeconds = 0
 	cfg.MaxCorrelations = -1           // unlimited — testbench must show all patterns
 	cfg.TrackCorrelationHistory = true // accumulate history for replay UI / output
 	return cfg
 }
 
 // streamingStorageCfg uses the production point-retention and series limits,
-// while retaining complete correlation history required by headless output.
+// but disables inactivity eviction so headless output retains complete scenario state.
 func streamingStorageCfg() observerimpl.StorageConfig {
 	cfg := observerimpl.DefaultStorageConfig()
+	cfg.InactiveSeriesTTLSeconds = 0
+	cfg.InactiveSeriesCheckIntervalSeconds = 0
 	cfg.MaxCorrelations = -1
 	cfg.TrackCorrelationHistory = true
 	return cfg