From b3f7c96343c01433179b8a03aef9f1c228d1784e Mon Sep 17 00:00:00 2001 From: Jason Hall Date: Wed, 14 Jan 2026 10:44:04 -0500 Subject: [PATCH] Add memory-bounded deduplication with LRU cache Implements configurable memory limits for the file deduplication cache to prevent unbounded memory growth in long-running production deployments. - Add LRU cache implementation with configurable max size - Add -max-unique-files flag (default: 0 = unbounded) - Add snoop_events_evicted_total metric for monitoring - Track and log cache evictions with delta warnings - Add comprehensive tests for bounded and unbounded behavior When max-unique-files is set, the cache evicts least-recently-used entries to stay within the limit. Evicted files may be re-recorded if accessed again, making this suitable for workloads with temporal locality in file access patterns. Closes milestone 3 task: Memory-bounded deduplication Signed-off-by: Jason Hall --- CHECKLIST.md | 198 ------------------------------- cmd/snoop/main.go | 25 +++- pkg/metrics/metrics.go | 6 + pkg/processor/lru.go | 80 +++++++++++++ pkg/processor/lru_test.go | 204 ++++++++++++++++++++++++++++++++ pkg/processor/processor.go | 34 +++--- pkg/processor/processor_test.go | 137 +++++++++++++++++++-- plan.md | 5 +- 8 files changed, 458 insertions(+), 231 deletions(-) delete mode 100644 CHECKLIST.md create mode 100644 pkg/processor/lru.go create mode 100644 pkg/processor/lru_test.go diff --git a/CHECKLIST.md b/CHECKLIST.md deleted file mode 100644 index 83c910a..0000000 --- a/CHECKLIST.md +++ /dev/null @@ -1,198 +0,0 @@ -# Milestone 1 Completion Checklist - -## Code Implementation ✅ - -- [x] eBPF C program (`pkg/ebpf/bpf/snoop.c`) - - [x] Tracepoint for `sys_enter_openat` - - [x] Tracepoint for `sys_enter_execve` - - [x] Ring buffer for events (256KB) - - [x] Per-CPU heap for event building - - [x] Traced cgroups hash map - - [x] Cgroup filtering logic - - [x] Event structure (cgroup ID, PID, syscall, path) - -- [x] Go eBPF loader (`pkg/ebpf/probe.go`) - - [x] Load eBPF objects with cilium/ebpf - - [x] Attach tracepoints - - [x] Manage traced cgroups - - [x] Ring buffer reader - - [x] Event parsing - - [x] Cleanup on close - -- [x] Cgroup utilities (`pkg/cgroup/discovery.go`) - - [x] Get self cgroup ID - - [x] Get cgroup ID by path - - [x] Discovery interface - - [x] Self-excluding discovery stub - -- [x] Main application (`cmd/snoop/main.go`) - - [x] CLI flag parsing - - [x] Signal handling (SIGINT, SIGTERM) - - [x] Probe lifecycle management - - [x] Event display loop - - [x] Graceful shutdown - -## Build Infrastructure ✅ - -- [x] Go module (`go.mod`) - - [x] cilium/ebpf dependency - - [x] golang.org/x/sys dependency - -- [x] Dockerfile - - [x] Multi-stage build - - [x] clang, llvm, libbpf-dev - - [x] eBPF code generation - - [x] Go build - - [x] Minimal runtime image - -- [x] Makefile - - [x] `vmlinux` target (generate vmlinux.h) - - [x] `generate` target (eBPF code gen) - - [x] `build` target - - [x] `test` target - - [x] `docker-build` target - - [x] `clean` target - - [x] Help text - -- [x] ko configuration (`.ko.yaml`) - - [x] Build settings - - [x] Base image - -- [x] `.gitignore` - - [x] Binary artifacts - - [x] Generated eBPF files - - [x] IDE files - -## Testing Infrastructure ✅ - -- [x] Docker Compose (`deploy/docker-compose.yaml`) - - [x] Test app service - - [x] Snoop sidecar configuration - - [x] Volume mounts for cgroup/tracefs - - [x] Privileged mode - -- [x] Helper scripts - - [x] `scripts/find-cgroup.sh` - Find container cgroups - - [x] Executable permissions - -- [x] Unit tests - - [x] `pkg/cgroup/discovery_test.go` - -- [x] CI/CD (`.github/workflows/build.yaml`) - - [x] Ubuntu runner - - [x] Go setup - - [x] Dependency installation - - [x] vmlinux.h generation - - [x] eBPF code generation - - [x] Build step - - [x] Test step - - [x] go vet - - [x] staticcheck - -## Documentation ✅ - -- [x] `README.md` - - [x] Project overview - - [x] Current status - - [x] Requirements - - [x] Build instructions - - [x] Testing overview - - [x] Project structure - -- [x] `QUICKSTART.md` - - [x] Prerequisites - - [x] Generate kernel headers - - [x] Build steps - - [x] Run example - - [x] Docker build option - -- [x] `TESTING.md` - - [x] Prerequisites - - [x] Setup instructions - - [x] Test 1: Basic tracing - - [x] Test 2: Multiple syscalls - - [x] Test 3: Docker Compose - - [x] Troubleshooting section - - [x] Success criteria - -- [x] `STATUS.md` - - [x] What's been done - - [x] What's left - - [x] Project structure - - [x] Design decisions - - [x] Known limitations - - [x] Next milestone preview - -- [x] `CHECKLIST.md` (this file) - -- [x] `plan.md` updates - - [x] Current status banner - - [x] Milestone 1 completion status - - [x] Files created list - - [x] Current status description - -## Milestone 2 Progress ⏳ - -- [x] Additional syscalls (statx, newfstatat, faccessat, faccessat2, readlinkat, openat2, execveat) -- [ ] Path normalization logic -- [ ] Deduplication data structures -- [ ] JSON report output -- [ ] Prometheus metrics -- [ ] Path exclusion filtering -- [ ] Configuration via environment variables - -## Testing Required (Needs Linux) ⚠️ - -Cannot be completed on macOS, requires Linux system: - -- [ ] Generate vmlinux.h from kernel -- [ ] Generate eBPF Go bindings -- [ ] Build snoop binary -- [ ] Load eBPF program (verify no errors) -- [ ] Trace test container -- [ ] Verify events captured -- [ ] Verify cgroup filtering works -- [ ] Test graceful shutdown -- [ ] Verify no kernel panics - -## How to Mark Milestone 1 Complete - -Run the following on a Linux system (kernel 5.4+): - -```bash -# 1. Setup -make vmlinux -make generate -make build - -# 2. Run test -docker run -d --name test alpine sh -c "while true; do cat /etc/passwd > /dev/null; sleep 2; done" -CGROUP=$(./scripts/find-cgroup.sh test | grep "Cgroup Path:" | cut -d: -f2 | xargs) -sudo ./snoop -cgroup "$CGROUP" - -# 3. Expected output -# Should see: -# [PID XXXX] [Cgroup YYYY] [Syscall 257] /etc/passwd -# (Repeating every 2 seconds) - -# 4. Test filtering -# Snoop's own file accesses should NOT appear -# Only the test container's accesses - -# 5. Test shutdown -# Press Ctrl+C - should exit cleanly - -# 6. Cleanup -docker stop test -docker rm test -``` - -If all tests pass, Milestone 1 is complete! ✅ - -## Statistics - -- **Lines of Code**: ~450 lines (Go + C) -- **Files Created**: 20+ -- **Documentation**: 5 markdown files -- **Time to Complete**: ~1 session -- **Dependencies**: 2 (cilium/ebpf, golang.org/x/sys) diff --git a/cmd/snoop/main.go b/cmd/snoop/main.go index fe7c8fe..99497da 100644 --- a/cmd/snoop/main.go +++ b/cmd/snoop/main.go @@ -30,6 +30,7 @@ func main() { containerID string metricsAddr string logLevel string + maxUniqueFiles int ) flag.StringVar(&cgroupPath, "cgroup", "", "Cgroup path to trace (e.g., /system.slice/docker-abc123.scope)") @@ -40,17 +41,18 @@ func main() { flag.StringVar(&containerID, "container-id", "", "Container ID for report metadata") flag.StringVar(&metricsAddr, "metrics-addr", ":9090", "Address for Prometheus metrics endpoint (empty to disable)") flag.StringVar(&logLevel, "log-level", "info", "Log level (debug, info, warn, error)") + flag.IntVar(&maxUniqueFiles, "max-unique-files", 0, "Maximum unique files to track (0 = unbounded)") flag.Parse() // Initialize logging context ctx := clog.WithLogger(context.Background(), clog.New(clog.ParseLevel(logLevel))) - if err := run(ctx, cgroupPath, reportPath, reportInterval, excludePaths, imageRef, containerID, metricsAddr); err != nil { + if err := run(ctx, cgroupPath, reportPath, reportInterval, excludePaths, imageRef, containerID, metricsAddr, maxUniqueFiles); err != nil { clog.FromContext(ctx).Fatalf("Fatal error: %v", err) } } -func run(ctx context.Context, cgroupPath, reportPath string, reportInterval time.Duration, excludePaths, imageRef, containerID, metricsAddr string) error { +func run(ctx context.Context, cgroupPath, reportPath string, reportInterval time.Duration, excludePaths, imageRef, containerID, metricsAddr string, maxUniqueFiles int) error { log := clog.FromContext(ctx) ctx, cancel := context.WithCancel(ctx) defer cancel() @@ -121,14 +123,15 @@ func run(ctx context.Context, cgroupPath, reportPath string, reportInterval time } // Create processor and reporter - proc := processor.NewProcessor(ctx, exclusions) + proc := processor.NewProcessor(ctx, exclusions, maxUniqueFiles) rep := reporter.NewFileReporter(ctx, reportPath) startedAt := time.Now() log.Infof("Writing reports to: %s (interval: %s)", reportPath, reportInterval) - // Track last seen drops count for computing delta + // Track last seen drops and evictions count for computing deltas var lastDrops uint64 + var lastEvicted uint64 // Start periodic report writer reportTicker := time.NewTicker(reportInterval) @@ -152,6 +155,16 @@ func run(ctx context.Context, cgroupPath, reportPath string, reportInterval time lastDrops = drops } + // Update the evictions counter metric with the delta + if stats.EventsEvicted > lastEvicted { + delta := stats.EventsEvicted - lastEvicted + m.EventsEvicted.Add(float64(delta)) + if delta > 0 { + log.Warnf("Deduplication cache eviction: %d file paths evicted since last report", delta) + } + lastEvicted = stats.EventsEvicted + } + report := &reporter.Report{ ContainerID: containerID, ImageRef: imageRef, @@ -164,8 +177,8 @@ func run(ctx context.Context, cgroupPath, reportPath string, reportInterval time log.Errorf("Error writing report: %v", err) m.ReportWriteErrors.Inc() } else { - log.Infof("Report written: %d unique files, %d events processed, %d dropped", - stats.UniqueFiles, stats.EventsProcessed, drops) + log.Infof("Report written: %d unique files, %d events processed, %d dropped, %d evicted", + stats.UniqueFiles, stats.EventsProcessed, drops, stats.EventsEvicted) m.ReportWrites.Inc() } // Update gauge for unique files count diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go index a39945f..da1e0bc 100644 --- a/pkg/metrics/metrics.go +++ b/pkg/metrics/metrics.go @@ -15,6 +15,7 @@ type Metrics struct { EventsExcluded prometheus.Counter EventsDuplicate prometheus.Counter EventsDropped prometheus.Counter + EventsEvicted prometheus.Counter UniqueFiles prometheus.Gauge ReportWrites prometheus.Counter @@ -48,6 +49,10 @@ func New() *Metrics { Name: "snoop_events_dropped_total", Help: "Total number of events dropped due to ring buffer overflow.", }), + EventsEvicted: prometheus.NewCounter(prometheus.CounterOpts{ + Name: "snoop_events_evicted_total", + Help: "Total number of file paths evicted from deduplication cache due to memory limits.", + }), UniqueFiles: prometheus.NewGauge(prometheus.GaugeOpts{ Name: "snoop_unique_files", Help: "Current number of unique files recorded.", @@ -70,6 +75,7 @@ func New() *Metrics { m.EventsExcluded, m.EventsDuplicate, m.EventsDropped, + m.EventsEvicted, m.UniqueFiles, m.ReportWrites, m.ReportWriteErrors, diff --git a/pkg/processor/lru.go b/pkg/processor/lru.go new file mode 100644 index 0000000..05dc8d5 --- /dev/null +++ b/pkg/processor/lru.go @@ -0,0 +1,80 @@ +package processor + +import "container/list" + +// lruCache implements a simple Least Recently Used cache for string deduplication. +// It maintains a doubly-linked list for LRU ordering and a map for O(1) lookups. +type lruCache struct { + maxSize int + items map[string]*list.Element + order *list.List + evicted uint64 +} + +// newLRUCache creates a new LRU cache with the given maximum size. +// If maxSize is 0 or negative, the cache is unbounded. +func newLRUCache(maxSize int) *lruCache { + return &lruCache{ + maxSize: maxSize, + items: make(map[string]*list.Element), + order: list.New(), + } +} + +// add adds a key to the cache. Returns true if the key was already present. +// If the cache is at capacity, the least recently used item is evicted. +func (c *lruCache) add(key string) bool { + // Check if key already exists + if elem, exists := c.items[key]; exists { + // Move to front (most recently used) + c.order.MoveToFront(elem) + return true + } + + // Add new key + elem := c.order.PushFront(key) + c.items[key] = elem + + // Evict if over capacity (only if maxSize > 0) + if c.maxSize > 0 && c.order.Len() > c.maxSize { + c.evictOldest() + } + + return false +} + +// evictOldest removes the least recently used item from the cache. +func (c *lruCache) evictOldest() { + elem := c.order.Back() + if elem != nil { + c.order.Remove(elem) + delete(c.items, elem.Value.(string)) + c.evicted++ + } +} + +// len returns the current number of items in the cache. +func (c *lruCache) len() int { + return len(c.items) +} + +// evictions returns the total number of evictions that have occurred. +func (c *lruCache) evictions() uint64 { + return c.evicted +} + +// keys returns all keys currently in the cache (unsorted). +func (c *lruCache) keys() []string { + keys := make([]string, 0, len(c.items)) + for key := range c.items { + keys = append(keys, key) + } + return keys +} + +// reset clears all items from the cache. +func (c *lruCache) reset() { + c.items = make(map[string]*list.Element) + c.order = list.New() + c.evicted = 0 +} diff --git a/pkg/processor/lru_test.go b/pkg/processor/lru_test.go new file mode 100644 index 0000000..0eda5ce --- /dev/null +++ b/pkg/processor/lru_test.go @@ -0,0 +1,204 @@ +package processor + +import "testing" + +func TestLRUCache_Basic(t *testing.T) { + cache := newLRUCache(3) + + // Add first item + if exists := cache.add("a"); exists { + t.Error("expected 'a' to be new") + } + if cache.len() != 1 { + t.Errorf("len = %d, want 1", cache.len()) + } + + // Add same item again + if exists := cache.add("a"); !exists { + t.Error("expected 'a' to exist") + } + if cache.len() != 1 { + t.Errorf("len = %d, want 1", cache.len()) + } + + // Add more items + cache.add("b") + cache.add("c") + if cache.len() != 3 { + t.Errorf("len = %d, want 3", cache.len()) + } +} + +func TestLRUCache_Eviction(t *testing.T) { + cache := newLRUCache(3) + + // Fill cache + cache.add("a") + cache.add("b") + cache.add("c") + + if cache.len() != 3 { + t.Fatalf("len = %d, want 3", cache.len()) + } + if cache.evictions() != 0 { + t.Errorf("evictions = %d, want 0", cache.evictions()) + } + + // Add fourth item, should evict 'a' (oldest) + cache.add("d") + if cache.len() != 3 { + t.Errorf("len = %d, want 3 after eviction", cache.len()) + } + if cache.evictions() != 1 { + t.Errorf("evictions = %d, want 1", cache.evictions()) + } + + // 'a' should no longer exist + if exists := cache.add("a"); exists { + t.Error("expected 'a' to be evicted and treated as new") + } + if cache.evictions() != 2 { + t.Errorf("evictions = %d, want 2", cache.evictions()) + } + + // 'b' should have been evicted by adding 'a' + if exists := cache.add("b"); exists { + t.Error("expected 'b' to be evicted") + } +} + +func TestLRUCache_LRUOrdering(t *testing.T) { + cache := newLRUCache(3) + + // Add a, b, c (in that order, so 'a' is oldest) + cache.add("a") + cache.add("b") + cache.add("c") + + // Access 'a' to make it most recent + if exists := cache.add("a"); !exists { + t.Fatal("expected 'a' to exist") + } + + // Now add 'd', which should evict 'b' (now oldest) + cache.add("d") + + // 'a' should still exist (was refreshed) + if exists := cache.add("a"); !exists { + t.Error("expected 'a' to still exist after refresh") + } + + // 'c' should still exist + if exists := cache.add("c"); !exists { + t.Error("expected 'c' to exist") + } + + // 'd' should still exist + if exists := cache.add("d"); !exists { + t.Error("expected 'd' to exist") + } + + // 'b' should have been evicted + if exists := cache.add("b"); exists { + t.Error("expected 'b' to be evicted") + } + + // Now cache should contain: b, d, c (a was evicted when b was added) + // Verify 'a' was evicted + if exists := cache.add("a"); exists { + t.Error("expected 'a' to have been evicted after adding 'b'") + } +} + +func TestLRUCache_Unbounded(t *testing.T) { + // maxSize = 0 means unbounded + cache := newLRUCache(0) + + // Add many items + for i := 0; i < 1000; i++ { + cache.add(string(rune('a' + (i % 26)))) + } + + // Should have 26 unique items (a-z) + if cache.len() != 26 { + t.Errorf("len = %d, want 26", cache.len()) + } + + // No evictions should occur + if cache.evictions() != 0 { + t.Errorf("evictions = %d, want 0 (unbounded)", cache.evictions()) + } +} + +func TestLRUCache_NegativeSize(t *testing.T) { + // Negative maxSize should also be unbounded + cache := newLRUCache(-1) + + for i := 0; i < 100; i++ { + cache.add(string(rune('0' + i))) + } + + if cache.len() != 100 { + t.Errorf("len = %d, want 100", cache.len()) + } + if cache.evictions() != 0 { + t.Errorf("evictions = %d, want 0", cache.evictions()) + } +} + +func TestLRUCache_Keys(t *testing.T) { + cache := newLRUCache(5) + + items := []string{"foo", "bar", "baz"} + for _, item := range items { + cache.add(item) + } + + keys := cache.keys() + if len(keys) != len(items) { + t.Errorf("keys length = %d, want %d", len(keys), len(items)) + } + + // Check all items are present + keySet := make(map[string]bool) + for _, key := range keys { + keySet[key] = true + } + + for _, item := range items { + if !keySet[item] { + t.Errorf("expected key %q in keys", item) + } + } +} + +func TestLRUCache_Reset(t *testing.T) { + cache := newLRUCache(3) + + cache.add("a") + cache.add("b") + cache.add("c") + cache.add("d") // Causes eviction + + if cache.len() != 3 { + t.Fatalf("len = %d, want 3", cache.len()) + } + if cache.evictions() != 1 { + t.Fatalf("evictions = %d, want 1", cache.evictions()) + } + + cache.reset() + + if cache.len() != 0 { + t.Errorf("len after reset = %d, want 0", cache.len()) + } + if cache.evictions() != 0 { + t.Errorf("evictions after reset = %d, want 0", cache.evictions()) + } + + // Should be able to add items again + cache.add("x") + if cache.len() != 1 { + t.Errorf("len after adding to reset cache = %d, want 1", cache.len()) + } +} diff --git a/pkg/processor/processor.go b/pkg/processor/processor.go index 95216d8..a004fb8 100644 --- a/pkg/processor/processor.go +++ b/pkg/processor/processor.go @@ -20,7 +20,7 @@ type Event struct { // exclusion filtering, and deduplication. type Processor struct { ctx context.Context - seen map[string]struct{} + seen *lruCache seenMu sync.RWMutex excluded []string @@ -29,12 +29,14 @@ type Processor struct { eventsProcessed uint64 eventsExcluded uint64 eventsDuplicate uint64 + eventsEvicted uint64 mu sync.Mutex } // NewProcessor creates a new event processor with the given exclusion prefixes. // If excludePrefixes is nil, DefaultExclusions() will be used. -func NewProcessor(ctx context.Context, excludePrefixes []string) *Processor { +// maxUniqueFiles limits the deduplication cache size (0 = unbounded). +func NewProcessor(ctx context.Context, excludePrefixes []string, maxUniqueFiles int) *Processor { log := clog.FromContext(ctx) if excludePrefixes == nil { excludePrefixes = DefaultExclusions() @@ -43,9 +45,14 @@ func NewProcessor(ctx context.Context, excludePrefixes []string) *Processor { for _, prefix := range excludePrefixes { log.Debugf("Excluding paths with prefix: %s", prefix) } + if maxUniqueFiles > 0 { + log.Infof("Deduplication cache limited to %d unique files", maxUniqueFiles) + } else { + log.Info("Deduplication cache is unbounded") + } return &Processor{ ctx: ctx, - seen: make(map[string]struct{}), + seen: newLRUCache(maxUniqueFiles), excluded: excludePrefixes, } } @@ -87,10 +94,7 @@ func (p *Processor) Process(event *Event) (string, ProcessResult) { // Check for duplicates and add if new p.seenMu.Lock() - _, exists := p.seen[normalized] - if !exists { - p.seen[normalized] = struct{}{} - } + exists := p.seen.add(normalized) p.seenMu.Unlock() if exists { @@ -112,18 +116,14 @@ func (p *Processor) Files() []string { p.seenMu.RLock() defer p.seenMu.RUnlock() - files := make([]string, 0, len(p.seen)) - for path := range p.seen { - files = append(files, path) - } - return files + return p.seen.keys() } // UniqueFileCount returns the number of unique files seen. func (p *Processor) UniqueFileCount() int { p.seenMu.RLock() defer p.seenMu.RUnlock() - return len(p.seen) + return p.seen.len() } // Stats returns processing statistics. @@ -132,6 +132,7 @@ type Stats struct { EventsProcessed uint64 EventsExcluded uint64 EventsDuplicate uint64 + EventsEvicted uint64 UniqueFiles int } @@ -141,7 +142,8 @@ func (p *Processor) Stats() Stats { defer p.mu.Unlock() p.seenMu.RLock() - uniqueFiles := len(p.seen) + uniqueFiles := p.seen.len() + evicted := p.seen.evictions() p.seenMu.RUnlock() return Stats{ @@ -149,6 +151,7 @@ func (p *Processor) Stats() Stats { EventsProcessed: p.eventsProcessed, EventsExcluded: p.eventsExcluded, EventsDuplicate: p.eventsDuplicate, + EventsEvicted: evicted, UniqueFiles: uniqueFiles, } } @@ -157,7 +160,7 @@ func (p *Processor) Stats() Stats { // This is primarily useful for testing. func (p *Processor) Reset() { p.seenMu.Lock() - p.seen = make(map[string]struct{}) + p.seen.reset() p.seenMu.Unlock() p.mu.Lock() @@ -165,5 +168,6 @@ func (p *Processor) Reset() { p.eventsProcessed = 0 p.eventsExcluded = 0 p.eventsDuplicate = 0 + p.eventsEvicted = 0 p.mu.Unlock() } diff --git a/pkg/processor/processor_test.go b/pkg/processor/processor_test.go index 9666b06..e1e94e8 100644 --- a/pkg/processor/processor_test.go +++ b/pkg/processor/processor_test.go @@ -2,6 +2,7 @@ package processor import ( "context" + "fmt" "sort" "sync" "testing" @@ -11,7 +12,7 @@ func TestNewProcessor(t *testing.T) { ctx := context.Background() t.Run("with nil exclusions uses defaults", func(t *testing.T) { - p := NewProcessor(ctx, nil) + p := NewProcessor(ctx, nil, 0) if len(p.excluded) != len(DefaultExclusions()) { t.Errorf("expected default exclusions, got %v", p.excluded) } @@ -19,11 +20,18 @@ func TestNewProcessor(t *testing.T) { t.Run("with custom exclusions", func(t *testing.T) { exclusions := []string{"/tmp/", "/var/"} - p := NewProcessor(ctx, exclusions) + p := NewProcessor(ctx, exclusions, 0) if len(p.excluded) != 2 { t.Errorf("expected 2 exclusions, got %d", len(p.excluded)) } }) + + t.Run("with bounded cache", func(t *testing.T) { + p := NewProcessor(ctx, nil, 100) + if p.seen.maxSize != 100 { + t.Errorf("expected maxSize=100, got %d", p.seen.maxSize) + } + }) } func TestProcessorProcess(t *testing.T) { @@ -65,7 +73,7 @@ func TestProcessorProcess(t *testing.T) { }} { t.Run(tt.desc, func(t *testing.T) { ctx := context.Background() - p := NewProcessor(ctx, nil) + p := NewProcessor(ctx, nil, 0) event := &Event{ PID: 1234, Path: tt.path, @@ -84,7 +92,7 @@ func TestProcessorProcess(t *testing.T) { func TestProcessorDeduplication(t *testing.T) { ctx := context.Background() - p := NewProcessor(ctx, nil) + p := NewProcessor(ctx, nil, 0) // First access should be new event := &Event{PID: 1234, Path: "/etc/passwd"} @@ -115,7 +123,7 @@ func TestProcessorDeduplication(t *testing.T) { func TestProcessorDeduplicationNormalized(t *testing.T) { ctx := context.Background() - p := NewProcessor(ctx, nil) + p := NewProcessor(ctx, nil, 0) // Access with dots event1 := &Event{PID: 1234, Path: "/etc/./passwd"} @@ -139,7 +147,7 @@ func TestProcessorDeduplicationNormalized(t *testing.T) { func TestProcessorFiles(t *testing.T) { ctx := context.Background() - p := NewProcessor(ctx, nil) + p := NewProcessor(ctx, nil, 0) paths := []string{"/etc/passwd", "/usr/bin/bash", "/lib/libc.so.6"} for _, path := range paths { @@ -163,7 +171,7 @@ func TestProcessorFiles(t *testing.T) { func TestProcessorStats(t *testing.T) { ctx := context.Background() - p := NewProcessor(ctx, nil) + p := NewProcessor(ctx, nil, 0) // Process various events p.Process(&Event{PID: 1234, Path: "/etc/passwd"}) // new @@ -193,7 +201,7 @@ func TestProcessorStats(t *testing.T) { func TestProcessorReset(t *testing.T) { ctx := context.Background() - p := NewProcessor(ctx, nil) + p := NewProcessor(ctx, nil, 0) p.Process(&Event{PID: 1234, Path: "/etc/passwd"}) p.Process(&Event{PID: 1234, Path: "/etc/hostname"}) @@ -216,7 +224,7 @@ func TestProcessorReset(t *testing.T) { func TestProcessorConcurrency(t *testing.T) { ctx := context.Background() - p := NewProcessor(ctx, nil) + p := NewProcessor(ctx, nil, 0) var wg sync.WaitGroup // Simulate concurrent access from multiple goroutines @@ -258,7 +266,7 @@ func TestProcessorConcurrency(t *testing.T) { func TestProcessorCustomExclusions(t *testing.T) { ctx := context.Background() // Test with custom exclusions that don't include defaults - p := NewProcessor(ctx, []string{"/tmp/", "/custom/"}) + p := NewProcessor(ctx, []string{"/tmp/", "/custom/"}, 0) // Default exclusions should NOT apply _, result := p.Process(&Event{PID: 1234, Path: "/proc/self/status"}) @@ -277,3 +285,112 @@ func TestProcessorCustomExclusions(t *testing.T) { t.Errorf("/custom path: got %v, want ResultExcluded", result) } } + +func TestProcessorBoundedCache(t *testing.T) { + ctx := context.Background() + // Create processor with max 3 unique files + p := NewProcessor(ctx, []string{}, 3) + + // Add 3 files - should all be new + for i := 1; i <= 3; i++ { + path := fmt.Sprintf("/file%d", i) + _, result := p.Process(&Event{PID: 1234, Path: path}) + if result != ResultNew { + t.Errorf("file %d: got %v, want ResultNew", i, result) + } + } + + stats := p.Stats() + if stats.UniqueFiles != 3 { + t.Errorf("unique files = %d, want 3", stats.UniqueFiles) + } + if stats.EventsEvicted != 0 { + t.Errorf("evicted = %d, want 0", stats.EventsEvicted) + } + + // Add 4th file - should evict oldest (file1) + _, result := p.Process(&Event{PID: 1234, Path: "/file4"}) + if result != ResultNew { + t.Errorf("file4: got %v, want ResultNew", result) + } + + stats = p.Stats() + if stats.UniqueFiles != 3 { + t.Errorf("unique files after eviction = %d, want 3", stats.UniqueFiles) + } + if stats.EventsEvicted != 1 { + t.Errorf("evicted = %d, want 1", stats.EventsEvicted) + } + + // file1 should now be treated as new (was evicted) + _, result = p.Process(&Event{PID: 1234, Path: "/file1"}) + if result != ResultNew { + t.Errorf("evicted file1: got %v, want ResultNew", result) + } + + stats = p.Stats() + if stats.EventsEvicted != 2 { + t.Errorf("evicted after re-add = %d, want 2", stats.EventsEvicted) + } +} + +func TestProcessorBoundedCacheWithHighLoad(t *testing.T) { + ctx := context.Background() + // Create processor with max 10 unique files + p := NewProcessor(ctx, []string{}, 10) + + // Add 100 unique files + for i := 0; i < 100; i++ { + path := fmt.Sprintf("/file%d", i) + p.Process(&Event{PID: 1234, Path: path}) + } + + stats := p.Stats() + // Should only retain 10 files + if stats.UniqueFiles != 10 { + t.Errorf("unique files = %d, want 10", stats.UniqueFiles) + } + // Should have evicted 90 files + if stats.EventsEvicted != 90 { + t.Errorf("evicted = %d, want 90", stats.EventsEvicted) + } + // 100 events received, all processed + if stats.EventsProcessed != 100 { + t.Errorf("processed = %d, want 100", stats.EventsProcessed) + } +} + +func TestProcessorUnboundedVsBounded(t *testing.T) { + ctx := context.Background() + + // Unbounded processor + pUnbounded := NewProcessor(ctx, []string{}, 0) + // Bounded processor + pBounded := NewProcessor(ctx, []string{}, 5) + + // Add 20 unique files to both + for i := 0; i < 20; i++ { + path := fmt.Sprintf("/file%d", i) + pUnbounded.Process(&Event{PID: 1234, Path: path}) + pBounded.Process(&Event{PID: 1234, Path: path}) + } + + unboundedStats := pUnbounded.Stats() + boundedStats := pBounded.Stats() + + // Unbounded should have all 20 files + if unboundedStats.UniqueFiles != 20 { + t.Errorf("unbounded unique files = %d, want 20", unboundedStats.UniqueFiles) + } + if unboundedStats.EventsEvicted != 0 { + t.Errorf("unbounded evicted = %d, want 0", unboundedStats.EventsEvicted) + } + + // Bounded should have only 5 files + if boundedStats.UniqueFiles != 5 { + t.Errorf("bounded unique files = %d, want 5", boundedStats.UniqueFiles) + } + if boundedStats.EventsEvicted != 15 { + t.Errorf("bounded evicted = %d, want 15", boundedStats.EventsEvicted) + } +} diff --git a/plan.md b/plan.md index 137864c..0f26a57 100644 --- a/plan.md +++ b/plan.md @@ -8,8 +8,9 @@ Milestone 3 progress: - ✅ Prometheus metrics endpoint (`/metrics` on port 9090) - ✅ Structured logging with clog - ✅ Ring buffer overflow handling and metrics +- ✅ Memory-bounded deduplication with LRU cache -**Next**: Memory-bounded deduplication +**Next**: Health check endpoint See [Milestone 3](#milestone-3-production-hardening) for remaining tasks. @@ -364,7 +365,7 @@ Volume mounts: - [x] Prometheus metrics endpoint - [x] Structured logging with levels (clog) - [x] Ring buffer overflow handling and metrics -- [ ] Memory-bounded deduplication (LRU or bloom filter for extreme cases) +- [x] Memory-bounded deduplication (LRU cache with configurable max size) - [ ] Health check endpoint - [ ] Configuration validation - [ ] Resource limit recommendations documented