Implement pattern eviction with LFU decay algorithm and memory-based eviction strategies

Author: DDuongNguyen

pkg/config/setup/config.go

@@ -1953,6 +1953,13 @@ func logsagent(config pkgconfigmodel.Setup) {
 	// Tag logs with their auto multiline detection label without aggregating them
 	config.BindEnvAndSetDefault("logs_config.auto_multi_line_detection_tagging", true)
 
+	// Log Pattern Eviction
+	config.BindEnvAndSetDefault("logs_config.patterns.max_pattern_count", 700)
+	config.BindEnvAndSetDefault("logs_config.patterns.max_memory_bytes", 4*1024*1024) // 4MB
+	config.BindEnvAndSetDefault("logs_config.patterns.eviction_high_watermark", 0.80) // Trigger eviction at 80% capacity
+	config.BindEnvAndSetDefault("logs_config.patterns.eviction_low_watermark", 0.70)  // Evict back to 70%
+	config.BindEnvAndSetDefault("logs_config.patterns.age_decay_factor", 0.5)
+
 	// Number of logs pipeline instances. Defaults to number of logical CPU cores as defined by GOMAXPROCS or 4, whichever is lower.
 	logsPipelines := min(4, runtime.GOMAXPROCS(0))
 	config.BindEnvAndSetDefault("logs_config.pipelines", logsPipelines)

pkg/logs/patterns/clustering/pattern_eviction.go

@@ -0,0 +1,155 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2016-present Datadog, Inc.
+
+// Package clustering provides clustering functionality for grouping similar TokenLists,
+// extracting patterns wildcard, and managing pattern lifecycle through eviction policies.
+package clustering
+
+import (
+	"time"
+
+	"github.com/DataDog/datadog-agent/pkg/logs/patterns/eviction"
+	"github.com/DataDog/datadog-agent/pkg/logs/patterns/token"
+	"github.com/DataDog/datadog-agent/pkg/util/log"
+)
+
+// EvictionPolicy is the policy for evicting patterns from the cluster manager.
+type EvictionPolicy int
+
+const (
+	// EvictionPolicyLFUDecay uses LFU with exponential age decay
+	EvictionPolicyLFUDecay EvictionPolicy = iota
+)
+
+// CollectEvictables collects all patterns as evictables (implements eviction.EvictableCollection).
+func (cm *ClusterManager) CollectEvictables() []eviction.Evictable {
+	cm.mu.RLock()
+	defer cm.mu.RUnlock()
+
+	evictables := make([]eviction.Evictable, 0, cm.patternCount)
+	for _, clusters := range cm.hashBuckets {
+		for _, cluster := range clusters {
+			for _, pattern := range cluster.Patterns {
+				evictables = append(evictables, pattern)
+			}
+		}
+	}
+	return evictables
+}
+
+// RemoveEvictable removes a specific pattern from the cluster manager (implements eviction.EvictableCollection).
+func (cm *ClusterManager) RemoveEvictable(item eviction.Evictable) {
+	pattern := item.(*Pattern)
+	cm.removePattern(pattern)
+}
+
+// removePattern removes a specific pattern from the cluster manager's hash buckets.
+func (cm *ClusterManager) removePattern(pattern *Pattern) {
+	var (
+		targetHash uint64
+		targetSig  *token.Signature
+	)
+
+	if pattern.Sample != nil && !pattern.Sample.IsEmpty() {
+		sig := token.NewSignature(pattern.Sample)
+		targetSig = &sig
+		targetHash = sig.GetHashBucket()
+	} else if pattern.Template != nil && !pattern.Template.IsEmpty() {
+		sig := token.NewSignature(pattern.Template)
+		targetSig = &sig
+		targetHash = sig.GetHashBucket()
+	}
+
+	cm.mu.Lock()
+	defer cm.mu.Unlock()
+
+	removeFromBucket := func(hash uint64, clusters []*Cluster, verifySignature bool) bool {
+		for ci := 0; ci < len(clusters); ci++ {
+			cluster := clusters[ci]
+			if verifySignature && targetSig != nil && !targetSig.Equals(cluster.Signature) {
+				continue
+			}
+
+			for pi := 0; pi < len(cluster.Patterns); pi++ {
+				if cluster.Patterns[pi] != pattern {
+					continue
+				}
+
+				// Update counters before removing pattern
+				cm.patternCount--
+				cm.estimatedBytes -= pattern.EstimatedBytes()
+
+				// Remove pattern with swap delete for speed
+				lastPattern := len(cluster.Patterns) - 1
+				cluster.Patterns[pi] = cluster.Patterns[lastPattern]
+				cluster.Patterns = cluster.Patterns[:lastPattern]
+				cluster.UpdatedAt = time.Now()
+
+				if len(cluster.Patterns) == 0 {
+					// Remove cluster from bucket
+					lastCluster := len(clusters) - 1
+					clusters[ci] = clusters[lastCluster]
+					clusters = clusters[:lastCluster]
+
+					if len(clusters) == 0 {
+						delete(cm.hashBuckets, hash)
+					} else {
+						cm.hashBuckets[hash] = clusters
+					}
+				} else {
+					cm.hashBuckets[hash] = clusters
+				}
+				return true
+			}
+		}
+		return false
+	}
+
+	// Try targeted bucket first
+	if targetSig != nil {
+		if clusters, ok := cm.hashBuckets[targetHash]; ok {
+			if removeFromBucket(targetHash, clusters, true) {
+				return
+			}
+		}
+	}
+
+	// Fallback: full scan if signature was missing or not found in target bucket
+	for hash, clusters := range cm.hashBuckets {
+		if removeFromBucket(hash, clusters, false) {
+			return
+		}
+	}
+
+	log.Warnf("Pattern %d not found during eviction, may have been already removed", pattern.PatternID)
+}
+
+// EvictLowestScoringPatterns evicts up to numToEvict patterns with the lowest eviction scores.
+// Returns the list of evicted patterns.
+func (cm *ClusterManager) EvictLowestScoringPatterns(numToEvict int, decayFactor float64) []*Pattern {
+	evictables := eviction.EvictLowestScoring(cm, numToEvict, decayFactor)
+	if len(evictables) == 0 {
+		return nil
+	}
+	patterns := make([]*Pattern, len(evictables))
+	for i, ev := range evictables {
+		patterns[i] = ev.(*Pattern)
+	}
+	return patterns
+}
+
+// EvictToMemoryTarget evicts patterns until the target memory is freed.
+// It uses actual pattern sizes rather than averages for precision.
+func (cm *ClusterManager) EvictToMemoryTarget(targetBytesToFree int64, decayFactor float64) []*Pattern {
+	evictables := eviction.EvictToMemoryTarget(cm, targetBytesToFree, decayFactor)
+	if len(evictables) == 0 {
+		return nil
+	}
+	patterns := make([]*Pattern, len(evictables))
+	for i, ev := range evictables {
+		patterns[i] = ev.(*Pattern)
+	}
+	return patterns
+}

pkg/logs/patterns/clustering/pattern_eviction_manager.go

@@ -0,0 +1,66 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2016-present Datadog, Inc.
+
+// Package clustering provides clustering functionality for grouping similar TokenLists,
+// extracting patterns wildcard, and managing pattern lifecycle through eviction policies.
+package clustering
+
+import (
+	pkgconfigsetup "github.com/DataDog/datadog-agent/pkg/config/setup"
+	"github.com/DataDog/datadog-agent/pkg/logs/patterns/eviction"
+	"github.com/DataDog/datadog-agent/pkg/util/log"
+)
+
+// EvictionManager handles pattern eviction using dual watermark system.
+// It wraps the generic eviction.Manager with clustering-specific configuration.
+type EvictionManager struct {
+	*eviction.Manager
+}
+
+// NewEvictionManager creates a new EvictionManager from config
+func NewEvictionManager() *EvictionManager {
+	cfg := pkgconfigsetup.Datadog()
+
+	return &EvictionManager{
+		Manager: &eviction.Manager{
+			MaxItemCount:          cfg.GetInt("logs_config.patterns.max_pattern_count"),
+			MaxMemoryBytes:        int64(cfg.GetInt("logs_config.patterns.max_memory_bytes")),
+			EvictionHighWatermark: cfg.GetFloat64("logs_config.patterns.eviction_high_watermark"),
+			EvictionLowWatermark:  cfg.GetFloat64("logs_config.patterns.eviction_low_watermark"),
+			AgeDecayFactor:        cfg.GetFloat64("logs_config.patterns.age_decay_factor"),
+		},
+	}
+}
+
+// Evict performs eviction on the cluster manager based on which threshold was exceeded
+func (em *EvictionManager) Evict(cm *ClusterManager, patternCount int, estimatedBytes int64, countOverLimit, bytesOverLimit bool) {
+	var evicted []*Pattern
+
+	numToEvict, bytesToFree, strategy := em.EvictionTargets(patternCount, estimatedBytes, countOverLimit, bytesOverLimit)
+
+	switch strategy {
+	// Memory-based eviction
+	case eviction.StrategyByBytes:
+		evicted = cm.EvictToMemoryTarget(bytesToFree, em.AgeDecayFactor)
+
+		highWatermarkBytes := int64(float64(em.MaxMemoryBytes) * em.EvictionHighWatermark)
+		targetBytes := int64(float64(em.MaxMemoryBytes) * em.EvictionLowWatermark)
+		log.Infof("Evicted %d patterns: memory %d bytes exceeded high watermark %d bytes (%.0f%% of %d max), now targeting %d bytes (%.0f%%)",
+			len(evicted), estimatedBytes, highWatermarkBytes,
+			em.EvictionHighWatermark*100, em.MaxMemoryBytes,
+			targetBytes, em.EvictionLowWatermark*100)
+
+	// Log count-based eviction
+	case eviction.StrategyByCount:
+		evicted = cm.EvictLowestScoringPatterns(numToEvict, em.AgeDecayFactor)
+
+		highWatermarkCount := int(float64(em.MaxItemCount) * em.EvictionHighWatermark)
+		targetCount := int(float64(em.MaxItemCount) * em.EvictionLowWatermark)
+		log.Infof("Evicted %d patterns: count %d exceeded high watermark %d (%.0f%% of %d max), now targeting %d (%.0f%%)",
+			len(evicted), patternCount, highWatermarkCount,
+			em.EvictionHighWatermark*100, em.MaxItemCount,
+			targetCount, em.EvictionLowWatermark*100)
+	}
+}