[AAD-21] Start baseline qualification on detector readiness

Author: CelianR

comp/anomalydetection/observer/def/types.go

@@ -14,7 +14,6 @@ import (
 	"sort"
 	"strconv"
 	"strings"
-	"time"
 
 	severityeventsdef "github.com/DataDog/datadog-agent/comp/anomalydetection/severityevents/def"
 )
@@ -287,8 +286,9 @@ type DetectionResult struct {
 type SeriesDetector interface {
 	// Name returns the analysis name for debugging.
 	Name() string
-	// BaselineSpec reports how long the detector needs to build a usable model.
-	BaselineSpec() BaselineSpec
+	// Ready reports whether at least one series has reached the detector's
+	// actual scoring condition. It is monotonic until Reset.
+	Ready() bool
 	// Detect examines a series and returns any detected anomalies.
 	Detect(series Series) DetectionResult
 }
@@ -573,19 +573,14 @@ type StorageReader interface {
 	SeriesGeneration() uint64
 }
 
-// BaselineSpec describes the detector's model warmup before its qualification
-// baseline begins.
-type BaselineSpec struct {
-	WarmupDuration time.Duration
-}
-
 // Detector is the flexible detection interface where detectors pull data from storage.
 // This supports multivariate detection across multiple series.
 type Detector interface {
 	Name() string
 
-	// BaselineSpec reports how long the detector needs to build a usable model.
-	BaselineSpec() BaselineSpec
+	// Ready reports whether at least one series has reached the detector's
+	// actual scoring condition. It is monotonic until Reset.
+	Ready() bool
 
 	// Detect is called periodically by the scheduler.
 	// The detector queries storage for whatever data it needs.

comp/anomalydetection/observer/impl/baseline.go

@@ -7,13 +7,12 @@ package observerimpl
 
 import (
 	"sort"
-	"time"
 
 	observerdef "github.com/DataDog/datadog-agent/comp/anomalydetection/observer/def"
 )
 
 // BaselineConfig controls detector-specific baseline qualification windows.
-// DurationSec is the qualification duration after a detector's own warmup.
+// DurationSec is the qualification duration after a detector becomes ready.
 type BaselineConfig struct {
 	Enabled          bool
 	DurationSec      int64
@@ -26,13 +25,8 @@ func DefaultBaselineConfig() BaselineConfig {
 	return BaselineConfig{Enabled: true, DurationSec: 600, MuteNoisyMetrics: true}
 }
 
-// baselineReferenceInterval translates point-count detector requirements into
-// data-time warmups. It is deliberately a scheduling contract, not a claim
-// about the cadence of every incoming series.
-const baselineReferenceInterval = 15 * time.Second
-
 type detectorBaselineState struct {
-	spec               observerdef.BaselineSpec
+	ready              bool
 	warmupEndSec       int64
 	baselineEndSec     int64
 	completed          bool
@@ -44,19 +38,21 @@ type detectorBaselineState struct {
 // BaselineDetectorDebugStatus is a testbench-facing snapshot of one detector.
 type BaselineDetectorDebugStatus struct {
 	Name           string `json:"name"`
-	WarmupEndSec   int64  `json:"warmupEndSec"`
-	BaselineEndSec int64  `json:"baselineEndSec"`
+	Ready          bool   `json:"ready"`
+	WarmupEndSec   int64  `json:"warmupEndSec,omitempty"`
+	BaselineEndSec int64  `json:"baselineEndSec,omitempty"`
 	Completed      bool   `json:"completed"`
 	MutedCount     int    `json:"mutedCount"`
 }
 
 // BaselineDebugStatus is a testbench-facing snapshot of the baseline union.
 type BaselineDebugStatus struct {
-	Started     bool                          `json:"started"`
-	StartSec    int64                         `json:"startSec"`
-	AllComplete bool                          `json:"allComplete"`
-	MutedCount  int                           `json:"mutedCount"`
-	Detectors   []BaselineDetectorDebugStatus `json:"detectors"`
+	Started            bool                          `json:"started"`
+	StartSec           int64                         `json:"startSec"`
+	AnalyzedThroughSec int64                         `json:"analyzedThroughSec,omitempty"`
+	AllComplete        bool                          `json:"allComplete"`
+	MutedCount         int                           `json:"mutedCount"`
+	Detectors          []BaselineDetectorDebugStatus `json:"detectors"`
 }
 
 // baselineController coordinates independent detector windows. All methods
@@ -82,38 +78,43 @@ type baselineController struct {
 	mutedNames  map[string]struct{} // allocated only for the final verbose summary
 }
 
-func newBaselineController(cfg BaselineConfig, detectors []detectorBaselineSpecEntry) *baselineController {
+func newBaselineController(cfg BaselineConfig, detectorNames []string) *baselineController {
 	b := &baselineController{config: cfg, detectors: make(map[string]*detectorBaselineState), mutedHashes: make(map[uint64]struct{})}
-	for _, d := range detectors {
-		b.detectors[d.name] = &detectorBaselineState{spec: d.spec, pendingHashes: make(map[uint64]struct{})}
+	for _, name := range detectorNames {
+		b.detectors[name] = &detectorBaselineState{pendingHashes: make(map[uint64]struct{})}
 	}
 	return b
 }
 
-type detectorBaselineSpecEntry struct {
-	name string
-	spec observerdef.BaselineSpec
-}
-
-func baselineSpecs(detectors []observerdef.Detector) []detectorBaselineSpecEntry {
-	entries := make([]detectorBaselineSpecEntry, 0, len(detectors))
+func detectorNames(detectors []observerdef.Detector) []string {
+	names := make([]string, 0, len(detectors))
 	for _, detector := range detectors {
-		entries = append(entries, detectorBaselineSpecEntry{name: detector.Name(), spec: detector.BaselineSpec()})
+		names = append(names, detector.Name())
 	}
-	return entries
+	return names
 }
 
-// start seeds all detector windows from the first analysis data timestamp.
+// start records the first analysis timestamp. Individual qualification windows
+// begin only when their detector reports that it is ready to score.
 func (b *baselineController) start(dataSec int64) {
 	if b.started {
 		return
 	}
 	b.started = true
 	b.startSec = dataSec
-	for _, state := range b.detectors {
-		state.warmupEndSec = dataSec + int64(state.spec.WarmupDuration/time.Second)
-		state.baselineEndSec = state.warmupEndSec + b.config.DurationSec
+}
+
+// ready records a detector's first usable scoring advance and starts its
+// qualification baseline. It returns true only for that first transition.
+func (b *baselineController) ready(name string, dataSec int64) bool {
+	state := b.detectors[name]
+	if state == nil || state.ready {
+		return false
 	}
+	state.ready = true
+	state.warmupEndSec = dataSec
+	state.baselineEndSec = dataSec + b.config.DurationSec
+	return true
 }
 
 // isAnalyzingAt reports whether the detector's baseline decision is still in
@@ -123,12 +124,15 @@ func (b *baselineController) isAnalyzingAt(name string, dataSec int64) bool {
 	if state == nil || state.completed {
 		return false
 	}
+	if !state.ready {
+		return true
+	}
 	return dataSec < state.baselineEndSec
 }
 
 func (b *baselineController) mark(name string, h uint64) {
 	state := b.detectors[name]
-	if state == nil || state.completed {
+	if state == nil || state.completed || !state.ready {
 		return
 	}
 	state.windowAnomalyCount++
@@ -138,7 +142,7 @@ func (b *baselineController) mark(name string, h uint64) {
 func (b *baselineController) due(dataSec int64) []string {
 	var names []string
 	for name, state := range b.detectors {
-		if !state.completed && dataSec >= state.baselineEndSec {
+		if state.ready && !state.completed && dataSec >= state.baselineEndSec {
 			names = append(names, name)
 		}
 	}
@@ -147,7 +151,7 @@ func (b *baselineController) due(dataSec int64) []string {
 
 func (b *baselineController) complete(name string) (newHashes map[uint64]struct{}, snapshotChanged bool, anomalyCount int, allComplete bool) {
 	state := b.detectors[name]
-	if state == nil || state.completed {
+	if state == nil || !state.ready || state.completed {
 		return nil, false, 0, b.allComplete()
 	}
 	state.completed = true
@@ -198,7 +202,7 @@ func (b *baselineController) debugStatus() BaselineDebugStatus {
 		if state.completed {
 			mutedCount = state.mutedCount
 		}
-		status.Detectors = append(status.Detectors, BaselineDetectorDebugStatus{Name: name, WarmupEndSec: state.warmupEndSec, BaselineEndSec: state.baselineEndSec, Completed: state.completed, MutedCount: mutedCount})
+		status.Detectors = append(status.Detectors, BaselineDetectorDebugStatus{Name: name, Ready: state.ready, WarmupEndSec: state.warmupEndSec, BaselineEndSec: state.baselineEndSec, Completed: state.completed, MutedCount: mutedCount})
 	}
 	sort.Slice(status.Detectors, func(i, j int) bool { return status.Detectors[i].Name < status.Detectors[j].Name })
 	return status

comp/anomalydetection/observer/impl/baseline_test.go

@@ -7,7 +7,6 @@ package observerimpl
 
 import (
 	"testing"
-	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -27,7 +26,8 @@ type alwaysFiringDetector struct {
 // records series reclamation from another detector's baseline completion.
 type baselineTestDetector struct {
 	name          string
-	spec          observerdef.BaselineSpec
+	readyAtSec    int64
+	ready         bool
 	source        observerdef.SeriesDescriptor
 	ref           observerdef.SeriesRef
 	emitAfterSec  int64
@@ -36,10 +36,11 @@ type baselineTestDetector struct {
 }
 
 func (d *baselineTestDetector) Name() string { return d.name }
-func (d *baselineTestDetector) BaselineSpec() observerdef.BaselineSpec {
-	return d.spec
-}
+func (d *baselineTestDetector) Ready() bool  { return d.ready }
 func (d *baselineTestDetector) Detect(_ observerdef.StorageReader, dataSec int64) observerdef.DetectionResult {
+	if dataSec >= d.readyAtSec {
+		d.ready = true
+	}
 	if dataSec < d.emitAfterSec {
 		return observerdef.DetectionResult{}
 	}
@@ -59,9 +60,7 @@ func (d *baselineTestDetector) RemoveSeries(refs []observerdef.SeriesRef) {
 }
 
 func (d *alwaysFiringDetector) Name() string { return "always_firing" }
-func (*alwaysFiringDetector) BaselineSpec() observerdef.BaselineSpec {
-	return observerdef.BaselineSpec{}
-}
+func (*alwaysFiringDetector) Ready() bool    { return true }
 func (d *alwaysFiringDetector) Detect(_ observerdef.StorageReader, dataTime int64) observerdef.DetectionResult {
 	return observerdef.DetectionResult{
 		Anomalies: []observerdef.Anomaly{{
@@ -109,21 +108,20 @@ func makeBaselineEngine(cfg BaselineConfig, correlator observerdef.Correlator) (
 
 // ---- baselineController unit tests ----
 
-func TestBaselineController_DetectorSpecificWindows(t *testing.T) {
-	b := newBaselineController(BaselineConfig{DurationSec: 600}, []detectorBaselineSpecEntry{
-		{name: "fast", spec: observerdef.BaselineSpec{}},
-		{name: "slow", spec: observerdef.BaselineSpec{WarmupDuration: 5 * time.Minute}},
-	})
+func TestBaselineController_DetectorReadinessStartsIndependentWindows(t *testing.T) {
+	b := newBaselineController(BaselineConfig{DurationSec: 600}, []string{"fast", "slow"})
 	assert.False(t, b.debugStatus().Started)
 	b.start(1000)
 	status := b.debugStatus()
 	assert.True(t, status.Started)
 	require.Len(t, status.Detectors, 2)
-	assert.Equal(t, BaselineDetectorDebugStatus{Name: "fast", WarmupEndSec: 1000, BaselineEndSec: 1600}, status.Detectors[0])
-	assert.Equal(t, BaselineDetectorDebugStatus{Name: "slow", WarmupEndSec: 1300, BaselineEndSec: 1900}, status.Detectors[1])
+	assert.Equal(t, BaselineDetectorDebugStatus{Name: "fast"}, status.Detectors[0])
+	assert.Equal(t, BaselineDetectorDebugStatus{Name: "slow"}, status.Detectors[1])
 	assert.True(t, b.isAnalyzingAt("fast", 1000))
 	assert.True(t, b.isAnalyzingAt("slow", 1299))
-	assert.True(t, b.isAnalyzingAt("slow", 1300))
+	assert.True(t, b.ready("fast", 1000))
+	assert.False(t, b.ready("fast", 1001))
+	assert.True(t, b.ready("slow", 1300))
 	assert.True(t, b.isAnalyzingAt("slow", 1899))
 	assert.False(t, b.isAnalyzingAt("slow", 1900))
 	assert.False(t, b.isAnalyzingAt("unknown", 1000))
@@ -147,12 +145,26 @@ func TestBaselineController_DetectorSpecificWindows(t *testing.T) {
 	assert.True(t, b.debugStatus().AllComplete)
 }
 
+func TestBaselineController_WaitingDetectorSuppressesWithoutMuting(t *testing.T) {
+	b := newBaselineController(BaselineConfig{DurationSec: 60}, []string{"waiting"})
+	b.start(100)
+
+	assert.True(t, b.isAnalyzingAt("waiting", 100))
+	b.mark("waiting", 1) // defensive suppression before Ready must not mute
+	assert.Empty(t, b.detectors["waiting"].pendingHashes)
+	assert.Empty(t, b.due(1_000))
+	assert.False(t, b.allComplete())
+
+	b.ready("waiting", 130)
+	b.mark("waiting", 1)
+	assert.Equal(t, []string{"waiting"}, b.due(190))
+}
+
 func TestBaselineController_CompletionPublishesImmutableUnionAndReleasesPendingHashes(t *testing.T) {
-	b := newBaselineController(BaselineConfig{DurationSec: 600}, []detectorBaselineSpecEntry{
-		{name: "first", spec: observerdef.BaselineSpec{}},
-		{name: "second", spec: observerdef.BaselineSpec{}},
-	})
+	b := newBaselineController(BaselineConfig{DurationSec: 600}, []string{"first", "second"})
 	b.start(1000)
+	b.ready("first", 1000)
+	b.ready("second", 1000)
 	b.mark("first", 1)
 	b.mark("second", 1)
 	b.mark("second", 2)
@@ -178,11 +190,10 @@ func TestBaselineController_CompletionPublishesImmutableUnionAndReleasesPendingH
 }
 
 func TestBaselineController_DuplicateCompletionDoesNotReplaceUnionSnapshot(t *testing.T) {
-	b := newBaselineController(BaselineConfig{DurationSec: 600}, []detectorBaselineSpecEntry{
-		{name: "first", spec: observerdef.BaselineSpec{}},
-		{name: "second", spec: observerdef.BaselineSpec{}},
-	})
+	b := newBaselineController(BaselineConfig{DurationSec: 600}, []string{"first", "second"})
 	b.start(1000)
+	b.ready("first", 1000)
+	b.ready("second", 1000)
 	b.mark("first", 1)
 	b.mark("second", 1)
 
@@ -204,13 +215,6 @@ func TestBaselineController_VerboseNamesAreLazyAndReleased(t *testing.T) {
 	assert.Nil(t, b.mutedNames)
 }
 
-func TestRRCFBaselineSpec_UsesAlignedReadiness(t *testing.T) {
-	rrcf := NewRRCFDetector(RRCFConfig{NumTrees: 1, TreeSize: 64, ShingleSize: 4})
-	spec := rrcf.BaselineSpec()
-
-	assert.Equal(t, 78*baselineReferenceInterval, spec.WarmupDuration)
-}
-
 // ---- engine integration tests ----
 
 func TestBaseline_AnomaliesHeldDuringWindow(t *testing.T) {
@@ -233,14 +237,38 @@ func TestBaseline_AnomaliesForwardedAfterWindow(t *testing.T) {
 	assert.NotEmpty(t, correlator.received)
 }
 
+func TestBaseline_WaitingDetectorDoesNotMuteUntilReady(t *testing.T) {
+	storage := newTimeSeriesStorage()
+	ref := storage.Add("ns", "cpu", 1.0, 100, nil).Ref
+	detector := &baselineTestDetector{
+		name:          "waiting",
+		readyAtSec:    200,
+		emitAfterSec:  100,
+		includeSource: true,
+		ref:           ref,
+		source:        observerdef.SeriesDescriptor{Namespace: "ns", Name: "cpu", Aggregate: AggregateAverage},
+	}
+	e := newEngine(engineConfig{storage: storage, detectors: []observerdef.Detector{detector}, baseline: BaselineConfig{Enabled: true, DurationSec: 100, MuteNoisyMetrics: true}})
+
+	e.Advance(100) // detector emits before Ready; it is suppressed but cannot mute
+	assert.Empty(t, e.baseline.mutedHashes)
+	assert.False(t, e.baseline.detectors["waiting"].ready)
+
+	e.Advance(200) // readiness transition anomaly is included in qualification
+	assert.True(t, e.baseline.detectors["waiting"].ready)
+	assert.Len(t, e.baseline.detectors["waiting"].pendingHashes, 1)
+	e.Advance(300)
+	assert.Len(t, e.baseline.mutedHashes, 1)
+}
+
 func TestBaseline_FastCompletionRemovesSeriesFromSlowerDetector(t *testing.T) {
 	storage := newTimeSeriesStorage()
 	ref := storage.Add("ns", "cpu", 1.0, 100, nil).Ref
 	source := observerdef.SeriesDescriptor{Namespace: "ns", Name: "cpu", Aggregate: AggregateAverage}
 	fast := &baselineTestDetector{name: "fast", source: source, ref: ref, includeSource: true}
 	slow := &baselineTestDetector{
 		name:          "slow",
-		spec:          observerdef.BaselineSpec{WarmupDuration: 5 * time.Minute},
+		readyAtSec:    400,
 		source:        source,
 		ref:           ref,
 		includeSource: true,
@@ -272,7 +300,7 @@ func TestBaseline_FastDetectorForwardsWhileSlowerDetectorStillAnalyses(t *testin
 		emitAfterSec:  200,
 		includeSource: true,
 	}
-	slow := &baselineTestDetector{name: "slow", spec: observerdef.BaselineSpec{WarmupDuration: 5 * time.Minute}, emitAfterSec: 1<<62 - 1}
+	slow := &baselineTestDetector{name: "slow", readyAtSec: 400, emitAfterSec: 1<<62 - 1}
 	correlator := &recordingCorrelator{}
 	e := newEngine(engineConfig{
 		storage:     storage,
@@ -382,9 +410,7 @@ func TestBaseline_MuteNoisyMetricsFalseDoesNotDropMetrics(t *testing.T) {
 type storageAwareDetector struct{}
 
 func (d *storageAwareDetector) Name() string { return "storage_aware" }
-func (*storageAwareDetector) BaselineSpec() observerdef.BaselineSpec {
-	return observerdef.BaselineSpec{}
-}
+func (*storageAwareDetector) Ready() bool    { return true }
 func (d *storageAwareDetector) Detect(sr observerdef.StorageReader, dataTime int64) observerdef.DetectionResult {
 	metas := sr.ListSeries(observerdef.SeriesFilter{})
 	anomalies := make([]observerdef.Anomaly, 0, len(metas))