fix: scope event reducers by agent and gpid

Author: kylewanginchina

server/ingester/event/decoder/file_agg_reducer.go

@@ -29,7 +29,10 @@ func sameFileAggKey(a, b *dbwriter.FileAggEventStore) bool {
 	if a == nil || b == nil {
 		return false
 	}
-	return a.EventType == b.EventType &&
+	return a.VTAPID == b.VTAPID &&
+		a.RootPID == b.RootPID &&
+		a.GProcessID == b.GProcessID &&
+		a.EventType == b.EventType &&
 		a.ProcessKName == b.ProcessKName &&
 		a.AppInstance == b.AppInstance &&
 		a.FileDir == b.FileDir &&

server/ingester/event/decoder/file_agg_reducer_test.go

@@ -13,6 +13,9 @@ func makeRawFileEvent(processKName, eventType, fileName string, bytes uint32) *d
 	e.IsFileEvent = true
 	e.ProcessKName = processKName
 	e.EventType = eventType
+	e.VTAPID = 1
+	e.RootPID = 42
+	e.GProcessID = 100
 	e.FileDir = "/tmp/"
 	e.FileName = fileName
 	e.Bytes = bytes
@@ -78,3 +81,41 @@ func TestFileAggReducerFlushesOnKeyChange(t *testing.T) {
 		t.Fatalf("expected pending aggregate for codex-tui.log, got %+v", last)
 	}
 }
+
+func TestFileAggReducerFlushesOnDifferentGProcessID(t *testing.T) {
+	reducer := NewFileAggReducer()
+	first := makeRawFileEvent("codex", "write", "same.txt", 8)
+	second := makeRawFileEvent("codex", "write", "same.txt", 8)
+	second.GProcessID = 200
+
+	if flushed := reducer.Add(first); flushed != nil {
+		t.Fatalf("unexpected flush on first event")
+	}
+	flushed := reducer.Add(second)
+	if flushed == nil {
+		t.Fatalf("expected flush when gprocess_id changes")
+	}
+	if flushed.GProcessID != 100 {
+		t.Fatalf("flushed gprocess_id = %d, want 100", flushed.GProcessID)
+	}
+	flushed.Release()
+}
+
+func TestFileAggReducerFlushesOnDifferentAgentID(t *testing.T) {
+	reducer := NewFileAggReducer()
+	first := makeRawFileEvent("codex", "write", "same.txt", 8)
+	second := makeRawFileEvent("codex", "write", "same.txt", 8)
+	second.VTAPID = 2
+
+	if flushed := reducer.Add(first); flushed != nil {
+		t.Fatalf("unexpected flush on first event")
+	}
+	flushed := reducer.Add(second)
+	if flushed == nil {
+		t.Fatalf("expected flush when agent_id changes")
+	}
+	if flushed.VTAPID != 1 {
+		t.Fatalf("flushed agent_id = %d, want 1", flushed.VTAPID)
+	}
+	flushed.Release()
+}

server/ingester/event/decoder/file_mgmt_reducer.go

@@ -16,7 +16,7 @@ func NewFileMgmtReducer() *FileMgmtReducer {
 
 func (r *FileMgmtReducer) key(e *dbwriter.FileMgmtEventStore) string {
 	return fmt.Sprintf("%d|%d|%s|%s|%s|%s",
-		e.RootPID, e.GProcessID, e.MountSource, e.MountPoint, e.FileDir, e.FileName,
+		e.VTAPID, e.RootPID, e.MountSource, e.MountPoint, e.FileDir, e.FileName,
 	)
 }
 

server/ingester/event/decoder/file_mgmt_reducer_test.go

@@ -8,6 +8,7 @@ import (
 
 func makeFileMgmtEvent(processKName, eventType, fileName string) *dbwriter.FileMgmtEventStore {
 	e := dbwriter.AcquireFileMgmtEventStore()
+	e.VTAPID = 1
 	e.ProcessKName = processKName
 	e.EventType = eventType
 	e.FileDir = "/tmp/"
@@ -49,3 +50,22 @@ func TestFileMgmtReducerSuppressesDuplicateCreateUntilDelete(t *testing.T) {
 		out.Release()
 	}
 }
+
+func TestFileMgmtReducerKeepsDifferentAgentCreatesSeparate(t *testing.T) {
+	reducer := NewFileMgmtReducer()
+	first := makeFileMgmtEvent("bash", "create", "dup.txt")
+	second := makeFileMgmtEvent("bash", "create", "dup.txt")
+	second.VTAPID = 2
+
+	if out := reducer.Add(first); out == nil {
+		t.Fatalf("first create should pass through")
+	} else {
+		out.Release()
+	}
+
+	if out := reducer.Add(second); out == nil {
+		t.Fatalf("create from another agent should not be suppressed")
+	} else {
+		out.Release()
+	}
+}