[anomalydetection] Smart adaptive log sampling logic 3/4 (#53431)

### What does this PR do?

> [!NOTE]
> [Feature documentation](https://datadoghq.atlassian.net/wiki/x/EgK5nQE).

This links anomaly detection events to adaptive sampling to have different profiles given severity state.

```yaml
logs_config:
  experimental_adaptive_sampling:
    enabled: true
    rate_limit: 1
    burst_size: 100
    # New config
    smart_severity_profiles:
      enabled: true
      medium:
        rate_limit: 1
        burst_size: 1000
      high:
        rate_limit: 100
        burst_size: 1000
```

Adds:

- Logic to `preprocessor/sampler.go` such that it reads the `severityevents` state and apply the specific sampling profile given the severity
- `smart_adaptive_log_sampling_nix_test.go` e2e test to test this feature (1. check high sampling in low state, 2. check low sampling in high state)

### Motivation

### Describe how you validated your changes

1. Config smart severity profiles + adaptive sampling
2. The low profile must be the first one applied
3. When we have anomalies (medium / high severity), the other profiles must be set. We can set the burst size to a high value and check if bursts of logs are sent or not

There is an e2e tests for this.

### Additional Notes

Passthrough / cooldown are left for a [future PR](#53432).

<img width="573" height="800" alt="Screenshot 2026-07-09 at 09 41 50" src="https://github.com/user-attachments/assets/343e59d4-ab53-4977-8f81-071dbc2cbe6d" />

[AAD-4]: https://datadoghq.atlassian.net/browse/AAD-4?atlOrigin=eyJpIjoiNWRkNTljNzYxNjVmNDY3MDlhMDU5Y2ZhYzA5YTRkZjUiLCJwIjoiZ2l0aHViLWNvbS1KU1cifQ

Co-authored-by: eman.okyere <eman.okyere@datadoghq.com>

Author: CelianR

pkg/logs/internal/decoder/BUILD.bazel

@@ -31,9 +31,11 @@ go_library(
         "@com_github_datadog_datadog_agent_pkg_logs_util_testutils//:__subpackages__",
     ],
     deps = [
+        "//comp/anomalydetection/severityevents/def",
         "//comp/core/telemetry/def",
         "//comp/logs-library/metrics",
         "//comp/logs/agent/config",
+        "//comp/logs/severityprovider/def",
         "//pkg/config/setup",
         "//pkg/config/structure",
         "//pkg/logs/internal/decoder/preprocessor",
@@ -65,6 +67,7 @@ dd_agent_go_test(
     ],
     embed = [":decoder"],
     deps = [
+        "//comp/anomalydetection/severityevents/def",
         "//comp/logs/agent/config",
         "//pkg/config/mock",
         "//pkg/config/model",

pkg/logs/internal/decoder/decoder.go

@@ -9,7 +9,9 @@ import (
 	"regexp"
 	"time"
 
+	severityeventsdef "github.com/DataDog/datadog-agent/comp/anomalydetection/severityevents/def"
 	"github.com/DataDog/datadog-agent/comp/logs/agent/config"
+	severityprovider "github.com/DataDog/datadog-agent/comp/logs/severityprovider/def"
 	pkgconfigsetup "github.com/DataDog/datadog-agent/pkg/config/setup"
 	"github.com/DataDog/datadog-agent/pkg/config/structure"
 	"github.com/DataDog/datadog-agent/pkg/logs/internal/decoder/preprocessor"
@@ -189,6 +191,47 @@ func resolveNoisyLogDetectionEnabled(sourceNoisyLogDetection *bool) bool {
 
 const disabledSourcesConfigKey = "logs_config.experimental_adaptive_sampling.disabled_sources"
 
+const (
+	smartSeverityProfilesEnabledConfigKey         = "logs_config.experimental_adaptive_sampling.smart_severity_profiles.enabled"
+	smartSeverityProfilesMediumRateLimitConfigKey = "logs_config.experimental_adaptive_sampling.smart_severity_profiles.medium.rate_limit"
+	smartSeverityProfilesMediumBurstSizeConfigKey = "logs_config.experimental_adaptive_sampling.smart_severity_profiles.medium.burst_size"
+	smartSeverityProfilesHighRateLimitConfigKey   = "logs_config.experimental_adaptive_sampling.smart_severity_profiles.high.rate_limit"
+	smartSeverityProfilesHighBurstSizeConfigKey   = "logs_config.experimental_adaptive_sampling.smart_severity_profiles.high.burst_size"
+)
+
+// resolveSmartSeverityProfiles builds the Low/Medium/High profile triple. Each field of
+// Medium/High cascades independently from the level below when left unconfigured (Low ->
+// Medium -> High), so no combination of partially-configured fields can leave a higher
+// severity level less permissive than the one below it.
+func resolveSmartSeverityProfiles(low preprocessor.SamplerProfile) [severityeventsdef.NumSeverityLevels]preprocessor.SamplerProfile {
+	cfg := pkgconfigsetup.Datadog()
+
+	profiles := [severityeventsdef.NumSeverityLevels]preprocessor.SamplerProfile{
+		severityeventsdef.SeverityLow:    low,
+		severityeventsdef.SeverityMedium: low,
+		severityeventsdef.SeverityHigh:   low,
+	}
+
+	if cfg.IsConfigured(smartSeverityProfilesMediumRateLimitConfigKey) {
+		profiles[severityeventsdef.SeverityMedium].RateLimit = cfg.GetFloat64(smartSeverityProfilesMediumRateLimitConfigKey)
+	}
+	if cfg.IsConfigured(smartSeverityProfilesMediumBurstSizeConfigKey) {
+		profiles[severityeventsdef.SeverityMedium].BurstSize = clampBurstSize(cfg.GetFloat64(smartSeverityProfilesMediumBurstSizeConfigKey))
+	}
+
+	// High starts from Medium's already-resolved profile, then applies its own
+	// overrides per field.
+	profiles[severityeventsdef.SeverityHigh] = profiles[severityeventsdef.SeverityMedium]
+	if cfg.IsConfigured(smartSeverityProfilesHighRateLimitConfigKey) {
+		profiles[severityeventsdef.SeverityHigh].RateLimit = cfg.GetFloat64(smartSeverityProfilesHighRateLimitConfigKey)
+	}
+	if cfg.IsConfigured(smartSeverityProfilesHighBurstSizeConfigKey) {
+		profiles[severityeventsdef.SeverityHigh].BurstSize = clampBurstSize(cfg.GetFloat64(smartSeverityProfilesHighBurstSizeConfigKey))
+	}
+
+	return profiles
+}
+
 func newDisabledSet() map[string]struct{} {
 	entries := pkgconfigsetup.Datadog().GetStringSlice(disabledSourcesConfigKey)
 	m := make(map[string]struct{}, len(entries))
@@ -276,7 +319,15 @@ func resolveAdaptiveSamplerConfig(sourceAdaptiveSampling *config.SourceAdaptiveS
 		}
 	}
 
-	return validateAdaptiveSamplerConfig(c)
+	c = validateAdaptiveSamplerConfig(c)
+
+	c.SmartSeverityProfilesEnabled = pkgconfigsetup.Datadog().GetBool(smartSeverityProfilesEnabledConfigKey)
+	if c.SmartSeverityProfilesEnabled {
+		c.Profiles = resolveSmartSeverityProfiles(preprocessor.SamplerProfile{RateLimit: c.RateLimit, BurstSize: c.BurstSize})
+		c.SeverityProvider = severityprovider.Current
+	}
+
+	return c
 }
 
 func resolveNoisyLogDetectionConfig(sourceAdaptiveSampling *config.SourceAdaptiveSamplingOptions, tok *preprocessor.Tokenizer) preprocessor.AdaptiveSamplerConfig {
@@ -425,13 +476,19 @@ func validateAdaptiveSamplerConfig(c preprocessor.AdaptiveSamplerConfig) preproc
 		c.MaxPatterns = 1
 	}
 
-	if c.BurstSize <= 0 {
-		c.BurstSize = 1
-	}
+	c.BurstSize = clampBurstSize(c.BurstSize)
 
 	return c
 }
 
+// clampBurstSize floors burstSize at 1, avoiding negative starting credits.
+func clampBurstSize(burstSize float64) float64 {
+	if burstSize <= 0 {
+		return 1
+	}
+	return burstSize
+}
+
 func getLegacyAutoMultilineHandler(outputFn func(*message.Message), multiLinePattern *regexp.Regexp, maxContentSize int, source *sources.ReplaceableSource, detectedPattern *DetectedPattern, tailerInfo *status.InfoRegistry) LineHandler {
 
 	if multiLinePattern != nil {

pkg/logs/internal/decoder/decoder_test.go

@@ -11,6 +11,7 @@ import (
 	"strings"
 	"testing"
 
+	severityeventsdef "github.com/DataDog/datadog-agent/comp/anomalydetection/severityevents/def"
 	"github.com/DataDog/datadog-agent/comp/logs/agent/config"
 	configmock "github.com/DataDog/datadog-agent/pkg/config/mock"
 	pkgconfigmodel "github.com/DataDog/datadog-agent/pkg/config/model"
@@ -741,6 +742,101 @@ func TestResolveAdaptiveSamplerConfig(t *testing.T) {
 		assert.Equal(t, 0.8, got.MatchThreshold)
 		assert.True(t, got.DetectionOnly)
 	})
+
+	t.Run("smart severity profiles enabled: unconfigured medium/high fall back to the base (low) values", func(t *testing.T) {
+		mockConfig.Set("logs_config.experimental_adaptive_sampling.smart_severity_profiles.enabled", true, pkgconfigmodel.SourceAgentRuntime)
+		defer mockConfig.Set("logs_config.experimental_adaptive_sampling.smart_severity_profiles.enabled", false, pkgconfigmodel.SourceAgentRuntime)
+
+		got := resolveAdaptiveSamplerConfig(nil, preprocessor.NewTokenizer(0))
+
+		assert.True(t, got.SmartSeverityProfilesEnabled)
+		low := preprocessor.SamplerProfile{RateLimit: got.RateLimit, BurstSize: got.BurstSize}
+		assert.Equal(t, low, got.Profiles[severityeventsdef.SeverityLow])
+		assert.Equal(t, low, got.Profiles[severityeventsdef.SeverityMedium])
+		assert.Equal(t, low, got.Profiles[severityeventsdef.SeverityHigh])
+	})
+
+	// Must run before any subtest that Sets (and later resets) medium.burst_size:
+	// mockConfig has no true "unset", so a prior Set — even one whose deferred
+	// cleanup restores the default value — permanently marks the key as
+	// configured for every subtest that follows.
+	t.Run("smart severity profiles enabled: high cascades a partially-configured medium field-by-field", func(t *testing.T) {
+		mockConfig.Set("logs_config.experimental_adaptive_sampling.smart_severity_profiles.enabled", true, pkgconfigmodel.SourceAgentRuntime)
+		mockConfig.Set("logs_config.experimental_adaptive_sampling.smart_severity_profiles.medium.rate_limit", 5.0, pkgconfigmodel.SourceAgentRuntime)
+		defer func() {
+			mockConfig.Set("logs_config.experimental_adaptive_sampling.smart_severity_profiles.enabled", false, pkgconfigmodel.SourceAgentRuntime)
+			mockConfig.Set("logs_config.experimental_adaptive_sampling.smart_severity_profiles.medium.rate_limit", 1.0, pkgconfigmodel.SourceAgentRuntime)
+		}()
+
+		got := resolveAdaptiveSamplerConfig(nil, preprocessor.NewTokenizer(0))
+
+		want := preprocessor.SamplerProfile{RateLimit: 5.0, BurstSize: got.BurstSize}
+		assert.Equal(t, want, got.Profiles[severityeventsdef.SeverityMedium])
+		assert.Equal(t, want, got.Profiles[severityeventsdef.SeverityHigh], "high's unconfigured rate_limit must cascade from medium, not revert to low, so high is never less permissive than medium")
+	})
+
+	t.Run("smart severity profiles enabled: unset high falls back to configured medium", func(t *testing.T) {
+		mockConfig.Set("logs_config.experimental_adaptive_sampling.smart_severity_profiles.enabled", true, pkgconfigmodel.SourceAgentRuntime)
+		mockConfig.Set("logs_config.experimental_adaptive_sampling.smart_severity_profiles.medium.rate_limit", 5.0, pkgconfigmodel.SourceAgentRuntime)
+		mockConfig.Set("logs_config.experimental_adaptive_sampling.smart_severity_profiles.medium.burst_size", 200.0, pkgconfigmodel.SourceAgentRuntime)
+		defer func() {
+			mockConfig.Set("logs_config.experimental_adaptive_sampling.smart_severity_profiles.enabled", false, pkgconfigmodel.SourceAgentRuntime)
+			mockConfig.Set("logs_config.experimental_adaptive_sampling.smart_severity_profiles.medium.rate_limit", 1.0, pkgconfigmodel.SourceAgentRuntime)
+			mockConfig.Set("logs_config.experimental_adaptive_sampling.smart_severity_profiles.medium.burst_size", 1000.0, pkgconfigmodel.SourceAgentRuntime)
+		}()
+
+		got := resolveAdaptiveSamplerConfig(nil, preprocessor.NewTokenizer(0))
+
+		assert.Equal(t, preprocessor.SamplerProfile{RateLimit: 5.0, BurstSize: 200.0}, got.Profiles[severityeventsdef.SeverityMedium])
+		assert.Equal(t, got.Profiles[severityeventsdef.SeverityMedium], got.Profiles[severityeventsdef.SeverityHigh])
+	})
+
+	t.Run("smart severity profiles enabled: explicit medium/high overrides win", func(t *testing.T) {
+		mockConfig.Set("logs_config.experimental_adaptive_sampling.smart_severity_profiles.enabled", true, pkgconfigmodel.SourceAgentRuntime)
+		mockConfig.Set("logs_config.experimental_adaptive_sampling.smart_severity_profiles.medium.rate_limit", 5.0, pkgconfigmodel.SourceAgentRuntime)
+		mockConfig.Set("logs_config.experimental_adaptive_sampling.smart_severity_profiles.medium.burst_size", 200.0, pkgconfigmodel.SourceAgentRuntime)
+		mockConfig.Set("logs_config.experimental_adaptive_sampling.smart_severity_profiles.high.rate_limit", 20.0, pkgconfigmodel.SourceAgentRuntime)
+		mockConfig.Set("logs_config.experimental_adaptive_sampling.smart_severity_profiles.high.burst_size", 500.0, pkgconfigmodel.SourceAgentRuntime)
+		defer func() {
+			mockConfig.Set("logs_config.experimental_adaptive_sampling.smart_severity_profiles.enabled", false, pkgconfigmodel.SourceAgentRuntime)
+			mockConfig.Set("logs_config.experimental_adaptive_sampling.smart_severity_profiles.medium.rate_limit", 1.0, pkgconfigmodel.SourceAgentRuntime)
+			mockConfig.Set("logs_config.experimental_adaptive_sampling.smart_severity_profiles.medium.burst_size", 1000.0, pkgconfigmodel.SourceAgentRuntime)
+			mockConfig.Set("logs_config.experimental_adaptive_sampling.smart_severity_profiles.high.rate_limit", 1.0, pkgconfigmodel.SourceAgentRuntime)
+			mockConfig.Set("logs_config.experimental_adaptive_sampling.smart_severity_profiles.high.burst_size", 1000.0, pkgconfigmodel.SourceAgentRuntime)
+		}()
+
+		got := resolveAdaptiveSamplerConfig(nil, preprocessor.NewTokenizer(0))
+
+		assert.Equal(t, preprocessor.SamplerProfile{RateLimit: 2.5, BurstSize: 50.0}, got.Profiles[severityeventsdef.SeverityLow])
+		assert.Equal(t, preprocessor.SamplerProfile{RateLimit: 5.0, BurstSize: 200.0}, got.Profiles[severityeventsdef.SeverityMedium])
+		assert.Equal(t, preprocessor.SamplerProfile{RateLimit: 20.0, BurstSize: 500.0}, got.Profiles[severityeventsdef.SeverityHigh])
+	})
+
+	t.Run("smart severity profiles enabled: a non-positive configured burst size is clamped to 1", func(t *testing.T) {
+		mockConfig.Set("logs_config.experimental_adaptive_sampling.smart_severity_profiles.enabled", true, pkgconfigmodel.SourceAgentRuntime)
+		mockConfig.Set("logs_config.experimental_adaptive_sampling.smart_severity_profiles.high.burst_size", 0.0, pkgconfigmodel.SourceAgentRuntime)
+		defer func() {
+			mockConfig.Set("logs_config.experimental_adaptive_sampling.smart_severity_profiles.enabled", false, pkgconfigmodel.SourceAgentRuntime)
+			mockConfig.Set("logs_config.experimental_adaptive_sampling.smart_severity_profiles.high.burst_size", 1000.0, pkgconfigmodel.SourceAgentRuntime)
+		}()
+
+		got := resolveAdaptiveSamplerConfig(nil, preprocessor.NewTokenizer(0))
+
+		assert.Equal(t, preprocessor.SamplerProfile{RateLimit: 1.0, BurstSize: 1.0}, got.Profiles[severityeventsdef.SeverityHigh])
+	})
+
+	t.Run("smart severity profiles enabled: a negative configured burst size is clamped to 1", func(t *testing.T) {
+		mockConfig.Set("logs_config.experimental_adaptive_sampling.smart_severity_profiles.enabled", true, pkgconfigmodel.SourceAgentRuntime)
+		mockConfig.Set("logs_config.experimental_adaptive_sampling.smart_severity_profiles.high.burst_size", -50.0, pkgconfigmodel.SourceAgentRuntime)
+		defer func() {
+			mockConfig.Set("logs_config.experimental_adaptive_sampling.smart_severity_profiles.enabled", false, pkgconfigmodel.SourceAgentRuntime)
+			mockConfig.Set("logs_config.experimental_adaptive_sampling.smart_severity_profiles.high.burst_size", 1000.0, pkgconfigmodel.SourceAgentRuntime)
+		}()
+
+		got := resolveAdaptiveSamplerConfig(nil, preprocessor.NewTokenizer(0))
+
+		assert.Equal(t, preprocessor.SamplerProfile{RateLimit: 1.0, BurstSize: 1.0}, got.Profiles[severityeventsdef.SeverityHigh])
+	})
 }
 
 func TestDecoderWithDockerJSONPartialLineDetectionOnlyMarksOversizedLogicalLineTruncated(t *testing.T) {

pkg/logs/internal/decoder/preprocessor/BUILD.bazel

@@ -36,6 +36,7 @@ go_library(
         "@com_github_datadog_datadog_agent_pkg_logs_util_testutils//:__subpackages__",
     ],
     deps = [
+        "//comp/anomalydetection/severityevents/def",
         "//comp/core/telemetry/def",
         "//comp/core/telemetry/impl",
         "//comp/logs-library/metrics",
@@ -63,6 +64,7 @@ dd_agent_go_test(
         "pattern_table_test.go",
         "preprocessor_test.go",
         "sampler_benchmark_test.go",
+        "sampler_smart_severity_profile_test.go",
         "sampler_test.go",
         "stack_trace_aggregator_test.go",
         "timestamp_detector_test.go",
@@ -73,6 +75,7 @@ dd_agent_go_test(
     ],
     embed = [":preprocessor"],
     deps = [
+        "//comp/anomalydetection/severityevents/def",
         "//comp/logs-library/metrics",
         "//comp/logs/agent/config",
         "//pkg/config/mock",

pkg/logs/internal/decoder/preprocessor/sampler.go

@@ -12,6 +12,7 @@ import (
 	"strconv"
 	"time"
 
+	severityeventsdef "github.com/DataDog/datadog-agent/comp/anomalydetection/severityevents/def"
 	telemetryimpl "github.com/DataDog/datadog-agent/comp/core/telemetry/impl"
 	"github.com/DataDog/datadog-agent/pkg/logs/message"
 )
@@ -117,6 +118,23 @@ type AdaptiveSamplerConfig struct {
 	// message through without rate-limiting. Using a closure lets the check track
 	// ReplaceableSource swaps and future Remote Config updates.
 	IsSourceDisabled func() bool
+	// SmartSeverityProfilesEnabled switches RateLimit/BurstSize based on the level
+	// read from SeverityProvider (see Profiles). Profiles[SeverityLow] must match
+	// RateLimit/BurstSize above.
+	SmartSeverityProfilesEnabled bool
+	// Profiles holds the RateLimit/BurstSize pair per SeverityLevel.
+	// Only consulted when SmartSeverityProfilesEnabled is true.
+	Profiles [severityeventsdef.NumSeverityLevels]SamplerProfile
+	// SeverityProvider returns the current anomaly-detection severity level, or false
+	// when no reader is registered yet. Only consulted when SmartSeverityProfilesEnabled
+	// is true. Left nil in tests that don't exercise smart severity profiles.
+	SeverityProvider func() (severityeventsdef.SeverityLevel, bool)
+}
+
+// SamplerProfile is a RateLimit/BurstSize pair for one SeverityLevel.
+type SamplerProfile struct {
+	RateLimit float64
+	BurstSize float64
 }
 
 // AdaptiveSamplerFilter matches messages by raw-content regex, structural sample,
@@ -148,6 +166,13 @@ type AdaptiveSampler struct {
 	source            string // used as a telemetry tag
 	now               func() time.Time
 	baseBytesEstimate int
+
+	// appliedLevel tracks the last severity profile applied by
+	// applyProfileIfChanged. appliedLevelInitialized stays false until a real
+	// reader is registered, so the sampler does not treat the no-reader case as
+	// an implicit Low profile.
+	appliedLevel            severityeventsdef.SeverityLevel
+	appliedLevelInitialized bool
 }
 
 // NewAdaptiveSampler creates a new AdaptiveSampler.
@@ -219,6 +244,42 @@ func (s *AdaptiveSampler) appendPatternHashTagIfEnabled(msg *message.Message, to
 	}
 }
 
+// applyProfileIfChanged switches RateLimit/BurstSize to the currently published
+// level, when SmartSeverityProfilesEnabled is set. Escalation grants every
+// pattern a fresh burst immediately. The first available Medium/High level is
+// also treated as an escalation from the base Low profile. De-escalation leaves
+// credits untouched, letting the refill-time clamp in processMatchedEntry
+// shrink them naturally.
+func (s *AdaptiveSampler) applyProfileIfChanged() {
+	if s.config.SeverityProvider == nil {
+		return
+	}
+	level, ok := s.config.SeverityProvider()
+	if !ok {
+		return
+	}
+	if s.appliedLevelInitialized && level == s.appliedLevel {
+		return
+	}
+
+	wasInitialized := s.appliedLevelInitialized
+	previousLevel := s.appliedLevel
+	profile := s.config.Profiles[level]
+	escalation := (wasInitialized && level > previousLevel) ||
+		(!wasInitialized && level > severityeventsdef.SeverityLow)
+	s.config.RateLimit = profile.RateLimit
+	s.config.BurstSize = profile.BurstSize
+
+	if escalation {
+		for i := range s.entries {
+			s.entries[i].credits = s.config.BurstSize
+		}
+	}
+
+	s.appliedLevel = level
+	s.appliedLevelInitialized = true
+}
+
 // Process applies credit-based rate limiting to the message.
 // Returns the message if allowed, nil if dropped.
 func (s *AdaptiveSampler) Process(msg *message.Message, tokens []Token) *message.Message {
@@ -227,6 +288,9 @@ func (s *AdaptiveSampler) Process(msg *message.Message, tokens []Token) *message
 	if !msg.HasContent() {
 		return msg
 	}
+	if s.config.SmartSeverityProfilesEnabled {
+		s.applyProfileIfChanged()
+	}
 	if s.config.IsSourceDisabled != nil && s.config.IsSourceDisabled() {
 		return msg
 	}