Update logging from auditd module (elastic#6018)

Use logp.Logger for all logging from the auditd module. I also increased the logging level for two statements because they will be useful for troubleshooting (without having to ask users to re-run with debug enabled).

Vagrantfile

@@ -138,7 +138,7 @@ Vagrant.configure(2) do |config|
 
     c.vm.provision "shell", inline: $unixProvision, privileged: false
     c.vm.provision "shell", inline: $linuxGvmProvision, privileged: false
-    config.vm.provision "shell", inline: "dnf install -y make gcc python-pip python-virtualenv git"
+    c.vm.provision "shell", inline: "dnf install -y make gcc python-pip python-virtualenv git"
 
     c.vm.synced_folder ".", "/vagrant", type: "virtualbox"
   end

auditbeat/module/auditd/audit_linux.go

@@ -21,15 +21,11 @@ import (
 )
 
 const (
-	logPrefix = "[" + moduleName + "]"
-
 	// Use old namespace for data until we do some field renaming for GA.
 	namespace = "audit.kernel"
 )
 
 var (
-	debugf = logp.MakeDebug(moduleName)
-
 	auditdMetrics = monitoring.Default.NewRegistry(moduleName)
 	lostMetric    = monitoring.NewInt(auditdMetrics, "lost")
 )
@@ -50,6 +46,7 @@ type MetricSet struct {
 	mb.BaseMetricSet
 	config Config
 	client *libaudit.AuditClient
+	log    *logp.Logger
 }
 
 // New constructs a new MetricSet.
@@ -61,10 +58,11 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 		return nil, errors.Wrap(err, "failed to unpack the auditd config")
 	}
 
+	log := logp.NewLogger(moduleName)
 	_, _, kernel, _ := kernelVersion()
-	debugf("auditd module is running as euid=%v on kernel=%v", os.Geteuid(), kernel)
+	log.Infof("auditd module is running as euid=%v on kernel=%v", os.Geteuid(), kernel)
 
-	client, err := newAuditClient(&config)
+	client, err := newAuditClient(&config, log)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create audit client")
 	}
@@ -75,10 +73,11 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 		BaseMetricSet: base,
 		client:        client,
 		config:        config,
+		log:           log,
 	}, nil
 }
 
-func newAuditClient(c *Config) (*libaudit.AuditClient, error) {
+func newAuditClient(c *Config, log *logp.Logger) (*libaudit.AuditClient, error) {
 	hasMulticast := hasMulticastSupport()
 
 	switch c.SocketType {
@@ -89,13 +88,13 @@ func newAuditClient(c *Config) (*libaudit.AuditClient, error) {
 		// using unicast.
 		if rules, _ := c.rules(); len(rules) == 0 && hasMulticast {
 			c.SocketType = "multicast"
-			logp.Info("%v socket_type=multicast will be used.", logPrefix)
+			log.Info("socket_type=multicast will be used.")
 		}
 	case "multicast":
 		if !hasMulticast {
-			logp.Warn("%v socket_type is set to multicast "+
-				"but based on the kernel version multicast audit subscriptions "+
-				"are not supported. unicast will be used instead.", logPrefix)
+			log.Warn("socket_type is set to multicast but based on the " +
+				"kernel version multicast audit subscriptions are not " +
+				"supported. unicast will be used instead.")
 			c.SocketType = "unicast"
 		}
 	}
@@ -116,14 +115,14 @@ func (ms *MetricSet) Run(reporter mb.PushReporterV2) {
 
 	if err := ms.addRules(reporter); err != nil {
 		reporter.Error(err)
-		logp.Err("%v %v", logPrefix, err)
+		ms.log.Errorw("Failure adding audit rules", "error", err)
 		return
 	}
 
 	out, err := ms.receiveEvents(reporter.Done())
 	if err != nil {
 		reporter.Error(err)
-		logp.Err("%v %v", logPrefix, err)
+		ms.log.Errorw("Failure receiving audit events", "error", err)
 		return
 	}
 
@@ -144,7 +143,7 @@ func (ms *MetricSet) addRules(reporter mb.PushReporterV2) error {
 	}
 
 	if len(rules) == 0 {
-		logp.Info("%v No audit_rules were specified.", logPrefix)
+		ms.log.Info("No audit_rules were specified.")
 		return nil
 	}
 
@@ -159,7 +158,7 @@ func (ms *MetricSet) addRules(reporter mb.PushReporterV2) error {
 	if err != nil {
 		return errors.Wrap(err, "failed to delete existing rules")
 	}
-	logp.Info("%v Deleted %v pre-existing audit rules.", logPrefix, n)
+	ms.log.Infof("Deleted %v pre-existing audit rules.", n)
 
 	// Add rules from config.
 	var failCount int
@@ -168,12 +167,12 @@ func (ms *MetricSet) addRules(reporter mb.PushReporterV2) error {
 			// Treat rule add errors as warnings and continue.
 			err = errors.Wrapf(err, "failed to add audit rule '%v'", rule.flags)
 			reporter.Error(err)
-			logp.Warn("%v %v", logPrefix, err)
+			ms.log.Warnw("Failure adding audit rule", err)
 			failCount++
 		}
 	}
-	logp.Info("%v Successfully added %d of %d audit rules.",
-		logPrefix, len(rules)-failCount, len(rules))
+	ms.log.Infof("Successfully added %d of %d audit rules.",
+		len(rules)-failCount, len(rules))
 	return nil
 }
 
@@ -193,7 +192,7 @@ func (ms *MetricSet) initClient() error {
 	if err != nil {
 		return errors.Wrap(err, "failed to get audit status")
 	}
-	debugf("audit status from kernel at start: status=%+v", status)
+	ms.log.Infow("audit status from kernel at start", "audit_status", status)
 
 	if fm, _ := ms.config.failureMode(); status.Failure != fm {
 		if err = ms.client.SetFailure(libaudit.FailureMode(fm), libaudit.NoWait); err != nil {
@@ -253,8 +252,10 @@ func (ms *MetricSet) receiveEvents(done <-chan struct{}) (<-chan []*auparse.Audi
 			}
 
 			if err := reassembler.Push(raw.Type, raw.Data); err != nil {
-				debugf("dropping message record_type=%v message='%v': ",
-					raw.Type, string(raw.Data), err)
+				ms.log.Debugw("Dropping audit message",
+					"record_type", raw.Type,
+					"message", string(raw.Data),
+					"error", err)
 				continue
 			}
 		}

auditbeat/module/auditd/audit_linux_test.go

@@ -11,6 +11,7 @@ import (
 	"github.com/stretchr/testify/assert"
 
 	"github.com/elastic/beats/auditbeat/core"
+	"github.com/elastic/beats/libbeat/logp"
 	mbtest "github.com/elastic/beats/metricbeat/mb/testing"
 	"github.com/elastic/go-libaudit"
 	"github.com/elastic/procfs"
@@ -25,6 +26,8 @@ var audit = flag.Bool("audit", false, "interact with the real audit framework")
 var userLoginMsg = `type=USER_LOGIN msg=audit(1492896301.818:19955): pid=12635 uid=0 auid=4294967295 ses=4294967295 msg='op=login acct=28696E76616C6964207573657229 exe="/usr/sbin/sshd" hostname=? addr=179.38.151.221 terminal=sshd res=failed'`
 
 func TestData(t *testing.T) {
+	logp.TestingSetup()
+
 	// Create a mock netlink client that provides the expected responses.
 	mock := NewMock().
 		// Get Status response for initClient
@@ -65,6 +68,7 @@ func TestUnicastClient(t *testing.T) {
 		t.Skip("-audit was not specified")
 	}
 
+	logp.TestingSetup()
 	FailIfAuditdIsRunning(t)
 
 	c := map[string]interface{}{
@@ -109,6 +113,7 @@ func TestMulticastClient(t *testing.T) {
 		t.Skip("no multicast support")
 	}
 
+	logp.TestingSetup()
 	FailIfAuditdIsRunning(t)
 
 	c := map[string]interface{}{

metricbeat/mb/testing/modules.go

@@ -269,7 +269,10 @@ func RunPushMetricSetV2(timeout time.Duration, waitEvents int, metricSet mb.Push
 			select {
 			case <-timer.C:
 				return
-			case e := <-r.eventsC:
+			case e, ok := <-r.eventsC:
+				if !ok {
+					return
+				}
 				events = append(events, e)
 				if waitEvents > 0 && waitEvents <= len(events) {
 					return