diff --git a/procmon.go b/procmon.go index 19b840a..40d5e09 100644 --- a/procmon.go +++ b/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 { diff --git a/procmon_linux.go b/procmon_linux.go index e985016..e3d6669 100644 --- a/procmon_linux.go +++ b/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) diff --git a/procmon_linux_test.go b/procmon_linux_test.go index 7c61132..e2813a4 100644 --- a/procmon_linux_test.go +++ b/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") + } +}