Add input metrics to Auditbeat

[chrisberkhout]

Set up Auditbeat to use libbeat's inputmon package to collect metrics, as is already done for Packetbeat and Filebeat.

It will be important to ensure that a suitable unique, persistent ID is used to refer to each Auditbeat process.

Make sure the Agent integration pulls these metrics from Auditbeat.

Use case:

Once Auditbeat is set up for metric collection, the system.socket dataset can be instrumented to better understand memory usage in certain reported cases.

[elasticmachine]

Pinging @elastic/security-external-integrations (Team:Security-External Integrations)

[chrisberkhout]

As discussed with with @andrewkroh.

[andrewkroh]

The metricbeat framework that Auditbeat is built upon provides each MetricSet instance with its own monitoring.Registry. Any metrics added to that registry were not exposed on the /inputs/ API, so I have opened #36971 to address that part.

beats/metricbeat/mb/mb.go

Line 145 in 2bdc35b

Metrics() *monitoring.Registry // MetricSet specific metrics

You can begin instrumenting auditbeat dataset by doing something like this:

diff --git a/x-pack/auditbeat/module/system/socket/socket_linux.go b/x-pack/auditbeat/module/system/socket/socket_linux.go
index c7b7a97945..b267422174 100644
--- a/x-pack/auditbeat/module/system/socket/socket_linux.go
+++ b/x-pack/auditbeat/module/system/socket/socket_linux.go
@@ -32,6 +32,7 @@ import (
 	"github.com/elastic/beats/v7/x-pack/auditbeat/tracing"
 	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/elastic/elastic-agent-libs/mapstr"
+	"github.com/elastic/elastic-agent-libs/monitoring"
 	"github.com/elastic/go-perf"
 	"github.com/elastic/go-sysinfo"
 	"github.com/elastic/go-sysinfo/providers/linux"
@@ -77,6 +78,8 @@ type MetricSet struct {
 	isDebug      bool
 	isDetailed   bool
 	terminated   sync.WaitGroup
+
+	cacheSize *monitoring.Uint
 }
 
 func init() {
@@ -138,6 +141,7 @@ func newSocketMetricset(config Config, base mb.BaseMetricSet) (*MetricSet, error
 		detailLog:       logp.NewLogger(detailSelector),
 		isDetailed:      logp.HasSelector(detailSelector),
 		sniffer:         sniffer,
+		cacheSize:       monitoring.NewUint(base.Metrics(), "cache_size"),
 	}
 	// Setup the metricset before Run() so that startup can be halted in case of
 	// error.

Assuming your auditbeat.yml config contains

http.host: 127.0.0.1
http.port: 6060

then that new metric would be visible when you query curl http://localhost:6060/inputs/.

[andrewkroh]

The metrics registered by the auditd module are global (i.e. not per instance). I'm leery of removing them from the global namespace because users might be monitoring them. But we could expose the same counter on the per-instance registry so that those metrics become visible on /inputs/.

beats/auditbeat/module/auditd/audit_linux.go

Lines 66 to 72 in ee864b5

	var (
	auditdMetrics = monitoring.Default.NewRegistry(moduleName)
	reassemblerGapsMetric = monitoring.NewInt(auditdMetrics, "reassembler_seq_gaps")
	kernelLostMetric = monitoring.NewInt(auditdMetrics, "kernel_lost")
	userspaceLostMetric = monitoring.NewInt(auditdMetrics, "userspace_lost")
	receivedMetric = monitoring.NewInt(auditdMetrics, "received_msgs")
	)

For example, I think this would work fine:

	base.Metrics().Add("reassembler_seq_gaps", reassemblerGapsMetric, monitoring.Full)
	base.Metrics().Add("kernel_lost", kernelLostMetric, monitoring.Full)
	base.Metrics().Add("userspace_lost", userspaceLostMetric, monitoring.Full)
	base.Metrics().Add("received_msgs", receivedMetric, monitoring.Full)

[andrewkroh]

Another issue I noticed is that the Metricbeat framework provides a logger to each MetricSet instance, but Auditbeat is not using that logger. Instead it creates its own which might result in inconsistencies, particularly after #36971 merges which adds the Fleet stream id to the logger as context. Having the id included with any logs will help us correlate logs with the metrics and associated configuration.

We should make sure that we use that logger as the starting point for any loggers within out modules. For example

diff --git a/auditbeat/module/file_integrity/metricset.go b/auditbeat/module/file_integrity/metricset.go
index bcada27db9..d2a0a18d6d 100644
--- a/auditbeat/module/file_integrity/metricset.go
+++ b/auditbeat/module/file_integrity/metricset.go
@@ -171,7 +171,7 @@ func (ms *MetricSet) init(reporter mb.PushReporterV2) bool {
 
        ms.scanStart = time.Now().UTC()
        if ms.config.ScanAtStart {
-               ms.scanner, err = NewFileSystemScanner(ms.config, ms.findNewPaths())
+               ms.scanner, err = NewFileSystemScanner(ms.Logger(), ms.config, ms.findNewPaths())
                if err != nil {
                        err = fmt.Errorf("failed to initialize file scanner: %w", err)
                        reporter.Error(err)
diff --git a/auditbeat/module/file_integrity/scanner.go b/auditbeat/module/file_integrity/scanner.go
index cef40df630..5b2933d056 100644
--- a/auditbeat/module/file_integrity/scanner.go
+++ b/auditbeat/module/file_integrity/scanner.go
@@ -52,9 +52,9 @@ type scanner struct {
 // NewFileSystemScanner creates a new EventProducer instance that scans the
 // configured file paths. Files and directories in new paths are recorded with
 // the action `found`.
-func NewFileSystemScanner(c Config, newPathsInConfig map[string]struct{}) (EventProducer, error) {
+func NewFileSystemScanner(logger *logp.Logger, c Config, newPathsInConfig map[string]struct{}) (EventProducer, error) {
        return &scanner{
-               log:      logp.NewLogger(moduleName).With("scanner_id", atomic.AddUint32(&scannerID, 1)),
+               log:      logger.With("scanner_id", atomic.AddUint32(&scannerID, 1)),
                config:   c,
                newPaths: newPathsInConfig,
                eventC:   make(chan Event, 1),

[elasticmachine]

Pinging @elastic/sec-linux-platform (Team:Security-Linux Platform)