metricbeat/module/mongodb/collstats: Add extra collstats metrics (#42171

)

* mongo collStats PoC

* introduce waitgroup

* [metricbeats][mongodb] handle extra collstats metrics

* fix linter errors

* fix imports

* update changelog

* add max and nindexes to collstats data

* update copyright years in NOTICE.txt

* update NOTICE.txt

* impove code readability, add code comments

* replace WaitGroup with errgroup

* run gofumpt to fix imports

* fix loop variable captured by func literal

---------

Co-authored-by: subham sarkar <subham.sarkar@elastic.co>
(cherry picked from commit 0b5cfa9)

CHANGELOG.next.asciidoc

@@ -267,6 +267,10 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - Added Palo Alto Networks module {pull}40686[40686]
 - Restore docker.network.in.* and docker.network.out.* fields in docker module {pull}40968[40968]
 - Collect .NET CLR (IIS) Memory, Exceptions and LocksAndThreads metrics {pull}41929[41929]
+- Added `tier_preference`, `creation_date` and `version` fields to the `elasticsearch.index` metricset. {pull}41944[41944]
+- Add `use_performance_counters` to collect CPU metrics using performance counters on Windows for `system/cpu` and `system/core` {pull}41965[41965]
+- Add support of additional `collstats` metrics in mongodb module. {pull}42171[42171]
+- Preserve queries for debugging when `merge_results: true` in SQL module {pull}42271[42271]
 
 *Metricbeat*
 

metricbeat/docs/fields.asciidoc

@@ -50967,6 +50967,87 @@ type: long
 Number of database commands executed.
 
 
+type: long
+
+--
+
+
+*`mongodb.collstats.stats.stats.size`*::
++
+--
+The total uncompressed size in memory of all records in a collection.
+
+
+type: long
+
+--
+
+*`mongodb.collstats.stats.stats.count`*::
++
+--
+The number of objects or documents in this collection.
+
+
+type: long
+
+--
+
+*`mongodb.collstats.stats.stats.avgObjSize`*::
++
+--
+The average size of an object in the collection (in bytes).
+
+
+type: long
+
+--
+
+*`mongodb.collstats.stats.stats.storageSize`*::
++
+--
+The total amount of storage allocated to this collection for document storage (in bytes).
+
+
+type: long
+
+--
+
+*`mongodb.collstats.stats.stats.totalIndexSize`*::
++
+--
+The total size of all indexes (in bytes).
+
+
+type: long
+
+--
+
+*`mongodb.collstats.stats.stats.totalSize`*::
++
+--
+The sum of the storageSize and totalIndexSize (in bytes).
+
+
+type: long
+
+--
+
+*`mongodb.collstats.stats.stats.max`*::
++
+--
+Shows the maximum number of documents that may be present in a capped collection.
+
+
+type: long
+
+--
+
+*`mongodb.collstats.stats.stats.nindexes`*::
++
+--
+The number of indexes on the collection. All collections have at least one index on the _id field.
+
+
 type: long
 
 --

metricbeat/module/mongodb/collstats/_meta/data.json

@@ -69,11 +69,21 @@
                 "time": {
                     "us": 0
                 }
+            },
+            "stats": {
+                "totalSize": 8192,
+                "max": 5000,
+                "nindexes": 1,
+                "size": 36,
+                "count": 1,
+                "avgObjSize": 36,
+                "storageSize": 4096,
+                "totalIndexSize": 4096
             }
         }
     },
     "service": {
         "address": "172.28.0.5:27017",
         "type": "mongodb"
     }
-}
+}

metricbeat/module/mongodb/collstats/_meta/fields.yml

@@ -102,3 +102,39 @@
       type: long
       description: >
         Number of database commands executed.
+
+    - name: stats
+      type: group
+      fields:
+        - name: stats.size
+          type: long
+          description: >
+            The total uncompressed size in memory of all records in a collection.
+        - name: stats.count
+          type: long
+          description: >
+            The number of objects or documents in this collection.
+        - name: stats.avgObjSize
+          type: long
+          description: >
+            The average size of an object in the collection (in bytes).
+        - name: stats.storageSize
+          type: long
+          description: >
+            The total amount of storage allocated to this collection for document storage (in bytes).
+        - name: stats.totalIndexSize
+          type: long
+          description: >
+            The total size of all indexes (in bytes).
+        - name: stats.totalSize
+          type: long
+          description: >
+            The sum of the storageSize and totalIndexSize (in bytes).
+        - name: stats.max
+          type: long
+          description: >
+            Shows the maximum number of documents that may be present in a capped collection.
+        - name: stats.nindexes
+          type: long
+          description: >
+            The number of indexes on the collection. All collections have at least one index on the _id field.

metricbeat/module/mongodb/collstats/collstats.go

@@ -22,10 +22,13 @@ import (
 	"errors"
 	"fmt"
 
+	"golang.org/x/sync/errgroup"
+
 	"github.com/elastic/beats/v7/metricbeat/mb"
 	"github.com/elastic/beats/v7/metricbeat/module/mongodb"
 
 	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo"
 )
 
 func init() {
@@ -70,10 +73,6 @@ func (m *Metricset) Fetch(reporter mb.ReporterV2) error {
 		}
 	}()
 
-	if err != nil {
-		return fmt.Errorf("could not get a list of databases: %w", err)
-	}
-
 	// This info is only stored in 'admin' database
 	db := client.Database("admin")
 	res := db.RunCommand(context.Background(), bson.D{bson.E{Key: "top"}})
@@ -92,10 +91,19 @@ func (m *Metricset) Fetch(reporter mb.ReporterV2) error {
 
 	totals, ok := result["totals"].(map[string]interface{})
 	if !ok {
-		return errors.New("collection 'totals' are not a map")
+		return errors.New("collection 'totals' is not a map")
+	}
+
+	if err = res.Err(); err != nil {
+		return fmt.Errorf("'top' command failed: %w", err)
 	}
 
+	collStatsErrGroup := &errgroup.Group{}
+	collStatsErrGroup.SetLimit(10) // limit number of goroutines running at the same time
+
 	for group, info := range totals {
+		group := group // make sure it works properly on older Go versions
+
 		if group == "note" {
 			continue
 		}
@@ -106,16 +114,60 @@ func (m *Metricset) Fetch(reporter mb.ReporterV2) error {
 			continue
 		}
 
-		event, err := eventMapping(group, infoMap)
-		if err != nil {
-			reporter.Error(fmt.Errorf("mapping of the event data filed: %w", err))
-			continue
-		}
+		collStatsErrGroup.Go(func() error {
+			names, err := splitKey(group)
+			if err != nil {
+				reporter.Error(fmt.Errorf("splitting a collection key failed: %w", err))
+
+				// the error is captured by reporter. no need to return it (to avoid double reporting of the same error)
+				return nil
+			}
+
+			database, collection := names[0], names[1]
 
-		reporter.Event(mb.Event{
-			MetricSetFields: event,
+			collStats, err := fetchCollStats(client, database, collection)
+			if err != nil {
+				reporter.Error(fmt.Errorf("fetching collStats failed: %w", err))
+
+				// the error is captured by reporter. no need to return it (to avoid double reporting of the same error)
+				return nil
+			}
+
+			infoMap["stats"] = collStats
+
+			event, err := eventMapping(group, infoMap)
+			if err != nil {
+				reporter.Error(fmt.Errorf("mapping of the event data failed: %w", err))
+
+				// the error is captured by reporter. no need to return it (to avoid double reporting of the same error)
+				return nil
+			}
+
+			reporter.Event(mb.Event{
+				MetricSetFields: event,
+			})
+
+			return nil
 		})
 	}
 
+	if err := collStatsErrGroup.Wait(); err != nil {
+		return fmt.Errorf("error processing mongodb collstats: %w", err)
+	}
+
 	return nil
 }
+
+func fetchCollStats(client *mongo.Client, dbName, collectionName string) (map[string]interface{}, error) {
+	db := client.Database(dbName)
+	collStats := db.RunCommand(context.Background(), bson.M{"collStats": collectionName})
+	if err := collStats.Err(); err != nil {
+		return nil, fmt.Errorf("collStats command failed: %w", err)
+	}
+	var statsRes map[string]interface{}
+	if err := collStats.Decode(&statsRes); err != nil {
+		return nil, fmt.Errorf("could not decode mongo response for database=%s, collection=%s: %w", dbName, collectionName, err)
+	}
+
+	return statsRes, nil
+}

metricbeat/module/mongodb/collstats/data.go

@@ -25,15 +25,17 @@ import (
 )
 
 func eventMapping(key string, data mapstr.M) (mapstr.M, error) {
-	names := strings.SplitN(key, ".", 2)
-
-	if len(names) < 2 {
-		return nil, errors.New("collection name invalid")
+	names, err := splitKey(key)
+	if err != nil {
+		return nil, err
 	}
 
+	// NOTE: splitKey handles the case where the collection can have "." in the name
+	database, collection := names[0], names[1]
+
 	event := mapstr.M{
-		"db":         names[0],
-		"collection": names[1],
+		"db":         database,
+		"collection": collection,
 		"name":       key,
 		"total": mapstr.M{
 			"time": mapstr.M{
@@ -91,6 +93,16 @@ func eventMapping(key string, data mapstr.M) (mapstr.M, error) {
 			},
 			"count": mustGetMapStrValue(data, "commands.count"),
 		},
+		"stats": mapstr.M{
+			"size":           mustGetMapStrValue(data, "stats.size"),
+			"count":          mustGetMapStrValue(data, "stats.count"),
+			"avgObjSize":     mustGetMapStrValue(data, "stats.avgObjSize"),
+			"storageSize":    mustGetMapStrValue(data, "stats.storageSize"),
+			"totalIndexSize": mustGetMapStrValue(data, "stats.totalIndexSize"),
+			"totalSize":      mustGetMapStrValue(data, "stats.totalSize"),
+			"max":            mustGetMapStrValue(data, "stats.max"),
+			"nindexes":       mustGetMapStrValue(data, "stats.nindexes"),
+		},
 	}
 
 	return event, nil
@@ -100,3 +112,13 @@ func mustGetMapStrValue(m mapstr.M, key string) interface{} {
 	v, _ := m.GetValue(key)
 	return v
 }
+
+func splitKey(key string) ([]string, error) {
+	dbColl := strings.SplitN(key, ".", 2)
+
+	if len(dbColl) < 2 {
+		return nil, errors.New("collection name invalid")
+	}
+
+	return dbColl, nil
+}

metricbeat/module/mongodb/collstats/data_test.go

@@ -21,7 +21,7 @@ package collstats
 
 import (
 	"encoding/json"
-	"io/ioutil"
+	"os"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -31,7 +31,7 @@ import (
 
 func TestEventMapping(t *testing.T) {
 
-	content, err := ioutil.ReadFile("./_meta/test/input.json")
+	content, err := os.ReadFile("./_meta/test/input.json")
 	assert.NoError(t, err)
 
 	data := mapstr.M{}