Implement pattern clustering with hash-based bucketing and signature-based grouping

Author: DDuongNguyen

pkg/logs/patterns/clustering/cluster.go

@@ -0,0 +1,164 @@
+// 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 (
+	"strings"
+	"time"
+
+	"github.com/DataDog/datadog-agent/pkg/logs/patterns/clustering/merging"
+	"github.com/DataDog/datadog-agent/pkg/logs/patterns/token"
+)
+
+// Cluster represents a cluster with a group of TokenLists that have identical signatures.
+// A cluster may contain multiple patterns if token lists with the same signature cannot be merged since structural Fidelity is Valuable.
+// Examples:
+// "Status: OK"     → HTTP response format
+// "Status; OK"     → CSV-like format
+// "Status OK"      → Plain text format
+// These are different log formats, even if semantically similar → we need to keep them separate.
+type Cluster struct {
+	Signature token.Signature
+	Patterns  []*Pattern // Multiple patterns per cluster
+
+	// Timestamp tracking for the cluster itself
+	CreatedAt time.Time // When cluster was first created
+	UpdatedAt time.Time // When cluster was last modified (any pattern changed)
+}
+
+// NewCluster creates a new cluster.
+func NewCluster(signature token.Signature) *Cluster {
+	now := time.Now()
+	return &Cluster{
+		Signature: signature,
+		Patterns:  nil, // Will be generated when needed
+		CreatedAt: now,
+		UpdatedAt: now,
+	}
+}
+
+// =============================================================================
+// Core Clustering Logic
+// =============================================================================
+
+// AddTokenListToPatterns adds a TokenList to the appropriate pattern in the cluster.
+// If no matching pattern exists, creates a new one.
+func (c *Cluster) AddTokenListToPatterns(tokenList *token.TokenList, cm *ClusterManager) *Pattern {
+	// Ensure patterns are generated
+	if len(c.Patterns) == 0 {
+		// No patterns yet, create first one
+		patternID := cm.generatePatternID()
+		pattern := newPattern(tokenList, patternID)
+
+		c.Patterns = []*Pattern{pattern}
+		// Update the cluster's new pattern at timestamp
+		c.UpdatedAt = time.Now()
+		return pattern
+	}
+
+	// Try to find a matching pattern
+	for _, p := range c.Patterns {
+		// Check if this TokenList can merge with this pattern's sample
+		if p.Sample != nil && merging.CanMergeTokenLists(tokenList, p.Sample) {
+			// CRITICAL: Also verify it can merge with the template
+			// If template has evolved differently, regeneratePattern will fail
+			// and we should create a new pattern instead
+			// Note: CanMergeTokenLists is not symmetric, so check both directions
+			if p.Template != nil {
+				templateCompatible1 := merging.CanMergeTokenLists(p.Template, tokenList)
+				templateCompatible2 := merging.CanMergeTokenLists(tokenList, p.Template)
+				templateCompatible := templateCompatible1 || templateCompatible2
+				if !templateCompatible {
+					// Log matches sample but not template - template has evolved incompatibly
+					// Skip this pattern and continue searching or create new one
+					// This will create a new pattern instead
+					continue
+				}
+			}
+
+			// Merge into existing pattern (same PatternID is preserved)
+			p.LogCount++
+			p.LastAccessAt = time.Now() // Update last access time for eviction
+			p.UpdatedAt = time.Now()
+			c.UpdatedAt = time.Now()
+
+			// Incrementally merge the new token list into the pattern template
+			// regeneratePattern will update template if merge succeeds
+			if c.regeneratePattern(p, tokenList) {
+				return p // Return existing pattern with updated template
+			}
+			// regeneratePattern failed - template couldn't merge with tokenList
+			// This shouldn't happen if we checked above, but handle it gracefully
+			// Create a new pattern instead
+			break
+		}
+	}
+
+	// No matching pattern found, create a new one
+	patternID := cm.generatePatternID()
+	pattern := newPattern(tokenList, patternID)
+	c.Patterns = append(c.Patterns, pattern)
+	c.UpdatedAt = time.Now()
+	return pattern
+}
+
+// regeneratePattern incrementally merges a new token list into the pattern.
+// Returns true if merge succeeded, false if merge failed.
+func (c *Cluster) regeneratePattern(p *Pattern, newTokenList *token.TokenList) bool {
+	if p.Template == nil {
+		return false
+	}
+
+	// Incremental merge: merge new log with existing template
+	merged := merging.MergeTokenLists(p.Template, newTokenList)
+	if merged == nil {
+		// Merge failed - template and newTokenList are incompatible
+		return false
+	}
+
+	p.Template = merged
+	p.Positions = make([]int, 0, merged.Length())
+
+	// Build wildcard positions list when 2 tokenlists are mergable.
+	for i := 0; i < merged.Length(); i++ {
+		tok := merged.Tokens[i]
+		if tok.Wildcard == token.IsWildcard {
+			p.Positions = append(p.Positions, i)
+
+			// Special handling for path wildcards
+			if tok.Type == token.TokenAbsolutePath && p.Sample != nil && i < p.Sample.Length() {
+				firstPath := p.Sample.Tokens[i].Value
+				merged.Tokens[i].Value = getPathPattern(firstPath)
+			}
+		}
+	}
+
+	p.UpdatedAt = time.Now()
+	return true
+}
+
+// getPathPattern converts a path to hierarchical wildcard pattern
+func getPathPattern(path string) string {
+	if path == "/" {
+		return "/"
+	}
+
+	// Remove leading/trailing slashes and split
+	trimmed := strings.Trim(path, "/")
+	if trimmed == "" {
+		return "/"
+	}
+
+	parts := strings.Split(trimmed, "/")
+	result := ""
+	for i := 0; i < len(parts); i++ {
+		result += "/*"
+	}
+
+	return result
+}

pkg/logs/patterns/clustering/cluster_manager.go

@@ -0,0 +1,173 @@
+// 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 (
+	"sync"
+
+	"github.com/DataDog/datadog-agent/pkg/logs/patterns/clustering/merging"
+	"github.com/DataDog/datadog-agent/pkg/logs/patterns/token"
+	"github.com/DataDog/datadog-agent/pkg/trace/log"
+)
+
+// PatternChangeType indicates what changed when adding a TokenList to the cluster manager
+type PatternChangeType int
+
+const (
+	// PatternNoChange means the TokenList was added to an existing cluster without structural changes
+	PatternNoChange PatternChangeType = iota
+	// PatternNew means a brand new pattern was created (first time seeing this signature)
+	PatternNew
+	// PatternUpdated means an existing pattern's structure changed (more wildcards added)
+	PatternUpdated
+)
+
+// ClusterManager manages the clustering of TokenLists using hash-based bucketing.
+type ClusterManager struct {
+	mu          sync.RWMutex
+	hashBuckets map[uint64][]*Cluster
+	nextID      uint64
+
+	// patternCount tracks the total number of patterns across all clusters.
+	// This is maintained incrementally to avoid O(N) scans on every Add().
+	patternCount int
+
+	// estimatedBytes tracks an approximate memory footprint of patterns stored in this manager.
+	// This is an estimate (not exact Go heap usage) and is intended for threshold-based eviction triggers.
+	estimatedBytes int64
+}
+
+// NewClusterManager creates a new ClusterManager.
+func NewClusterManager() *ClusterManager {
+	return &ClusterManager{
+		hashBuckets: make(map[uint64][]*Cluster),
+		nextID:      1,
+	}
+}
+
+// PatternCount returns the total number of patterns currently stored.
+func (cm *ClusterManager) PatternCount() int {
+	cm.mu.RLock()
+	defer cm.mu.RUnlock()
+	return cm.patternCount
+}
+
+// EstimatedBytes returns the approximate memory footprint (in bytes) of patterns currently stored.
+func (cm *ClusterManager) EstimatedBytes() int64 {
+	cm.mu.RLock()
+	defer cm.mu.RUnlock()
+	return cm.estimatedBytes
+}
+
+// Add processes a TokenList and adds it to the appropriate cluster.
+// Returns:
+// - pattern: the pattern that was created/updated
+// - changeType: what changed (new/updated/no change)
+// - patternCount: total patterns after this addition
+// - estimatedBytes: total estimated memory after this addition
+func (cm *ClusterManager) Add(tokenList *token.TokenList) (*Pattern, PatternChangeType, int, int64) {
+	if tokenList == nil || tokenList.IsEmpty() {
+		log.Errorf("Cluster Manager failed to add log: %v for patterning. Token list is empty or nil.", tokenList.String())
+		return nil, PatternNoChange, 0, 0
+	}
+
+	// Lock the cluster manager to prevent concurrent access to the hash buckets. Current implementation is single-threaded on local pipeline, but we will eventually build a shared cluster manager across multiple pipelines.
+	// todo: implement a shared cluster manager across multiple pipelines
+	cm.mu.Lock()
+	defer cm.mu.Unlock()
+
+	// Create new signature and hash it
+	signature := token.NewSignature(tokenList)
+	hash := signature.Hash
+
+	// Get hash bucket
+	clusters := cm.hashBuckets[hash]
+
+	// Look for existing cluster with matching signature
+	for _, cluster := range clusters {
+		if !cluster.Signature.Equals(signature) {
+			continue
+		}
+
+		// Find which pattern within the cluster the tokenList will match
+		var matchedPattern *Pattern
+		var oldWildcardCount int
+		oldPatternCount := len(cluster.Patterns)
+		var oldMatchedBytes int64
+		for _, p := range cluster.Patterns {
+			if p.Sample != nil && merging.CanMergeTokenLists(tokenList, p.Sample) {
+				matchedPattern = p
+				oldWildcardCount = p.GetWildcardCount()
+				oldMatchedBytes = p.EstimatedBytes()
+				break
+			}
+		}
+
+		// Add the tokenList to the cluster (merges or creates new pattern)
+		pattern := cluster.AddTokenListToPatterns(tokenList, cm)
+
+		// Update counters (pattern count + estimated bytes)
+		// If a new pattern was appended to this cluster, increment counts.
+		if len(cluster.Patterns) > oldPatternCount {
+			cm.patternCount++
+			cm.estimatedBytes += pattern.EstimatedBytes()
+		} else if matchedPattern != nil && matchedPattern.PatternID == pattern.PatternID {
+			// Existing pattern updated; adjust bytes based on template evolution.
+			cm.estimatedBytes += pattern.EstimatedBytes() - oldMatchedBytes
+		}
+
+		// Check if a new pattern was created (no match found or merge failed)
+		if matchedPattern == nil || matchedPattern.PatternID != pattern.PatternID {
+			return pattern, PatternNew, cm.patternCount, cm.estimatedBytes
+		}
+
+		// Check if wildcard count changed (pattern evolved)
+		if pattern.GetWildcardCount() != oldWildcardCount {
+			return pattern, PatternUpdated, cm.patternCount, cm.estimatedBytes
+		}
+
+		return pattern, PatternNoChange, cm.patternCount, cm.estimatedBytes
+	}
+
+	// If no matching pattern was found, create a new cluster and pattern.
+	newCluster := NewCluster(signature)
+	// Add the token list to create the first pattern
+	pattern := newCluster.AddTokenListToPatterns(tokenList, cm)
+	cm.hashBuckets[hash] = append(clusters, newCluster)
+
+	// New cluster always creates exactly one new pattern
+	cm.patternCount++
+	cm.estimatedBytes += pattern.EstimatedBytes()
+
+	return pattern, PatternNew, cm.patternCount, cm.estimatedBytes
+}
+
+// Clear removes all clusters.
+func (cm *ClusterManager) Clear() {
+	cm.mu.Lock()
+	defer cm.mu.Unlock()
+	cm.hashBuckets = make(map[uint64][]*Cluster)
+	cm.patternCount = 0
+	cm.estimatedBytes = 0
+}
+
+// GetStats returns the current pattern count and estimated memory usage.
+// This is a read-only operation that acquires a read lock for thread safety.
+func (cm *ClusterManager) GetStats() (patternCount int, estimatedBytes int64) {
+	cm.mu.RLock()
+	defer cm.mu.RUnlock()
+	return cm.patternCount, cm.estimatedBytes
+}
+
+// generatePatternID generates a unique pattern ID using a monotonic counter.
+// Must be called while holding the ClusterManager lock.
+func (cm *ClusterManager) generatePatternID() uint64 {
+	id := cm.nextID
+	cm.nextID++
+	return id
+}