cherry: procmon memory fix (#503, #507)

go.mod

@@ -4,7 +4,7 @@ go 1.27.3
 
 require (
 	github.com/coreos/go-iptables v0.6.0
-	github.com/elastic/go-libaudit/v2 v2.3.2
+	github.com/elastic/go-libaudit/v2 v2.6.2
 	github.com/florianl/go-nflog/v2 v2.0.1
 	github.com/google/gopacket v1.1.19
 	github.com/jarcoal/httpmock v1.3.0
@@ -23,6 +23,7 @@ require (
 	github.com/docker/distribution v2.8.1+incompatible // indirect
 	github.com/docker/go-connections v0.4.0 // indirect
 	github.com/docker/go-units v0.5.0 // indirect
+	github.com/elastic/go-licenser v0.4.1 // indirect
 	github.com/go-ini/ini v1.67.0 // indirect
 	github.com/go-logr/logr v1.4.3 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect

go.sum

@@ -45,6 +45,10 @@ github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkp
 github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
 github.com/elastic/go-libaudit/v2 v2.3.2 h1:qWNcA3nkwNEGh1UBDbDTVF55KR6SM1W2Ji1LGDqFEpw=
 github.com/elastic/go-libaudit/v2 v2.3.2/go.mod h1:+ZE0czqmbqtnRkl0fNgpI+HvVVRo/ZMJdcXv/PaKcOo=
+github.com/elastic/go-libaudit/v2 v2.6.2 h1:1PM6wVBTJHJQYsKl8jfA9/Aw9pFty5uUezPiUfKtOI4=
+github.com/elastic/go-libaudit/v2 v2.6.2/go.mod h1:8205nkf2oSrXFlO4H5j8/cyVMoSF3Y7jt+FjgS4ubQU=
+github.com/elastic/go-licenser v0.4.1 h1:1xDURsc8pL5zYT9R29425J3vkHdt4RT5TNEMeRN48x4=
+github.com/elastic/go-licenser v0.4.1/go.mod h1:V56wHMpmdURfibNBggaSBfqgPxyT1Tldns1i87iTEvU=
 github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg=
 github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U=
 github.com/florianl/go-nflog/v2 v2.0.1 h1:8csWIqFQ2vDaZJkxezY3dXDB7bEFg0VRFsYd2Bzj3II=
@@ -210,6 +214,7 @@ github.com/yashtewari/glob-intersection v0.2.0 h1:8iuHdN88yYuCzCdjt0gDe+6bAhUwBe
 github.com/yashtewari/glob-intersection v0.2.0/go.mod h1:LK7pIC3piUjovexikBbJ26Yml7g8xa5bsjfx2v1fwok=
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
+github.com/yuin/goldmark v1.4.0/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
 go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64=
 go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.63.0 h1:RbKq8BG0FI8OiXhBfcRtqqHcZcka+gU3cskNuf05R18=
@@ -248,9 +253,12 @@ golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPh
 golang.org/x/crypto v0.50.0 h1:zO47/JPrL6vsNkINmLoo/PH1gcxpls50DNogFvB5ZGI=
 golang.org/x/crypto v0.50.0/go.mod h1:3muZ7vA7PBCE6xgPX7nkzzjiUq87kRItoJQM1Yo8S+Q=
 golang.org/x/lint v0.0.0-20200302205851-738671d3881b/go.mod h1:3xt1FjdF8hUf6vQPIChWIBhFzV8gjjsPE/fR3IyQdNY=
+golang.org/x/lint v0.0.0-20210508222113-6edffad5e616/go.mod h1:3xt1FjdF8hUf6vQPIChWIBhFzV8gjjsPE/fR3IyQdNY=
 golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg=
 golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
+golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
+golang.org/x/mod v0.5.1/go.mod h1:5OXOZSfqPIIbmVBIIKWRFfZjPR0E5r58TLhUjH0a2Ro=
 golang.org/x/mod v0.35.0 h1:Ww1D637e6Pg+Zb2KrWfHQUnH2dQRLBQyAtpr/haaJeM=
 golang.org/x/mod v0.35.0/go.mod h1:+GwiRhIInF8wPm+4AoT6L0FA1QWAad3OMdTRx4tFYlU=
 golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
@@ -268,11 +276,13 @@ golang.org/x/net v0.0.0-20201224014010-6772e930b67b/go.mod h1:m0MpNAwzfU5UDzcl9v
 golang.org/x/net v0.0.0-20210119194325-5f4716e94777/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20210525063256-abc453219eb5/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
+golang.org/x/net v0.0.0-20210805182204-aaa1db679c0d/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.53.0 h1:d+qAbo5L0orcWAr0a9JweQpjXF19LMXJE8Ey7hwOdUA=
 golang.org/x/net v0.53.0/go.mod h1:JvMuJH7rrdiCfbeHoo3fCQU24Lf5JJwT9W3sJFulfgs=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4=
 golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -295,6 +305,8 @@ golang.org/x/sys v0.0.0-20210216163648-f7da38b97c65/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20210305230114-8fe3ee5dd75b/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210525143221-35b2ab0089ea/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20210809222454-d867a43fc93e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20211102192858-4dd72447c267/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.43.0 h1:Rlag2XtaFTxp19wS8MXlJwTvoh8ArU6ezoyFsMyCTNI=
 golang.org/x/sys v0.43.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
@@ -311,6 +323,7 @@ golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtn
 golang.org/x/tools v0.0.0-20200130002326-2f3ba24bd6e7/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
 golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
 golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
+golang.org/x/tools v0.1.7/go.mod h1:LGqMHiF4EqQNHR1JncWGqT5BVaXmza+X+BDGol+dOxo=
 golang.org/x/tools v0.44.0 h1:UP4ajHPIcuMjT1GqzDWRlalUEoY+uzoZKnhOjbIPD2c=
 golang.org/x/tools v0.44.0/go.mod h1:KA0AfVErSdxRZIsOVipbv3rQhVXTnlU6UhKxHd1seDI=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=

procmon.go

@@ -1,10 +1,17 @@
 package main
 
 import (
+	"errors"
 	"fmt"
+	"math/bits"
 	"strconv"
 	"sync"
+	"sync/atomic"
+	"syscall"
 	"time"
+
+	"github.com/elastic/go-libaudit/v2"
+	"github.com/elastic/go-libaudit/v2/auparse"
 )
 
 const (
@@ -13,15 +20,31 @@ const (
 	processMonitorTag = "procmon"
 )
 
+// Audit records are grouped into events by go-libaudit's reassembler, with the
+// same limits Auditbeat uses by default: at most 50 events in flight, and an
+// event without its last record is given up after 2 seconds.
+const (
+	reassemblerMaxInFlight = 50
+	reassemblerTimeout     = 2 * time.Second
+	reassemblerMaintain    = 500 * time.Millisecond
+)
+
 type ProcessMonitor struct {
 	CorrelationId         string
 	Repo                  string
 	ApiClient             *ApiClient
 	DNSProxy              *DNSProxy
 	WorkingDirectory      string
 	DisableFileMonitoring bool
-	Events                map[int]*Event
-	mutex                 sync.RWMutex
+
+	// dispatchEvent replaces the event handler in tests
+	dispatchEvent func(*Event)
+
+	sequenceGaps    atomic.Int64
+	receiveOverflow atomic.Int64
+	receiveErrors   atomic.Int64
+	invalidRecords  atomic.Int64
+	recordErrors    atomic.Int64
 }
 
 type Process struct {
@@ -36,61 +59,54 @@ type Process struct {
 }
 
 type Event struct {
-	FileName          string
-	Path              string
-	Syscall           string
-	Exe               string
-	IPAddress         string
-	Port              string
-	Pid               string
-	ProcessArguments  []string
-	PPid              string
-	Euid              string
-	Timestamp         time.Time
-	EventType         string
-	Status            string
-	SentForProcessing bool
+	FileName         string
+	Path             string
+	Syscall          string
+	Exe              string
+	IPAddress        string
+	Port             string
+	Pid              string
+	ProcessArguments []string
+	PPid             string
+	Euid             string
+	Timestamp        time.Time
+	EventType        string
+	Status           string
 }
 
-func (p *ProcessMonitor) PrepareEvent(sequence int, eventMap map[string]interface{}) {
-	p.mutex.Lock()
-	_, found := p.Events[sequence]
-	if !found {
-		p.Events[sequence] = &Event{}
-	}
-
+func (event *Event) addRecord(eventMap map[string]interface{}) {
 	// Tags
 	value, found := eventMap["tags"]
 
 	if found {
 		tags := value.([]string)
 		for _, tag := range tags {
-			p.Events[sequence].EventType = tag
-			p.Events[sequence].Syscall = getValue("syscall", eventMap)
-			p.Events[sequence].Exe = getValue("exe", eventMap)
-			p.Events[sequence].Pid = getValue("pid", eventMap)
-			p.Events[sequence].PPid = getValue("ppid", eventMap)
-			p.Events[sequence].Euid = getValue("euid", eventMap)
+			event.EventType = tag
+			event.Syscall = getValue("syscall", eventMap)
+			event.Exe = getValue("exe", eventMap)
+			event.Pid = getValue("pid", eventMap)
+			event.PPid = getValue("ppid", eventMap)
+			event.Euid = getValue("euid", eventMap)
 			timestamp, err := time.Parse("2006-01-02 15:04:05.999999999 +0000 UTC", getValue("@timestamp", eventMap))
 			if err != nil {
 				timestamp = time.Now().UTC()
 			}
-			p.Events[sequence].Timestamp = timestamp
-			p.Events[sequence].Status = getValue("result", eventMap)
+			event.Timestamp = timestamp
+			event.Status = getValue("result", eventMap)
 		}
 	}
 
 	// Current working directory
 	_, found = eventMap["cwd"]
 	if found {
-		p.Events[sequence].Path = getValue("cwd", eventMap)
+		event.Path = getValue("cwd", eventMap)
 	}
 
 	// Connect
 	_, found = eventMap["addr"]
 	if found {
-		p.Events[sequence].IPAddress = getValue("addr", eventMap)
-		p.Events[sequence].Port = getValue("port", eventMap)
+		event.IPAddress = getValue("addr", eventMap)
+		event.Port = getValue("port", eventMap)
 	}
 
 	// Process start
@@ -103,7 +119,7 @@ func (p *ProcessMonitor) PrepareEvent(sequence int, eventMap map[string]interfac
 			WriteLog(fmt.Sprintf("could not parse argc:%v", argc))
 		}
 		for i := 0; i < argCount; i++ {
-			p.Events[sequence].ProcessArguments = append(p.Events[sequence].ProcessArguments, fmt.Sprintf("%v", eventMap[fmt.Sprintf("a%d", i)]))
+			event.ProcessArguments = append(event.ProcessArguments, fmt.Sprintf("%v", eventMap[fmt.Sprintf("a%d", i)]))
 		}
 	}
 
@@ -112,12 +128,11 @@ func (p *ProcessMonitor) PrepareEvent(sequence int, eventMap map[string]interfac
 
 	if found {
 		nameTypeStr := fmt.Sprintf("%v", nameType)
-		if nameTypeStr == "DELETE" || nameTypeStr == "CREATE" || nameTypeStr == "NORMAL" {
-			p.Events[sequence].FileName = getValue("name", eventMap)
+		// the first qualifying PATH wins, as when events were dispatched on it
+		if (nameTypeStr == "DELETE" || nameTypeStr == "CREATE" || nameTypeStr == "NORMAL") && event.FileName == "" {
+			event.FileName = getValue("name", eventMap)
 		}
 	}
-
-	p.mutex.Unlock()
 }
 
 func getValue(key string, eventMap map[string]interface{}) string {
@@ -129,17 +144,120 @@ func getValue(key string, eventMap map[string]interface{}) string {
 	return ""
 }
 
-func (p *ProcessMonitor) markEventSent(event *Event) {
-	p.mutex.Lock()
-	event.SentForProcessing = true
-	p.mutex.Unlock()
+// auditStream receives complete events from the reassembler and dispatches
+// the ones the event handler uses. Every event is delivered once, including
+// incomplete ones that are evicted, so nothing is kept after delivery.
+type auditStream struct {
+	monitor  *ProcessMonitor
+	dispatch func(*Event)
+
+	// ops serializes push with maintain and flush, so an event cannot be
+	// given up between a record being admitted and being pushed
+	ops sync.Mutex
+
+	// syscallEvents holds the sequence numbers of the events in the
+	// reassembler, which all start with a SYSCALL record (see push)
+	mu            sync.Mutex
+	syscallEvents map[uint32]struct{}
 }
 
-func isEventReady(event *Event) bool {
-	if event.SentForProcessing {
+// push passes a record to the reassembler if it belongs to an audited
+// syscall. The records the agent uses (EXECVE, SOCKADDR, CWD, PATH,
+// PROCTITLE) follow the event's SYSCALL record. Other records, such as BPF or
+// ANOM_PROMISCUOUS, can arrive alone without the records that complete an
+// event, and since the reassembler delivers events in sequence order, each
+// one would hold back every later event until it times out.
+func (s *auditStream) push(r *libaudit.Reassembler, msg *auparse.AuditMessage) {
+	s.ops.Lock()
+	defer s.ops.Unlock()
+
+	s.mu.Lock()
+	if s.syscallEvents == nil {
+		s.syscallEvents = make(map[uint32]struct{})
+	}
+	_, found := s.syscallEvents[msg.Sequence]
+	if !found && msg.RecordType == auparse.AUDIT_SYSCALL {
+		s.syscallEvents[msg.Sequence] = struct{}{}
+		found = true
+	}
+	s.mu.Unlock()
+
+	if found {
+		r.PushMessage(msg)
+	}
+}
+
+// maintain gives up on events whose last record did not arrive in time.
+func (s *auditStream) maintain(r *libaudit.Reassembler) error {
+	s.ops.Lock()
+	defer s.ops.Unlock()
+	return r.Maintain()
+}
+
+// flush delivers the events still in the reassembler and closes it.
+func (s *auditStream) flush(r *libaudit.Reassembler) {
+	s.ops.Lock()
+	defer s.ops.Unlock()
+	r.Close()
+}
+
+func (s *auditStream) ReassemblyComplete(msgs []*auparse.AuditMessage) {
+	if len(msgs) == 0 {
+		return
+	}
+	// records that arrive after this, such as the EOE after a PROCTITLE,
+	// are not passed on
+	s.mu.Lock()
+	delete(s.syscallEvents, msgs[0].Sequence)
+	s.mu.Unlock()
+
+	event := &Event{}
+	for _, msg := range msgs {
+		eventMap := msg.ToMapStr()
+		if parseErr, found := eventMap["error"]; found {
+			noteLoss(&s.monitor.recordErrors, 1, fmt.Sprintf("audit record fields not parsed: %v", parseErr))
+		}
+		event.addRecord(eventMap)
+	}
+	if isEventReady(event) {
+		s.dispatch(event)
+	}
+}
+
+// EventsLost is called for gaps in the audit sequence numbers. A gap is not
+// always a loss, since events this agent skips use sequence numbers too.
+func (s *auditStream) EventsLost(count int) {
+	// go-libaudit v2.3.2 underflows when a record arrives after its event was
+	// given up, which shows up as a huge count
+	if count <= 0 || count > 1<<24 {
+		return
+	}
+	noteLoss(&s.monitor.sequenceGaps, int64(count), "gaps in audit sequence numbers")
+}
+
+// noteReceiveError counts a failed receive. A full netlink socket (ENOBUFS)
+// means the kernel dropped messages; the socket stays usable either way.
+func (p *ProcessMonitor) noteReceiveError(err error) {
+	if errors.Is(err, syscall.ENOBUFS) {
+		noteLoss(&p.receiveOverflow, 1, "audit netlink buffer overflowed, kernel dropped messages")
+		return
+	}
+	noteLoss(&p.receiveErrors, 1, fmt.Sprintf("audit receive error: %v", err))
+}
+
+// noteLoss adds n to the counter and logs when the total reaches the next
+// power of two, so a sustained loss does not flood the log. It reports whether
+// it logged.
+func noteLoss(counter *atomic.Int64, n int64, what string) bool {
+	total := counter.Add(n)
+	if bits.Len64(uint64(total)) == bits.Len64(uint64(total-n)) {
 		return false
 	}
+	WriteLog(fmt.Sprintf("[procmon] %s, total=%d", what, total))
+	return true
+}
 
+func isEventReady(event *Event) bool {
 	switch event.EventType {
 	case netMonitorTag:
 		if event.IPAddress != "" && event.Port != "" {