[AAD-36] Use anomaly scorer by default (#54716)

<!--Please give us some feedback on your experience writing this PR ! https://app.datadoghq.com/forms/43db4c02-6837-400c-8083-692e141b1b88 !-->

### What does this PR do?

This uses anomaly scorer by default instead of time cluster which is now disabled by default.

### Motivation

We want to enable the scorer by default since we rely on it for smart adaptive sampling feature so our current direction is towards detection change points instead of sending events when we have bad behaviors.

### Describe how you validated your changes

Ran the benchmarks

### Additional Notes

Scenario | Before F1 (time cluster) | After F1 (scorer) | Δ F1
-- | -- | -- | --
block-building-outage | 0.285714 | 0.634606 | +0.348892
cascading-payment-failure | 0.063224 | 0.000083 | -0.063141
cassandra-repair-degradation | 0.227017 | 0.000000 | -0.227017
dns-upstream-outage | 0.054111 | 0.297905 | +0.243794
kafka-partition-saturation | 0.177494 | 0.577355 | +0.399861
lock-contention | 0.618715 | 0.032148 | -0.586567
memcached-saturation | 0.000006 | 0.557888 | +0.557882
pool-saturation | 0.000259 | 0.814789 | +0.814530
redis-cascade-billing | ~0 | ~0 | ~0
redis-cpu-saturation | 0.615742 | 0.016948 | -0.598794
tiered-cache-header-corruption | 0.001845 | ~0 | -0.001845
**Average** | 0.185830 | 0.266520 | +0.080690 (+43.42%)


Co-authored-by: celian.raimbault <celian.raimbault@datadoghq.com>

Author: CelianR

comp/anomalydetection/observer/AGENTS.md

@@ -41,7 +41,7 @@ and the testbench use the same engine.
 | `impl/agent_logs.go` | Agent internal log tap (source: `agent_logs`) |
 | `impl/log_pattern_extractor.go` | Log → virtual metrics via pattern clustering |
 | `impl/log_metrics_extractor.go` | Log → virtual metrics via regex extraction |
-| `impl/anomaly_correlator_time_cluster.go` | Default time-proximity correlator |
+| `impl/anomaly_correlator_time_cluster.go` | Time-proximity correlator |
 | `impl/anomaly_correlator_passthrough.go` | Passthrough correlator (one ActiveCorrelation per anomaly) |
 | `impl/anomaly_scorer.go` | Unified EWMA anomaly scorer (Correlator + standalone replay); derives severity, delegates push subscriptions to `severityevents/impl.Dispatcher` |
 | `impl/correlation_emitter.go` | Shared first-seen/recurrence helper used by all non-scorer correlators |
@@ -59,7 +59,7 @@ Registered in `impl/component_catalog.go`. Enabled by default unless noted:
 | Detector | `bocpd` | on |
 | Detector | `rrcf` | on |
 | Detector | `cusum`, `scanmw`, `scanwelch`, `holt_residual`, `tukey_biweight` | off |
-| Correlator | `time_cluster` | on |
+| Correlator | `time_cluster` | off |
 | Correlator | `cross_signal`, `passthrough` | off |
 | Correlator | `anomaly_scorer` | off |
 

comp/anomalydetection/observer/impl/component_catalog.go

@@ -90,11 +90,14 @@ type ComponentSettings struct {
 	configs map[string]any
 }
 
-// ApplyTestbenchDetectorDefaults applies the shorter detector warmups used by
-// the offline testbench. Production builds ComponentSettings from the agent
-// config and never calls this function. An explicit testbench --config entry
-// for a detector takes precedence over this profile.
-func ApplyTestbenchDetectorDefaults(settings ComponentSettings) ComponentSettings {
+// ApplyTestbenchDefaults applies replay-specific defaults used by the offline
+// testbench. Production builds ComponentSettings from the agent config and
+// never calls this function. Explicit testbench --config component entries
+// take precedence over this profile.
+func ApplyTestbenchDefaults(settings ComponentSettings) ComponentSettings {
+	if settings.Enabled == nil {
+		settings.Enabled = make(map[string]bool)
+	}
 	if settings.configs == nil {
 		settings.configs = make(map[string]any)
 	}
@@ -117,6 +120,19 @@ func ApplyTestbenchDetectorDefaults(settings ComponentSettings) ComponentSetting
 			settings.configs[name] = cfg
 		}
 	}
+
+	// Evals should score the scorer's correlation episodes. Preserve explicit
+	// component choices in --config, which are used for ablations and manual
+	// comparisons.
+	if _, explicitlyEnabled := settings.Enabled["anomaly_scorer"]; !explicitlyEnabled {
+		settings.Enabled["anomaly_scorer"] = true
+	}
+	if _, explicitlyConfigured := settings.configs["anomaly_scorer"]; !explicitlyConfigured {
+		scorer := DefaultAnomalyScorerConfig()
+		scorer.CorrelationEvents = true
+		scorer.CooldownSecs = 0
+		settings.configs["anomaly_scorer"] = scorer
+	}
 	return settings
 }
 
@@ -291,7 +307,7 @@ func defaultCatalog() *componentCatalog {
 				kind:           componentCorrelator,
 				defaultConfig:  DefaultTimeClusterConfig(),
 				factory:        func(cfg any) any { return NewTimeClusterCorrelator(cfg.(TimeClusterConfig)) },
-				defaultEnabled: true,
+				defaultEnabled: false,
 				readConfig:     readTimeClusterConfig,
 				parseJSON: func(defaults any, raw []byte) (any, error) {
 					cfg := defaults.(TimeClusterConfig)

comp/anomalydetection/observer/impl/component_catalog_test.go

@@ -67,8 +67,8 @@ func TestValidateDetectorTeardownContract_AllowlistEscape(t *testing.T) {
 	require.NoError(t, cat.validateDetectorTeardownContract())
 }
 
-func TestApplyTestbenchDetectorDefaults(t *testing.T) {
-	settings := ApplyTestbenchDetectorDefaults(ComponentSettings{})
+func TestApplyTestbenchDefaults(t *testing.T) {
+	settings := ApplyTestbenchDefaults(ComponentSettings{})
 
 	require.Equal(t, 40, settings.configs["bocpd"].(BOCPDConfig).WarmupPoints)
 	holt := settings.configs["holt_residual"].(HoltResidualConfig)
@@ -77,16 +77,25 @@ func TestApplyTestbenchDetectorDefaults(t *testing.T) {
 	tukey := settings.configs["tukey_biweight"].(TukeyBiweightConfig)
 	require.Equal(t, 40, tukey.WindowSize)
 	require.Equal(t, 40, tukey.MinPoints)
+	require.True(t, settings.Enabled["anomaly_scorer"])
+	require.NotContains(t, settings.Enabled, "time_cluster")
+	scorer := settings.configs["anomaly_scorer"].(AnomalyScorerConfig)
+	require.True(t, scorer.CorrelationEvents)
+	require.Zero(t, scorer.CooldownSecs)
 }
 
-func TestApplyTestbenchDetectorDefaults_PreservesExplicitConfig(t *testing.T) {
+func TestApplyTestbenchDefaults_PreservesExplicitConfig(t *testing.T) {
 	settings, err := ParseSettingsFromJSON(map[string]json.RawMessage{
-		"bocpd": json.RawMessage(`{"warmup_points": 42}`),
+		"bocpd":          json.RawMessage(`{"warmup_points": 42}`),
+		"anomaly_scorer": json.RawMessage(`{"enabled":false}`),
+		"time_cluster":   json.RawMessage(`{"enabled":true}`),
 	})
 	require.NoError(t, err)
 
-	settings = ApplyTestbenchDetectorDefaults(settings)
+	settings = ApplyTestbenchDefaults(settings)
 	require.Equal(t, 42, settings.configs["bocpd"].(BOCPDConfig).WarmupPoints)
+	require.False(t, settings.Enabled["anomaly_scorer"])
+	require.True(t, settings.Enabled["time_cluster"])
 }
 
 // bareDetectorForValidator is a minimal observerdef.Detector that

internal/qbranch/anomalydetection-testbench/README.md

@@ -91,7 +91,8 @@ $ dda inv anomalydetection.eval-component-workspace-report evals # This will fet
 
 | Name | Default | Description |
 |------|---------|-------------|
-| `time_cluster` | enabled | Groups anomalies that occur close together in time |
+| `anomaly_scorer` | enabled | Produces anomaly periods from the EWMA anomaly-severity score |
+| `time_cluster` | disabled | Groups anomalies that occur close together in time |
 | `cross_signal` | disabled | Cross-signal pattern correlator (fixed known patterns) |
 | `passthrough` | disabled | Passes every anomaly through as its own correlation (for TP metric scoring) |
 
@@ -108,7 +109,7 @@ Extractors are always enabled and convert raw observations into timeseries:
 ## Examples
 
 ```bash
-# Run with all defaults (bocpd + rrcf + time_cluster)
+# Run with all defaults (bocpd + rrcf + anomaly_scorer)
 dda inv -- anomalydetection.launch-testbench
 
 # Only BOCPD + TimeCluster

internal/qbranch/anomalydetection-testbench/main.go

@@ -150,7 +150,7 @@ func main() {
 			MuteNoisyMetrics: *muteNoisyMetrics,
 		}
 	}
-	componentSettings = observerimpl.ApplyTestbenchDetectorDefaults(componentSettings)
+	componentSettings = observerimpl.ApplyTestbenchDefaults(componentSettings)
 
 	if *headless == "" {
 		fmt.Printf("Observer Test Bench\n")

pkg/config/schema/yaml/core_schema.yaml

@@ -5193,7 +5193,7 @@ properties:
               enabled:
                 node_type: setting
                 type: boolean
-                default: true
+                default: false
               min_cluster_size:
                 node_type: setting
                 type: integer

tasks/anomalydetection.py

@@ -205,7 +205,8 @@ def eval_scenarios(
     source of truth for anomaly detection accuracy.
 
     Uses testbench --only to control which components are active.
-    Default (no --only): uses testbench defaults (bocpd,rrcf,time_cluster + other default-enabled components).
+    Default (no --only): uses testbench defaults (bocpd, rrcf, and
+      anomaly_scorer; time_cluster is disabled).
     With --only: enables ONLY listed components + extractors, disables everything else.
       time_cluster is auto-added if not specified.
     With --config: JSON params file for testbench; overrides --only when both are set.

test/new-e2e/tests/anomalydetection/anomalydetection_nix_test.go

@@ -50,9 +50,16 @@ func TestAnomalyDetectionMetricsTriggered(t *testing.T) {
 	agentConfig := `
 log_level: debug
 anomaly_detection:
-  anomaly_scorer:
-    dry_run:
+  reporting:
+    events:
       enabled: true
+  anomaly_scorer:
+    # CUSUM produces a single anomalous series in this test. Keep the scorer
+    # thresholds below its first EWMA update so the report path sees High.
+    low_threshold: 0.000001
+    high_threshold: 0.00001
+    output:
+      correlation_events: true
   metrics:
     enabled: true
   logs:
@@ -75,7 +82,7 @@ anomaly_detection:
 }
 
 // TestMetricsTriggeredEmitsOnDSDSpike sends a stable gauge baseline then a large
-// spike, expecting CUSUM to fire and the stdout reporter to emit its marker.
+// spike, expecting CUSUM to fire and the scorer to open an episode.
 //
 // Point counts: 15 baseline (well above the 5-point CUSUM minimum) followed by
 // 10 spike points — total ~25 seconds of data. The spike is 5000× the baseline
@@ -143,8 +150,8 @@ func (s *metricsTriggeredSuite) TestMetricsTriggeredEmitsOnDSDSpike() {
 		s.T().Log("done sending metrics")
 	}()
 
-	waitForReportsTelemetry(s)
-	s.T().Log("reports telemetry detected")
+	waitForScorerEpisode(s, metricName)
+	s.T().Log("scorer episode detected")
 }
 
 // logTriggeredSuite exercises the external log collection path of the observer.
@@ -187,9 +194,16 @@ log_level: debug
 logs_config:
   file_scan_period: 1
 anomaly_detection:
-  anomaly_scorer:
-    dry_run:
+  reporting:
+    events:
       enabled: true
+  anomaly_scorer:
+    # BOCPD produces one anomalous series. Cross High on its first EWMA update
+    # so the reporter reliably observes a scorer correlation event.
+    low_threshold: 0.000001
+    high_threshold: 0.00001
+    output:
+      correlation_events: true
   metrics:
     enabled: false
   logs:
@@ -312,6 +326,6 @@ func (s *logTriggeredSuite) TestLogsTriggeredEmitsOnFileSpike() {
 		s.T().Log("done writing log lines")
 	}()
 
-	waitForReportsTelemetry(s)
-	s.T().Log("reports telemetry detected via log trigger")
+	waitForScorerEpisode(s, "filename:e2e-anomaly-test.log")
+	s.T().Log("scorer episode detected via log trigger")
 }

test/new-e2e/tests/anomalydetection/helpers_test.go

@@ -43,13 +43,18 @@ const (
 	telemetryLogsIngested   = "observer.logs.ingested"
 	telemetryReportsEmitted = "observer.reports.emitted"
 	telemetryReportsOngoing = "observer.reports.ongoing"
-
 	// scorerHelperEscalationMarker is emitted by anomalyScorer.OnSeverityTransition
 	// when output.logs=true and the EWMA rises above low_threshold (an escalation event).
 	// Logged at info level, captured by journald, and serves as the assertion target.
 	// Full example: "[observer] anomaly scorer anomaly_scorer severity escalation to Medium (was Low, t=...)"
 	scorerHelperEscalationMarker = "[observer] anomaly scorer anomaly_scorer severity escalation"
 
+	// scorerEpisodeStartedMarker is emitted by the stdout reporter when the
+	// scorer opens an episode after reaching High severity. The reporter appends
+	// either the scorer metadata or a multiline contributor summary, so retain
+	// only the prefix common to both renderings.
+	scorerEpisodeStartedMarker = "[observer] scorer episode started:"
+
 	// scorerHelperRegisteredMarker is logged once at agent startup when the
 	// anomaly scorer is successfully wired with telemetry. Waiting for it
 	// before sending metrics ensures the scorer is active.
@@ -174,14 +179,17 @@ func sendGauge(s observerTestSuite, name string, value float64) {
 	}
 }
 
-func waitForReportsTelemetry(s observerTestSuite) {
+func waitForScorerEpisode(s observerTestSuite, expectedAnomalySource string) {
 	s.T().Helper()
-	s.T().Log("waiting for observer reports telemetry...")
+	s.T().Log("waiting for anomaly scorer episode...")
 	s.EventuallyWithT(func(c *assert.CollectT) {
 		assert.True(c, s.Env().Agent.Client.IsReady(), "agent should be ready")
-		tel := observerTelemetryOutput(s)
-		assert.True(c, containsMetric(tel, telemetryReportsEmitted),
-			"observer telemetry should expose reports emitted counter after anomalies")
+		out, err := s.Env().RemoteHost.Execute("sudo journalctl -u datadog-agent --no-pager")
+		assert.NoError(c, err, "journalctl execution failed")
+		assert.Contains(c, out, scorerEpisodeStartedMarker,
+			"journal should contain the scorer episode-started marker after anomalies")
+		assert.Contains(c, out, expectedAnomalySource,
+			"journal should contain an anomaly from the test input source")
 	}, 3*time.Minute, 5*time.Second)
-	s.T().Log("observer reports telemetry detected")
+	s.T().Log("anomaly scorer episode detected")
 }