Stop worker goroutine on auditd metricset shutdown

CHANGELOG.asciidoc

@@ -69,6 +69,7 @@ https://github.com/elastic/beats/compare/v6.2.3...master[Check the HEAD diff]
 - Fixed a deadlock in the file_integrity module under Windows. {issue}6864[6864]
 - Fixed parsing of AppArmor audit messages. {pull}6978[6978]
 - Allow `auditbeat setup` to run without requiring elevated privileges for the audit client. {issue}7111[7111]
+- Fix goroutine leak that occured when the auditd module was stopped. {pull}7163[7163]
 
 *Filebeat*
 

auditbeat/module/auditd/_meta/data.json

@@ -32,10 +32,16 @@
         "module": "auditd",
         "type": "user_login"
     },
+    "network": {
+        "direction": "incoming"
+    },
     "process": {
         "exe": "/usr/sbin/sshd",
         "pid": "12635"
     },
+    "source": {
+        "ip": "179.38.151.221"
+    },
     "user": {
         "auid": "unset",
         "name_map": {

auditbeat/module/auditd/_meta/execve.json

@@ -40,8 +40,8 @@
         "session": "11",
         "summary": {
             "actor": {
-                "primary": "1001",
-                "secondary": "1001"
+                "primary": "ubuntu",
+                "secondary": "ubuntu"
             },
             "how": "/bin/uname",
             "object": {
@@ -88,6 +88,13 @@
         "fsgid": "1002",
         "fsuid": "1001",
         "gid": "1002",
+        "name_map": {
+            "auid": "ubuntu",
+            "euid": "ubuntu",
+            "fsuid": "ubuntu",
+            "suid": "ubuntu",
+            "uid": "ubuntu"
+        },
         "sgid": "1002",
         "suid": "1001",
         "uid": "1001"

auditbeat/module/auditd/audit_linux.go

@@ -243,12 +243,17 @@ func (ms *MetricSet) receiveEvents(done <-chan struct{}) (<-chan []*auparse.Audi
 	go maintain(done, reassembler)
 
 	go func() {
+		defer ms.log.Debug("receiveEvents goroutine exited")
 		defer close(out)
 		defer reassembler.Close()
 
 		for {
 			raw, err := ms.client.Receive(false)
 			if err != nil {
+				if errors.Cause(err) == syscall.EBADF {
+					// Client has been closed.
+					break
+				}
 				continue
 			}
 
@@ -289,8 +294,14 @@ func maintain(done <-chan struct{}, reassembler *libaudit.Reassembler) {
 }
 
 func filterRecordType(typ auparse.AuditMessageType) bool {
+	switch {
+	// REPLACE messages are tests to check if Auditbeat is still healthy by
+	// seeing if unicast messages can be sent without error from the kernel.
+	// Ignore them.
+	case typ == auparse.AUDIT_REPLACE:
+		return true
 	// Messages from 1300-2999 are valid audit message types.
-	if typ < auparse.AUDIT_USER_AUTH || typ > auparse.AUDIT_LAST_USER_MSG2 {
+	case typ < auparse.AUDIT_USER_AUTH || typ > auparse.AUDIT_LAST_USER_MSG2:
 		return true
 	}
 

auditbeat/module/auditd/audit_linux_test.go

@@ -16,6 +16,7 @@ import (
 
 	"github.com/elastic/beats/auditbeat/core"
 	"github.com/elastic/beats/libbeat/logp"
+	"github.com/elastic/beats/metricbeat/mb"
 	mbtest "github.com/elastic/beats/metricbeat/mb/testing"
 	"github.com/elastic/go-libaudit"
 	"github.com/elastic/go-libaudit/auparse"
@@ -67,14 +68,10 @@ func TestData(t *testing.T) {
 	auditMetricSet.client = &libaudit.AuditClient{Netlink: mock}
 
 	events := mbtest.RunPushMetricSetV2(10*time.Second, 1, ms)
-	for _, e := range events {
-		if e.Error != nil {
-			t.Fatalf("received error: %+v", e.Error)
-		}
-	}
 	if len(events) == 0 {
 		t.Fatal("received no events")
 	}
+	assertNoErrors(t, events)
 
 	beatEvent := mbtest.StandardizeEvent(ms, events[0], core.AddDatasetToEvent)
 	mbtest.WriteEventToDataJSON(t, beatEvent, "")
@@ -110,23 +107,8 @@ func TestUnicastClient(t *testing.T) {
 
 	ms := mbtest.NewPushMetricSetV2(t, c)
 	events := mbtest.RunPushMetricSetV2(5*time.Second, 0, ms)
-	for _, e := range events {
-		t.Log(e)
-
-		if e.Error != nil {
-			t.Errorf("received error: %+v", e.Error)
-		}
-	}
-
-	for _, e := range events {
-		v, err := e.MetricSetFields.GetValue("thing.primary")
-		if err == nil {
-			if exe, ok := v.(string); ok && exe == "/bin/cat" {
-				return
-			}
-		}
-	}
-	assert.Fail(t, "expected an execve event for /bin/cat")
+	assertNoErrors(t, events)
+	assertHasBinCatExecve(t, events)
 }
 
 func TestMulticastClient(t *testing.T) {
@@ -155,14 +137,8 @@ func TestMulticastClient(t *testing.T) {
 
 	ms := mbtest.NewPushMetricSetV2(t, c)
 	events := mbtest.RunPushMetricSetV2(5*time.Second, 0, ms)
-	for _, e := range events {
-		if e.Error != nil {
-			t.Fatalf("received error: %+v", e.Error)
-		}
-	}
-
-	// The number of events is non-deterministic so there is no validation.
-	t.Logf("received %d messages via multicast", len(events))
+	assertNoErrors(t, events)
+	assertHasBinCatExecve(t, events)
 }
 
 func TestKernelVersion(t *testing.T) {
@@ -224,3 +200,29 @@ func buildSampleEvent(t testing.TB, lines []string, filename string) {
 		t.Fatal(err)
 	}
 }
+
+func assertHasBinCatExecve(t *testing.T, events []mb.Event) {
+	t.Helper()
+
+	for _, e := range events {
+		v, err := e.RootFields.GetValue("process.exe")
+		if err == nil {
+			if exe, ok := v.(string); ok && exe == "/bin/cat" {
+				return
+			}
+		}
+	}
+	assert.Fail(t, "expected an execve event for /bin/cat")
+}
+
+func assertNoErrors(t *testing.T, events []mb.Event) {
+	t.Helper()
+
+	for _, e := range events {
+		t.Log(e)
+
+		if e.Error != nil {
+			t.Errorf("received error: %+v", e.Error)
+		}
+	}
+}