Stop audited syscalls from waiting on a full audit queue

procmon.go

@@ -27,6 +27,15 @@ const (
 	reassemblerMaintain    = 500 * time.Millisecond
 )
 
+// Kernel audit queue settings, from Auditbeat's defaults. The kernel's own
+// defaults (64 records, 60 second wait) make audited syscalls sleep whenever
+// the agent falls behind, which slows the job instead of dropping records.
+const (
+	auditBacklogLimit       = 8192
+	auditFallbackRateLimit  = 5000 // only when the wait time cannot be set to 0
+	kernelLostCheckInterval = 15 * time.Second
+)
+
 type ProcessMonitor struct {
 	CorrelationId         string
 	Repo                  string
@@ -43,6 +52,7 @@ type ProcessMonitor struct {
 	receiveErrors   atomic.Int64
 	invalidRecords  atomic.Int64
 	recordErrors    atomic.Int64
+	kernelLost      atomic.Int64
 }
 
 type Process struct {

procmon_linux.go

@@ -38,6 +38,13 @@ func (p *ProcessMonitor) MonitorProcesses(errc chan error) {
 		errc <- errors.Wrap(err, "failed to get audit client status")
 	}
 
+	if err = configureBackpressure(client, status); err != nil {
+		WriteLog(fmt.Sprintf("[procmon] failed to configure audit backlog: %v", err))
+	}
+	stopLostCheck := make(chan struct{})
+	defer close(stopLostCheck)
+	go p.monitorKernelLost(status.Lost, stopLostCheck)
+
 	if status.Enabled == 0 {
 		if err = client.SetEnabled(true, libaudit.WaitForReply); err != nil {
 			errc <- errors.Wrap(err, "failed to set audit client")
@@ -159,6 +166,103 @@ func (p *ProcessMonitor) MonitorProcesses(errc chan error) {
 	}
 }
 
+// auditQueueControl is the part of the audit client that sets the kernel queue.
+type auditQueueControl interface {
+	SetFailure(fm libaudit.FailureMode, wm libaudit.WaitMode) error
+	SetBacklogLimit(limit uint32, wm libaudit.WaitMode) error
+	SetBacklogWaitTime(waitTime int32, wm libaudit.WaitMode) error
+	SetRateLimit(perSecondLimit uint32, wm libaudit.WaitMode) error
+}
+
+// configureBackpressure follows Auditbeat's default "auto" strategy: raise the
+// backlog limit and have the kernel drop records immediately when the queue is
+// full, instead of making the audited syscall wait. Kernels that cannot set the
+// wait time get a rate limit instead. The failure mode is set to silent first,
+// as Auditbeat does, so a dropped record never panics the kernel.
+func configureBackpressure(c auditQueueControl, status *libaudit.AuditStatus) error {
+	if status.Enabled == 2 { // locked, changes are rejected
+		return nil
+	}
+
+	if status.Failure != uint32(libaudit.SilentOnFailure) {
+		if err := c.SetFailure(libaudit.SilentOnFailure, libaudit.WaitForReply); err != nil {
+			return errors.Wrap(err, "set failure mode")
+		}
+	}
+
+	if status.BacklogLimit != auditBacklogLimit {
+		if err := c.SetBacklogLimit(auditBacklogLimit, libaudit.WaitForReply); err != nil {
+			return errors.Wrap(err, "set backlog limit")
+		}
+	}
+
+	var rateLimit uint32
+	if status.FeatureBitmap&libaudit.AuditFeatureBitmapBacklogWaitTime != 0 {
+		if err := c.SetBacklogWaitTime(0, libaudit.WaitForReply); err != nil {
+			return errors.Wrap(err, "set backlog wait time")
+		}
+	} else {
+		rateLimit = auditFallbackRateLimit
+	}
+
+	if status.RateLimit != rateLimit {
+		if err := c.SetRateLimit(rateLimit, libaudit.WaitForReply); err != nil {
+			return errors.Wrap(err, "set rate limit")
+		}
+	}
+
+	WriteLog(fmt.Sprintf("[procmon] audit backlog limit %d, wait time 0: %t, rate limit %d", auditBacklogLimit, rateLimit == 0, rateLimit))
+	return nil
+}
+
+// monitorKernelLost polls the kernel's lost record counter on its own
+// connection, as Auditbeat does, so dropped records show up in the log. It
+// stops when done is closed.
+func (p *ProcessMonitor) monitorKernelLost(lost uint32, done <-chan struct{}) {
+	var client *libaudit.AuditClient
+	defer func() {
+		if client != nil {
+			client.Close()
+		}
+	}()
+	ticker := time.NewTicker(kernelLostCheckInterval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-done:
+			return
+		case <-ticker.C:
+		}
+		if client == nil {
+			c, err := libaudit.NewAuditClient(nil)
+			if err != nil {
+				continue
+			}
+			client = c
+		}
+		status, err := client.GetStatus()
+		if err != nil {
+			client.Close()
+			client = nil
+			continue
+		}
+		lost = p.noteKernelLost(lost, status.Lost)
+	}
+}
+
+// noteKernelLost counts records the kernel dropped since the last check and
+// returns the new reading. A smaller reading means the counter was reset.
+func (p *ProcessMonitor) noteKernelLost(last, current uint32) uint32 {
+	delta := current
+	if current >= last {
+		delta = current - last
+	}
+	if delta > 0 {
+		noteLoss(&p.kernelLost, int64(delta), "kernel reported lost audit records")
+	}
+	return current
+}
+
 func fileMonitorAuditRule(workingDirectory, goarch string) string {
 	if goarch == "arm64" {
 		return fmt.Sprintf("-a exit,always -F dir=%s -F perm=wa -S openat -S renameat -k %s", workingDirectory, fileMonitorTag)

procmon_linux_test.go

@@ -251,3 +251,132 @@ func TestReceiveWaitsForMaintenanceBeforeReturning(t *testing.T) {
 		t.Fatalf("receive should stop when the client is closed, got %v", err)
 	}
 }
+
+// fakeQueueControl records the kernel queue settings it is asked to change.
+type fakeQueueControl struct {
+	calls []string
+	err   error
+}
+
+func (f *fakeQueueControl) SetFailure(fm libaudit.FailureMode, _ libaudit.WaitMode) error {
+	f.calls = append(f.calls, fmt.Sprintf("failure=%d", fm))
+	return f.err
+}
+
+func (f *fakeQueueControl) SetBacklogLimit(limit uint32, _ libaudit.WaitMode) error {
+	f.calls = append(f.calls, fmt.Sprintf("backlog_limit=%d", limit))
+	return f.err
+}
+
+func (f *fakeQueueControl) SetBacklogWaitTime(waitTime int32, _ libaudit.WaitMode) error {
+	f.calls = append(f.calls, fmt.Sprintf("backlog_wait_time=%d", waitTime))
+	return f.err
+}
+
+func (f *fakeQueueControl) SetRateLimit(limit uint32, _ libaudit.WaitMode) error {
+	f.calls = append(f.calls, fmt.Sprintf("rate_limit=%d", limit))
+	return f.err
+}
+
+func Test_configureBackpressure(t *testing.T) {
+	const waitTimeSupported = libaudit.AuditFeatureBitmapBacklogWaitTime
+	cases := []struct {
+		name   string
+		status libaudit.AuditStatus
+		want   []string
+	}{
+		{
+			name:   "kernel defaults with wait time support",
+			status: libaudit.AuditStatus{Enabled: 1, Failure: 1, BacklogLimit: 64, BacklogWaitTime: 60000, FeatureBitmap: waitTimeSupported},
+			want:   []string{"failure=0", "backlog_limit=8192", "backlog_wait_time=0"},
+		},
+		{
+			name:   "panic failure mode is made silent before dropping",
+			status: libaudit.AuditStatus{Enabled: 1, Failure: 2, BacklogLimit: 8192, FeatureBitmap: waitTimeSupported},
+			want:   []string{"failure=0", "backlog_wait_time=0"},
+		},
+		{
+			name:   "no wait time support falls back to a rate limit",
+			status: libaudit.AuditStatus{Enabled: 1, BacklogLimit: 64},
+			want:   []string{"backlog_limit=8192", "rate_limit=5000"},
+		},
+		{
+			name:   "existing rate limit is cleared when wait time is supported",
+			status: libaudit.AuditStatus{Enabled: 1, BacklogLimit: 8192, RateLimit: 100, FeatureBitmap: waitTimeSupported},
+			want:   []string{"backlog_wait_time=0", "rate_limit=0"},
+		},
+		{
+			name:   "fallback already applied",
+			status: libaudit.AuditStatus{Enabled: 1, BacklogLimit: 8192, RateLimit: 5000},
+			want:   nil,
+		},
+		{
+			name:   "locked audit config is left alone",
+			status: libaudit.AuditStatus{Enabled: 2, BacklogLimit: 64, FeatureBitmap: waitTimeSupported},
+			want:   nil,
+		},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			f := &fakeQueueControl{}
+			status := c.status
+			if err := configureBackpressure(f, &status); err != nil {
+				t.Fatalf("configureBackpressure() error = %v", err)
+			}
+			if fmt.Sprint(f.calls) != fmt.Sprint(c.want) {
+				t.Errorf("calls = %v, want %v", f.calls, c.want)
+			}
+		})
+	}
+}
+
+func Test_configureBackpressure_error(t *testing.T) {
+	f := &fakeQueueControl{err: errors.New("EPERM")}
+	status := libaudit.AuditStatus{Enabled: 1, Failure: 2, BacklogLimit: 64, FeatureBitmap: libaudit.AuditFeatureBitmapBacklogWaitTime}
+	if err := configureBackpressure(f, &status); err == nil {
+		t.Fatal("configureBackpressure() error = nil, want error")
+	}
+	if len(f.calls) != 1 {
+		t.Errorf("calls = %v, want to stop before any drop setting", f.calls)
+	}
+}
+
+func Test_noteKernelLost(t *testing.T) {
+	p := &ProcessMonitor{}
+	steps := []struct {
+		current   uint32
+		wantTotal int64
+	}{
+		{10, 0},  // first reading after start: 10 were lost before the agent
+		{10, 0},  // nothing new
+		{25, 15}, // 15 new
+		{3, 18},  // counter was reset, 3 new since the reset
+		{3, 18},
+	}
+	last := uint32(10)
+	for i, s := range steps {
+		last = p.noteKernelLost(last, s.current)
+		if last != s.current {
+			t.Errorf("step %d: returned %d, want %d", i, last, s.current)
+		}
+		if got := p.kernelLost.Load(); got != s.wantTotal {
+			t.Errorf("step %d: kernelLost = %d, want %d", i, got, s.wantTotal)
+		}
+	}
+}
+
+func Test_monitorKernelLost_stops(t *testing.T) {
+	p := &ProcessMonitor{}
+	done := make(chan struct{})
+	stopped := make(chan struct{})
+	go func() {
+		p.monitorKernelLost(0, done)
+		close(stopped)
+	}()
+	close(done)
+	select {
+	case <-stopped:
+	case <-time.After(5 * time.Second):
+		t.Fatal("monitorKernelLost did not stop after done was closed")
+	}
+}