[add_session_metadata processor] Keep exited processes in the process DB

CHANGELOG.next.asciidoc

@@ -95,6 +95,7 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 
 *Auditbeat*
 - Set field types to correctly match ECS in sessionmd processor {issue}38955[38955] {pull}38994[38994]
+- Keep process info on exited processes, to avoid failing to enrich events in sessionmd processor {pull}39173[39173]
 
 *Filebeat*
 

x-pack/auditbeat/processors/sessionmd/add_session_metadata.go

@@ -57,7 +57,7 @@ func New(cfg *cfg.C) (beat.Processor, error) {
 	}
 
 	backfilledPIDs := db.ScrapeProcfs()
-	logger.Debugf("backfilled %d processes", len(backfilledPIDs))
+	logger.Infof("backfilled %d processes", len(backfilledPIDs))
 
 	var p provider.Provider
 
@@ -70,6 +70,9 @@ func New(cfg *cfg.C) (beat.Processor, error) {
 			if err != nil {
 				return nil, fmt.Errorf("failed to create provider: %w", err)
 			}
+			logger.Info("backend=auto using procfs")
+		} else {
+			logger.Info("backend=auto using ebpf")
 		}
 	case "ebpf":
 		p, err = ebpf_provider.NewProvider(ctx, logger, db)
@@ -111,6 +114,11 @@ func (p *addSessionMetadata) Run(ev *beat.Event) (*beat.Event, error) {
 	return result, nil
 }
 
+func (p *addSessionMetadata) Close() error {
+	p.db.Close()
+	return nil
+}
+
 func (p *addSessionMetadata) String() string {
 	return fmt.Sprintf("%v=[backend=%s, pid_field=%s]",
 		processorName, p.config.Backend, p.config.PIDField)

x-pack/auditbeat/processors/sessionmd/processdb/db.go

@@ -7,6 +7,7 @@
 package processdb
 
 import (
+	"container/heap"
 	"encoding/base64"
 	"errors"
 	"fmt"
@@ -82,6 +83,7 @@ type Process struct {
 	Cwd      string
 	Env      map[string]string
 	Filename string
+	ExitCode int32
 }
 
 var (
@@ -185,20 +187,26 @@ type DB struct {
 	entryLeaders             map[uint32]EntryType
 	entryLeaderRelationships map[uint32]uint32
 	procfs                   procfs.Reader
+	stopChan                 chan struct{}
+	removalCandidates        rcHeap
 }
 
 func NewDB(reader procfs.Reader, logger logp.Logger) (*DB, error) {
 	once.Do(initialize)
 	if initError != nil {
 		return &DB{}, initError
 	}
-	return &DB{
+	db := DB{
 		logger:                   logp.NewLogger("processdb"),
 		processes:                make(map[uint32]Process),
 		entryLeaders:             make(map[uint32]EntryType),
 		entryLeaderRelationships: make(map[uint32]uint32),
 		procfs:                   reader,
-	}, nil
+		stopChan:                 make(chan struct{}),
+		removalCandidates:        make(rcHeap, 0),
+	}
+	db.startReaper()
+	return &db, nil
 }
 
 func (db *DB) calculateEntityIDv1(pid uint32, startTime time.Time) string {
@@ -406,9 +414,18 @@ func (db *DB) InsertExit(exit types.ProcessExitEvent) {
 	defer db.mutex.Unlock()
 
 	pid := exit.PIDs.Tgid
-	delete(db.processes, pid)
-	delete(db.entryLeaders, pid)
-	delete(db.entryLeaderRelationships, pid)
+	process, ok := db.processes[pid]
+	if !ok {
+		db.logger.Errorf("could not insert exit, pid %v not found in db", pid)
+		return
+	}
+	process.ExitCode = exit.ExitCode
+	db.processes[pid] = process
+	heap.Push(&db.removalCandidates, removalCandidate{
+		pid:       pid,
+		startTime: process.PIDs.StartTimeNS,
+		exitTime:  time.Now(),
+	})
 }
 
 func interactiveFromTTY(tty types.TTYDev) bool {
@@ -437,6 +454,7 @@ func fullProcessFromDBProcess(p Process) types.Process {
 	ret.Thread.Capabilities.Effective, _ = capabilities.FromUint64(p.Creds.CapEffective)
 	ret.TTY.CharDevice.Major = p.CTTY.Major
 	ret.TTY.CharDevice.Minor = p.CTTY.Minor
+	ret.ExitCode = p.ExitCode
 
 	return ret
 }
@@ -716,3 +734,7 @@ func (db *DB) scrapeAncestors(proc Process) {
 		db.insertProcess(p)
 	}
 }
+
+func (db *DB) Close() {
+	close(db.stopChan)
+}