[Auditbeat] Cherry-pick #10897 to 7.x: System module: Fix and unify b…

…ucket closing logic (#11026)

Cherry-pick of PR #10897 to 7.x branch. Original message: 

The `host` dataset is erroneously trying to save state in its `Close()` method. It should have saved the state earlier - usually at the end of `Fetch()` - and then should only close the bucket (something it is not doing at all). At the same time, it is not saving state in its `reportState()` method. Combined, this can lead to an error when the dataset is terminated before the first regular `reportChanges()` is run.

This fixes both issues and furthermore unifies the bucket closing logic across all six datasets of the System module.

CHANGELOG.next.asciidoc

@@ -174,6 +174,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Enable System module config on Windows. {pull}10237[10237]
 - Package: Disable librpm signal handlers. {pull}10694[10694]
 - Login: Handle different bad login UTMP types. {pull}10865[10865]
+- System module: Fix and unify bucket closing logic. {pull}10897[10897]
 
 *Filebeat*
 

x-pack/auditbeat/module/system/host/host.go

@@ -190,7 +190,10 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 
 // Close cleans up the MetricSet when it finishes.
 func (ms *MetricSet) Close() error {
-	return ms.saveStateToDisk()
+	if ms.bucket != nil {
+		return ms.bucket.Close()
+	}
+	return nil
 }
 
 // Fetch collects data about the host. It is invoked periodically.
@@ -224,7 +227,7 @@ func (ms *MetricSet) reportState(report mb.ReporterV2) error {
 
 	report.Event(hostEvent(host, eventTypeState, eventActionHost))
 
-	return nil
+	return ms.saveStateToDisk()
 }
 
 // reportChanges detects and reports any changes to this host since the last call.
@@ -375,17 +378,19 @@ func inflect(noun string, count int) string {
 func (ms *MetricSet) saveStateToDisk() error {
 	var buf bytes.Buffer
 	encoder := gob.NewEncoder(&buf)
-	err := encoder.Encode(*ms.lastHost)
-	if err != nil {
-		return errors.Wrap(err, "error encoding host information")
-	}
+	if ms.lastHost != nil {
+		err := encoder.Encode(*ms.lastHost)
+		if err != nil {
+			return errors.Wrap(err, "error encoding host information")
+		}
 
-	err = ms.bucket.Store(bucketKeyLastHost, buf.Bytes())
-	if err != nil {
-		return errors.Wrap(err, "error writing host information to disk")
-	}
+		err = ms.bucket.Store(bucketKeyLastHost, buf.Bytes())
+		if err != nil {
+			return errors.Wrap(err, "error writing host information to disk")
+		}
 
-	ms.log.Debug("Wrote host information to disk.")
+		ms.log.Debug("Wrote host information to disk.")
+	}
 	return nil
 }
 

x-pack/auditbeat/module/system/login/utmp.go

@@ -84,8 +84,10 @@ func NewUtmpFileReader(log *logp.Logger, bucket datastore.Bucket, config config)
 
 // Close performs any cleanup tasks when the UTMP reader is done.
 func (r *UtmpFileReader) Close() error {
-	err := r.bucket.Close()
-	return errors.Wrap(err, "error closing bucket")
+	if r.bucket != nil {
+		return r.bucket.Close()
+	}
+	return nil
 }
 
 // ReadNew returns any new UTMP entries in any files matching the configured pattern.