MB mongodb module: connect on fetch, not on init

CHANGELOG.asciidoc

@@ -52,6 +52,7 @@ https://github.com/elastic/beats/compare/v6.0.0-beta2...master[Check the HEAD di
 - Fix a memory allocation issue where more memory was allocated than needed in the windows-perfmon metricset. {issue}5035[5035]
 - Don't start metricbeat if external modules config is wrong and reload is disabled {pull}5053[5053]
 - Fix kubernetes events module to be able to index time fields properly. {issue}5093[5093]
+- The MongoDB module now connects on each fetch, to avoid stopping the whole Metricbeat instance if MongoDB is not up when starting. {pull}5120[5120]
 
 *Packetbeat*
 

metricbeat/module/mongodb/dbstats/dbstats.go

@@ -28,7 +28,7 @@ func init() {
 // multiple fetch calls.
 type MetricSet struct {
 	mb.BaseMetricSet
-	mongoSession *mgo.Session
+	dialInfo *mgo.DialInfo
 }
 
 // New creates a new instance of the MetricSet
@@ -43,15 +43,9 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 	}
 	dialInfo.Timeout = base.Module().Config().Timeout
 
-	// instantiate direct connections to each of the configured Mongo hosts
-	mongoSession, err := mongodb.NewDirectSession(dialInfo)
-	if err != nil {
-		return nil, err
-	}
-
 	return &MetricSet{
 		BaseMetricSet: base,
-		mongoSession:  mongoSession,
+		dialInfo:      dialInfo,
 	}, nil
 }
 
@@ -62,16 +56,22 @@ func (m *MetricSet) Fetch() ([]common.MapStr, error) {
 	// events is the list of events collected from each of the databases.
 	var events []common.MapStr
 
+	// instantiate direct connections to each of the configured Mongo hosts
+	mongoSession, err := mongodb.NewDirectSession(m.dialInfo)
+	if err != nil {
+		return nil, err
+	}
+
 	// Get the list of databases names, which we'll use to call db.stats() on each
-	dbNames, err := m.mongoSession.DatabaseNames()
+	dbNames, err := mongoSession.DatabaseNames()
 	if err != nil {
 		logp.Err("Error retrieving database names from Mongo instance")
 		return events, err
 	}
 
 	// for each database, call db.stats() and append to events
 	for _, dbName := range dbNames {
-		db := m.mongoSession.DB(dbName)
+		db := mongoSession.DB(dbName)
 
 		result := common.MapStr{}
 

metricbeat/module/mongodb/mongodb.go

@@ -74,7 +74,7 @@ func NewDirectSession(dialInfo *mgo.DialInfo) (*mgo.Session, error) {
 	nodeDialInfo.Direct = true
 	nodeDialInfo.FailFast = true
 
-	logp.Info("Connecting to MongoDB node at %v", nodeDialInfo.Addrs)
+	logp.Debug("mongodb", "Connecting to MongoDB node at %v", nodeDialInfo.Addrs)
 
 	session, err := mgo.DialWithInfo(&nodeDialInfo)
 	if err != nil {

metricbeat/module/mongodb/status/status.go

@@ -30,7 +30,7 @@ func init() {
 // multiple fetch calls.
 type MetricSet struct {
 	mb.BaseMetricSet
-	mongoSession *mgo.Session
+	dialInfo *mgo.DialInfo
 }
 
 // New creates a new instance of the MetricSet
@@ -43,24 +43,25 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 	}
 	dialInfo.Timeout = base.Module().Config().Timeout
 
-	// instantiate direct connections to Mongo host
-	mongoSession, err := mongodb.NewDirectSession(dialInfo)
-	if err != nil {
-		return nil, err
-	}
-
 	return &MetricSet{
 		BaseMetricSet: base,
-		mongoSession:  mongoSession,
+		dialInfo:      dialInfo,
 	}, nil
 }
 
 // Fetch methods implements the data gathering and data conversion to the right format
 // It returns the event which is then forward to the output. In case of an error, a
 // descriptive error must be returned.
 func (m *MetricSet) Fetch() (common.MapStr, error) {
+
+	// instantiate direct connections to each of the configured Mongo hosts
+	mongoSession, err := mongodb.NewDirectSession(m.dialInfo)
+	if err != nil {
+		return nil, err
+	}
+
 	result := map[string]interface{}{}
-	if err := m.mongoSession.DB("admin").Run(bson.D{{Name: "serverStatus", Value: 1}}, &result); err != nil {
+	if err := mongoSession.DB("admin").Run(bson.D{{Name: "serverStatus", Value: 1}}, &result); err != nil {
 		return nil, err
 	}