Fix mongodb multiple hosts connection issue (#34624)

* Fix mongodb multiple hosts connection issue

CHANGELOG-developer.next.asciidoc

@@ -80,6 +80,7 @@ The list below covers the major changes between 7.0.0-rc2 and main only.
 - The beat.cgroup.memory.mem.usage.bytes metric is now a gauge {issue}31582[31582] {pull}32652[32652]
 - Fix the integration testcase docker port mapping for sql and oracle modules {pull}34221[34221]
 - Fix the ingest pipeline for mysql slowlog to parse schema name with dash {pull}34371[34372]
+- Fix the multiple host support for mongodb module {pull}34624[34624]
 
 ==== Added
 

metricbeat/docs/modules/mongodb.asciidoc

@@ -26,6 +26,12 @@ format:
 [mongodb://][user:pass@]host[:port][?options]
 -----------------------------------
 
+Or
+
+-----------------------------------------------------------------------------------------
+mongodb://[username:password@]host1[:port1][,...hostN[:portN]][/[defaultauthdb][?options]]
+-----------------------------------------------------------------------------------------
+
 The URL can be as simple as:
 
 [source,yaml]
@@ -42,6 +48,32 @@ Or more complex like:
   hosts: ["mongodb://myuser:mypass@localhost:40001", "otherhost:40001"]
 ----------------------------------------------------------------------
 
+Some more supported URLs are:
+
+[source,yaml]
+----------------------------------------------------------------------
+- module: mongodb
+  hosts: ["mongodb://localhost:27017,localhost:27022,localhost:27023"]
+----------------------------------------------------------------------
+
+[source,yaml]
+----------------------------------------------------------------------
+- module: mongodb
+  hosts: ["mongodb://localhost:27017/?directConnection=true"]
+----------------------------------------------------------------------
+
+When the parameter `directConnection=true` is included in the connection URI,
+all operations are executed on the host specified in the URI.
+It's important to note that `directConnection=true` must be explicitly specified in the URI,
+as it won't be added automatically unless specified.
+
+[source,yaml]
+----------------------------------------------------------------------
+- module: mongodb
+  hosts: ["mongodb://localhost:27017,localhost:27022,localhost:27023/?replicaSet=dbrs"]
+----------------------------------------------------------------------
+
+
 The username and password can be included in the URL or they can be set using
 the respective configuration options. The credentials in the URL take precedence
 over the username and password configuration options.
@@ -60,8 +92,8 @@ The default metricsets are `collstats`, `dbstats` and `status`.
 [float]
 === Compatibility
 
-The MongoDB metricsets were tested with MongoDB 3.4 and 3.0 and are expected to
-work with all versions >= 2.8.
+The MongoDB metricsets were tested with MongoDB 5.0 and are expected to
+work with all versions >= 5.0.
 
 [float]
 === MongoDB Privileges

metricbeat/module/mongodb/_meta/docs.asciidoc

@@ -15,6 +15,12 @@ format:
 [mongodb://][user:pass@]host[:port][?options]
 -----------------------------------
 
+Or
+
+-----------------------------------------------------------------------------------------
+mongodb://[username:password@]host1[:port1][,...hostN[:portN]][/[defaultauthdb][?options]]
+-----------------------------------------------------------------------------------------
+
 The URL can be as simple as:
 
 [source,yaml]
@@ -31,6 +37,32 @@ Or more complex like:
   hosts: ["mongodb://myuser:mypass@localhost:40001", "otherhost:40001"]
 ----------------------------------------------------------------------
 
+Some more supported URLs are:
+
+[source,yaml]
+----------------------------------------------------------------------
+- module: mongodb
+  hosts: ["mongodb://localhost:27017,localhost:27022,localhost:27023"]
+----------------------------------------------------------------------
+
+[source,yaml]
+----------------------------------------------------------------------
+- module: mongodb
+  hosts: ["mongodb://localhost:27017/?directConnection=true"]
+----------------------------------------------------------------------
+
+When the parameter `directConnection=true` is included in the connection URI,
+all operations are executed on the host specified in the URI.
+It's important to note that `directConnection=true` must be explicitly specified in the URI,
+as it won't be added automatically unless specified.
+
+[source,yaml]
+----------------------------------------------------------------------
+- module: mongodb
+  hosts: ["mongodb://localhost:27017,localhost:27022,localhost:27023/?replicaSet=dbrs"]
+----------------------------------------------------------------------
+
+
 The username and password can be included in the URL or they can be set using
 the respective configuration options. The credentials in the URL take precedence
 over the username and password configuration options.
@@ -49,8 +81,8 @@ The default metricsets are `collstats`, `dbstats` and `status`.
 [float]
 === Compatibility
 
-The MongoDB metricsets were tested with MongoDB 3.4 and 3.0 and are expected to
-work with all versions >= 2.8.
+The MongoDB metricsets were tested with MongoDB 5.0 and are expected to
+work with all versions >= 5.0.
 
 [float]
 === MongoDB Privileges

metricbeat/module/mongodb/collstats/collstats.go

@@ -59,7 +59,7 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 // format. It publishes the event which is then forwarded to the output. In case
 // of an error set the Error field of mb.Event or simply call report.Error().
 func (m *Metricset) Fetch(reporter mb.ReporterV2) error {
-	client, err := mongodb.NewClient(m.Metricset.Config, m.Module().Config().Timeout, 0)
+	client, err := mongodb.NewClient(m.Metricset.Config, m.HostData().URI, m.Module().Config().Timeout, 0)
 	if err != nil {
 		return fmt.Errorf("could not create mongodb client: %w", err)
 	}

metricbeat/module/mongodb/dbstats/dbstats.go

@@ -62,7 +62,7 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 // format. It publishes the event which is then forwarded to the output. In case
 // of an error set the Error field of mb.Event or simply call report.Error().
 func (m *MetricSet) Fetch(reporter mb.ReporterV2) error {
-	client, err := mongodb.NewClient(m.Metricset.Config, m.Module().Config().Timeout, 0)
+	client, err := mongodb.NewClient(m.Metricset.Config, m.HostData().URI, m.Module().Config().Timeout, 0)
 	if err != nil {
 		return fmt.Errorf("could not create mongodb client: %w", err)
 	}

metricbeat/module/mongodb/dbstats/dbstats_integration_test.go

@@ -50,22 +50,24 @@ func TestFetch(t *testing.T) {
 		assert.True(t, collections > 0)
 
 		objects := metricsetFields["objects"].(int32)
-		assert.True(t, objects > 0)
+		assert.True(t, objects >= 0)
 
 		avgObjSize, err := metricsetFields.GetValue("avg_obj_size.bytes")
 		assert.NoError(t, err)
-		assert.True(t, avgObjSize.(float64) > 0)
+		assert.True(t, avgObjSize.(float64) >= 0)
 
 		dataSize, err := metricsetFields.GetValue("data_size.bytes")
 		assert.NoError(t, err)
-		assert.True(t, dataSize.(float64) > 0)
+		assert.True(t, dataSize.(float64) >= 0)
 
 		storageSize, err := metricsetFields.GetValue("storage_size.bytes")
 		assert.NoError(t, err)
-		assert.True(t, storageSize.(float64) > 0)
+		assert.True(t, storageSize.(float64) >= 0)
 
-		numExtents := metricsetFields["num_extents"].(int32)
-		assert.True(t, numExtents >= 0)
+		if metricsetFields["num_extents"] != nil {
+			numExtents := metricsetFields["num_extents"].(int32)
+			assert.True(t, numExtents >= 0)
+		}
 
 		indexes := metricsetFields["indexes"].(int32)
 		assert.True(t, indexes >= 0)

metricbeat/module/mongodb/docker-compose.yml

@@ -1,12 +1,11 @@
-version: "2.3"
+version: '2.3'
 
 services:
   mongodb:
-    image: docker.elastic.co/integrations-ci/beats-mongodb:${MONGODB_VERSION:-3.4}-1
+    image: docker.elastic.co/integrations-ci/beats-mongodb:${MONGODB_VERSION:-5.0}-1
     build:
       context: ./_meta
       args:
-        MONGODB_VERSION: "${MONGODB_VERSION:-3.4}"
-    command: mongod --replSet beats
+        MONGODB_VERSION: ${MONGODB_VERSION:-5.0}
     ports:
-      - 27017
+      - 27017:27017

metricbeat/module/mongodb/metrics/metrics.go

@@ -58,7 +58,7 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 // format. It publishes the event which is then forwarded to the output. In case
 // of an error set the Error field of mb.Event or simply call report.Error().
 func (m *MetricSet) Fetch(reporter mb.ReporterV2) error {
-	client, err := mongodb.NewClient(m.Metricset.Config, m.Module().Config().Timeout, 0)
+	client, err := mongodb.NewClient(m.Metricset.Config, m.HostData().URI, m.Module().Config().Timeout, 0)
 	if err != nil {
 		return fmt.Errorf("could not create mongodb client: %w", err)
 	}

metricbeat/module/mongodb/mongodb.go

@@ -114,28 +114,11 @@ func ParseURL(module mb.Module, host string) (mb.HostData, error) {
 
 	parse.SetURLUser(u, c.Username, c.Password)
 
-	clientOptions := options.Client()
-	clientOptions.Auth = &options.Credential{
-		AuthMechanism:           c.Credentials.AuthMechanism,
-		AuthMechanismProperties: c.Credentials.AuthMechanismProperties,
-		AuthSource:              c.Credentials.AuthSource,
-		PasswordSet:             c.Credentials.PasswordSet,
-		Username:                c.Username,
-		Password:                c.Password,
-	}
-	clientOptions.SetDirect(true)
-	clientOptions.ApplyURI(host)
-
-	// https://docs.mongodb.com/manual/reference/connection-string/
-	_, err = url.Parse(clientOptions.GetURI())
-	if err != nil {
-		return mb.HostData{}, fmt.Errorf("error parsing URL: %w", err)
-	}
-
 	return parse.NewHostDataFromURL(u), nil
 }
 
-func NewClient(config ModuleConfig, timeout time.Duration, mode readpref.Mode) (*mongo.Client, error) {
+func NewClient(config ModuleConfig, uri string, timeout time.Duration, mode readpref.Mode) (*mongo.Client, error) {
+
 	clientOptions := options.Client()
 
 	// options.Credentials must be nil for the driver to work properly if no auth is provided. Zero values breaks
@@ -154,7 +137,8 @@ func NewClient(config ModuleConfig, timeout time.Duration, mode readpref.Mode) (
 			clientOptions.Auth.AuthMechanismProperties = config.Credentials.AuthMechanismProperties
 		}
 	}
-	clientOptions.SetHosts(config.Hosts)
+
+	clientOptions.ApplyURI(uri)
 
 	if mode == 0 {
 		mode = readpref.NearestMode
@@ -165,7 +149,6 @@ func NewClient(config ModuleConfig, timeout time.Duration, mode readpref.Mode) (
 		return nil, err
 	}
 	clientOptions.SetReadPreference(readPreference)
-	clientOptions.SetDirect(true)
 	clientOptions.SetConnectTimeout(timeout)
 
 	if config.TLS.IsEnabled() {

metricbeat/module/mongodb/mongodb_test.go

@@ -77,7 +77,7 @@ func TestParseMongoURL(t *testing.T) {
 		},
 		{
 			Name:     "with options",
-			URL:      "mongodb://localhost:40001?connect=direct&authSource=me",
+			URL:      "mongodb://localhost:40001/directConnection=true&authSource=me",
 			Username: "anotheruser",
 			Password: "anotherpass",
 
@@ -95,6 +95,16 @@ func TestParseMongoURL(t *testing.T) {
 			ExpectedUsername: "",
 			ExpectedPassword: "",
 		},
+		{
+			Name:     "with replicaSet option",
+			URL:      "mongodb://localhost:40001,localhost:40002/?replicaSet=dbrs",
+			Username: "anotheruser",
+			Password: "anotherpass",
+
+			ExpectedAddr:     "localhost:40001,localhost:40002",
+			ExpectedUsername: "anotheruser",
+			ExpectedPassword: "anotherpass",
+		},
 	}
 
 	for _, test := range tests {

metricbeat/module/mongodb/replstatus/replstatus.go

@@ -55,7 +55,7 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 // format. It publishes the event which is then forwarded to the output. In case
 // of an error set the Error field of mb.Event or simply call report.Error().
 func (m *MetricSet) Fetch(reporter mb.ReporterV2) error {
-	client, err := mongodb.NewClient(m.Metricset.Config, m.Module().Config().Timeout, readpref.PrimaryMode)
+	client, err := mongodb.NewClient(m.Metricset.Config, m.HostData().URI, m.Module().Config().Timeout, readpref.PrimaryMode)
 	if err != nil {
 		return fmt.Errorf("could not create mongodb client: %w", err)
 	}

metricbeat/module/mongodb/replstatus/replstatus_integration_test.go

@@ -104,9 +104,10 @@ func getConfig(host string) map[string]interface{} {
 }
 
 func initiateReplicaSet(t *testing.T, host string) error {
+	uri := "mongodb://" + host
 	client, err := mongodb.NewClient(mongodb.ModuleConfig{
 		Hosts: []string{host},
-	}, time.Second*5, readpref.PrimaryMode)
+	}, uri, time.Second*5, readpref.PrimaryMode)
 	if err != nil {
 		return fmt.Errorf("could not create mongodb client: %w", err)
 	}

metricbeat/module/mongodb/status/status.go

@@ -61,7 +61,7 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 // 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(r mb.ReporterV2) error {
-	client, err := mongodb.NewClient(m.Metricset.Config, m.Module().Config().Timeout, readpref.PrimaryMode)
+	client, err := mongodb.NewClient(m.Metricset.Config, m.HostData().URI, m.Module().Config().Timeout, readpref.PrimaryMode)
 	if err != nil {
 		return fmt.Errorf("could not create mongodb client: %w", err)
 	}

metricbeat/module/mongodb/status/status_integration_test.go

@@ -30,7 +30,6 @@ import (
 
 func TestFetch(t *testing.T) {
 	service := compose.EnsureUp(t, "mongodb")
-
 	f := mbtest.NewReportingMetricSetV2Error(t, getConfig(service.Host()))
 	events, errs := mbtest.ReportingFetchV2Error(f)
 
@@ -52,7 +51,7 @@ func TestFetch(t *testing.T) {
 	assert.True(t, available.(int32) > 0)
 
 	pageFaults, _ := event.GetValue("mongodb.status.extra_info.page_faults")
-	assert.True(t, pageFaults.(int32) >= 0)
+	assert.True(t, pageFaults.(int64) >= 0)
 }
 
 func TestData(t *testing.T) {