[receiver/mongodbreceiver] Added mongodb replica metrics and routing logic for multiple mongodb instances (#37517)

#### Description
- Added mongodb replica metrics and routing logic for multiple mongodb
instances
```
  mongodb.repl_queries_per_sec
  mongodb.repl_inserts_per_sec
  mongodb.repl_commands_per_sec
  mongodb.repl_getmores_per_sec
  mongodb.repl_deletes_per_sec
  mongodb.repl_updates_per_sec
```

<!-- Issue number (e.g. #1234) or full URL to issue, if applicable. -->
#### Link to tracking issue
- https://sumologic.atlassian.net/browse/OSC-901

<!--Describe what testing was performed and which tests were added.-->
#### Testing
Screenshot examples:
-
[mongodb.inserts_per_sec](https://github.com/user-attachments/assets/3311b5ce-f969-4ce1-97db-8af4337983e6)
-
[mongodb.queries_per_sec](https://github.com/user-attachments/assets/4a1c00c8-aa32-4198-bcfd-f9952a0d2388)


<!--Describe the documentation added.-->
#### Documentation
-
https://github.com/influxdata/telegraf/tree/master/plugins/inputs/mongodb
-
https://www.mongodb.com/docs/manual/reference/command/serverStatus/#opcounters

<!--Please delete paragraphs that you did not use before submitting.-->

Author: chan-tim-sumo

.chloggen/mongodbReplicaMetrics.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: mongodbreceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Added mongodb replica metrics and routing logic for multiple mongodb instances
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [37517]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: []

receiver/mongodbreceiver/client.go

@@ -26,6 +26,7 @@ type client interface {
 	DBStats(ctx context.Context, DBName string) (bson.M, error)
 	TopStats(ctx context.Context) (bson.M, error)
 	IndexStats(ctx context.Context, DBName, collectionName string) ([]bson.M, error)
+	RunCommand(ctx context.Context, db string, command bson.M) (bson.M, error)
 }
 
 // mongodbClient is a mongodb metric scraper client
@@ -37,12 +38,11 @@ type mongodbClient struct {
 
 // newClient creates a new client to connect and query mongo for the
 // mongodbreceiver
-func newClient(ctx context.Context, config *Config, logger *zap.Logger) (client, error) {
-	driver, err := mongo.Connect(ctx, config.ClientOptions())
+var newClient = func(ctx context.Context, config *Config, logger *zap.Logger, secondary bool) (client, error) {
+	driver, err := mongo.Connect(ctx, config.ClientOptions(secondary))
 	if err != nil {
 		return nil, err
 	}
-
 	return &mongodbClient{
 		cfg:    config,
 		logger: logger,

receiver/mongodbreceiver/client_test.go

@@ -69,6 +69,23 @@ func (fc *fakeClient) IndexStats(ctx context.Context, dbName, collectionName str
 	return args.Get(0).([]bson.M), args.Error(1)
 }
 
+func (fc *fakeClient) RunCommand(ctx context.Context, db string, command bson.M) (bson.M, error) {
+	args := fc.Called(ctx, db, command)
+	if args.Get(0) == nil {
+		return nil, args.Error(1)
+	}
+
+	result, ok := args.Get(0).(bson.M)
+	if !ok {
+		err := errors.New("mock returned invalid type")
+		zap.L().Error("type assertion failed",
+			zap.String("expected", "bson.M"))
+		return nil, err
+	}
+
+	return result, args.Error(1)
+}
+
 func TestListDatabaseNames(t *testing.T) {
 	mont := mtest.New(t, mtest.NewOptions().ClientType(mtest.Mock))
 

receiver/mongodbreceiver/config.go

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"go.mongodb.org/mongo-driver/mongo/options"
+	"go.mongodb.org/mongo-driver/mongo/readpref"
 	"go.opentelemetry.io/collector/config/confignet"
 	"go.opentelemetry.io/collector/config/configopaque"
 	"go.opentelemetry.io/collector/config/configtls"
@@ -59,7 +60,27 @@ func (c *Config) Validate() error {
 	return err
 }
 
-func (c *Config) ClientOptions() *options.ClientOptions {
+func (c *Config) ClientOptions(secondary bool) *options.ClientOptions {
+	if secondary {
+		// For secondary nodes, create a direct connection
+		clientOptions := options.Client().
+			SetHosts(c.hostlist()).
+			SetDirect(true).
+			SetReadPreference(readpref.SecondaryPreferred())
+
+		if c.Timeout > 0 {
+			clientOptions.SetConnectTimeout(c.Timeout)
+		}
+
+		if c.Username != "" && c.Password != "" {
+			clientOptions.SetAuth(options.Credential{
+				Username: c.Username,
+				Password: string(c.Password),
+			})
+		}
+
+		return clientOptions
+	}
 	clientOptions := options.Client()
 	connString := "mongodb://" + strings.Join(c.hostlist(), ",")
 	clientOptions.ApplyURI(connString)

receiver/mongodbreceiver/config_test.go

@@ -166,7 +166,7 @@ func TestOptions(t *testing.T) {
 		ReplicaSet: "rs-1",
 	}
 
-	clientOptions := cfg.ClientOptions()
+	clientOptions := cfg.ClientOptions(false)
 	require.Equal(t, clientOptions.Auth.Username, cfg.Username)
 	require.Equal(t,
 		clientOptions.ConnectTimeout.Milliseconds(),
@@ -192,7 +192,7 @@ func TestOptionsTLS(t *testing.T) {
 			},
 		},
 	}
-	opts := cfg.ClientOptions()
+	opts := cfg.ClientOptions(false)
 	require.NotNil(t, opts.TLSConfig)
 }
 

receiver/mongodbreceiver/documentation.md

@@ -340,6 +340,54 @@ The number of replicated operations executed.
 | ---- | ----------- | ------ |
 | operation | The MongoDB operation being counted. | Str: ``insert``, ``query``, ``update``, ``delete``, ``getmore``, ``command`` |
 
+### mongodb.repl_commands_per_sec
+
+The number of replicated commands executed per second.
+
+| Unit | Metric Type | Value Type |
+| ---- | ----------- | ---------- |
+| {command}/s | Gauge | Double |
+
+### mongodb.repl_deletes_per_sec
+
+The number of replicated deletes executed per second.
+
+| Unit | Metric Type | Value Type |
+| ---- | ----------- | ---------- |
+| {delete}/s | Gauge | Double |
+
+### mongodb.repl_getmores_per_sec
+
+The number of replicated getmores executed per second.
+
+| Unit | Metric Type | Value Type |
+| ---- | ----------- | ---------- |
+| {getmore}/s | Gauge | Double |
+
+### mongodb.repl_inserts_per_sec
+
+The number of replicated insertions executed per second.
+
+| Unit | Metric Type | Value Type |
+| ---- | ----------- | ---------- |
+| {insert}/s | Gauge | Double |
+
+### mongodb.repl_queries_per_sec
+
+The number of replicated queries executed per second.
+
+| Unit | Metric Type | Value Type |
+| ---- | ----------- | ---------- |
+| {query}/s | Gauge | Double |
+
+### mongodb.repl_updates_per_sec
+
+The number of replicated updates executed per second.
+
+| Unit | Metric Type | Value Type |
+| ---- | ----------- | ---------- |
+| {update}/s | Gauge | Double |
+
 ### mongodb.uptime
 
 The amount of time that the server has been running.

receiver/mongodbreceiver/go.mod

@@ -55,6 +55,7 @@ require (
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/snappy v0.0.4 // indirect
 	github.com/google/uuid v1.6.0 // indirect
+	github.com/grpc-ecosystem/grpc-gateway/v2 v2.25.1 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/klauspost/compress v1.17.9 // indirect
 	github.com/knadh/koanf/maps v0.1.1 // indirect
@@ -110,7 +111,7 @@ require (
 	golang.org/x/sync v0.11.0 // indirect
 	golang.org/x/sys v0.30.0 // indirect
 	golang.org/x/text v0.22.0 // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20241202173237-19429a94021a // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20250115164207-1a7da9e5054f // indirect
 	google.golang.org/grpc v1.70.0 // indirect
 	google.golang.org/protobuf v1.36.5 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect