PMM-12522 collect chunks metrics in low resolution (#849)

BupycHuk · web-flow · commit e39e36fce0bc · 2024-05-29T16:38:34.000+03:00
* PMM-12522 add component to logs. * PMM-12522 log shards DSNs. * PMM-12522 New chunks collector. * PMM-12522 Dropped new chunks collector and moved data to shards collector. * PMM-12522 Shards collector runs only for mongos instance. * PMM-12522 Fix tests. * PMM-12522 Fix linter. * PMM-12522 Cleanup. * PMM-12522 Cleanup. * PMM-12522 Cleanup.
diff --git a/docker-compose.yml b/docker-compose.yml
@@ -5,7 +5,7 @@ services:
         image: ${TEST_MONGODB_IMAGE:-mongo:4.2}
         ports:
             - "${TEST_MONGODB_S1_PRIMARY_PORT:-17001}:27017"
-        command: mongod --replSet rs1 --shardsvr --port 27017  --oplogSize 16
+        command: mongod --replSet rs1 --shardsvr --port 27017  --oplogSize 16  --bind_ip 0.0.0.0
         networks:
             - rs1
         depends_on:
@@ -18,7 +18,7 @@ services:
         image: ${TEST_MONGODB_IMAGE:-mongo:4.2}
         ports:
             - "${TEST_MONGODB_S1_SECONDARY1_PORT:-17002}:27017"
-        command: mongod --replSet rs1 --shardsvr --port 27017  --oplogSize 16
+        command: mongod --replSet rs1 --shardsvr --port 27017  --oplogSize 16  --bind_ip 0.0.0.0
         networks:
             - rs1
 
@@ -27,7 +27,7 @@ services:
         image: ${TEST_MONGODB_IMAGE:-mongo:4.2}
         ports:
             - "${TEST_MONGODB_S1_SECONDARY2_PORT:-17003}:27017"
-        command: mongod --replSet rs1 --shardsvr --port 27017  --oplogSize 16
+        command: mongod --replSet rs1 --shardsvr --port 27017  --oplogSize 16  --bind_ip 0.0.0.0
         networks:
             - rs1
 
@@ -192,6 +192,9 @@ services:
             - "mongos"
         networks:
             - mongo-shard
+            - rs1
+            - rs2
+            - cnf-serv
         volumes:
             - ./docker/scripts:/scripts
         environment:
diff --git a/docker/scripts/init-shard.sh b/docker/scripts/init-shard.sh
@@ -37,6 +37,10 @@ done
 echo "Started.."
 
 echo init-shard.sh time now: `date +"%T" `
+echo "Configuring sharding.."
+echo  "${RS1}/${mongodb11}:${PORT1},${mongodb12}:${PORT2},${mongodb13}:${PORT3}"
+echo  "${RS2}/${mongodb21}:${PORT1},${mongodb22}:${PORT2},${mongodb23}:${PORT3}"
+
 ${MONGODB_CLIENT} --host ${mongodb1}:${port} <<EOF
    sh.addShard( "${RS1}/${mongodb11}:${PORT1},${mongodb12}:${PORT2},${mongodb13}:${PORT3}" );
    sh.addShard( "${RS2}/${mongodb21}:${PORT1},${mongodb22}:${PORT2},${mongodb23}:${PORT3}" );
diff --git a/exporter/base_collector.go b/exporter/base_collector.go
@@ -26,14 +26,14 @@ import (
 
 type baseCollector struct {
 	client *mongo.Client
-	logger *logrus.Logger
+	logger *logrus.Entry
 
 	lock         sync.Mutex
 	metricsCache []prometheus.Metric
 }
 
 // newBaseCollector creates a skeletal collector, which is used to create other collectors.
-func newBaseCollector(client *mongo.Client, logger *logrus.Logger) *baseCollector {
+func newBaseCollector(client *mongo.Client, logger *logrus.Entry) *baseCollector {
 	return &baseCollector{
 		client: client,
 		logger: logger,
diff --git a/exporter/collstats_collector.go b/exporter/collstats_collector.go
@@ -40,7 +40,7 @@ type collstatsCollector struct {
 func newCollectionStatsCollector(ctx context.Context, client *mongo.Client, logger *logrus.Logger, compatible, discovery bool, topology labelsGetter, collections []string) *collstatsCollector {
 	return &collstatsCollector{
 		ctx:  ctx,
-		base: newBaseCollector(client, logger),
+		base: newBaseCollector(client, logger.WithFields(logrus.Fields{"collector": "collstats"})),
 
 		compatibleMode:  compatible,
 		discoveringMode: discovery,
@@ -79,7 +79,6 @@ func (d *collstatsCollector) collect(ch chan<- prometheus.Metric) {
 		collections, err = checkNamespacesForViews(d.ctx, client, d.collections)
 		if err != nil {
 			logger.Errorf("cannot list collections: %s", err.Error())
-
 			return
 		}
 	}
diff --git a/exporter/currentop_collector.go b/exporter/currentop_collector.go
@@ -44,7 +44,7 @@ func newCurrentopCollector(ctx context.Context, client *mongo.Client, logger *lo
 ) *currentopCollector {
 	return &currentopCollector{
 		ctx:               ctx,
-		base:              newBaseCollector(client, logger),
+		base:              newBaseCollector(client, logger.WithFields(logrus.Fields{"collector": "currentop"})),
 		compatibleMode:    compatible,
 		topologyInfo:      topology,
 		currentopslowtime: currentOpSlowTime,
diff --git a/exporter/dbstats_collector.go b/exporter/dbstats_collector.go
@@ -40,7 +40,7 @@ type dbstatsCollector struct {
 func newDBStatsCollector(ctx context.Context, client *mongo.Client, logger *logrus.Logger, compatible bool, topology labelsGetter, databaseRegex []string, freeStorage bool) *dbstatsCollector {
 	return &dbstatsCollector{
 		ctx:  ctx,
-		base: newBaseCollector(client, logger),
+		base: newBaseCollector(client, logger.WithFields(logrus.Fields{"collector": "dbstats"})),
 
 		compatibleMode: compatible,
 		topologyInfo:   topology,
diff --git a/exporter/debug.go b/exporter/debug.go
@@ -23,8 +23,8 @@ import (
 	"github.com/sirupsen/logrus"
 )
 
-func debugResult(log *logrus.Logger, m interface{}) {
-	if !log.IsLevelEnabled(logrus.DebugLevel) {
+func debugResult(log *logrus.Entry, m interface{}) {
+	if !log.Logger.IsLevelEnabled(logrus.DebugLevel) {
 		return
 	}
 
diff --git a/exporter/debug_test.go b/exporter/debug_test.go
@@ -55,7 +55,7 @@ func TestDebug(t *testing.T) {
   }
 }` + "\n"
 
-	debugResult(log, m)
+	debugResult(log.WithField("component", "test"), m)
 	assert.NoError(t, w.Close())
 	out, _ := io.ReadAll(r)
 
diff --git a/exporter/diagnostic_data_collector.go b/exporter/diagnostic_data_collector.go
@@ -43,7 +43,7 @@ type diagnosticDataCollector struct {
 func newDiagnosticDataCollector(ctx context.Context, client *mongo.Client, logger *logrus.Logger, compatible bool, topology labelsGetter) *diagnosticDataCollector {
 	return &diagnosticDataCollector{
 		ctx:  ctx,
-		base: newBaseCollector(client, logger),
+		base: newBaseCollector(client, logger.WithFields(logrus.Fields{"collector": "diagnostic_data"})),
 
 		compatibleMode: compatible,
 		topologyInfo:   topology,
@@ -102,6 +102,7 @@ func (d *diagnosticDataCollector) collect(ch chan<- prometheus.Metric) {
 	}
 
 	if d.compatibleMode {
+		logger.Debug("running special metrics for compatibility mode")
 		metrics = append(metrics, specialMetrics(d.ctx, client, m, logger)...)
 
 		if cem, err := cacheEvictedTotalMetric(m); err == nil {
@@ -114,6 +115,7 @@ func (d *diagnosticDataCollector) collect(ch chan<- prometheus.Metric) {
 				"component": "diagnosticDataCollector",
 			}).Errorf("Cannot get node type to check if this is a mongos: %s", err)
 		} else if nodeType == typeMongos {
+			logger.Debug("running special metrics for mongos")
 			metrics = append(metrics, mongosMetrics(d.ctx, client, logger)...)
 		}
 	}
diff --git a/exporter/exporter.go b/exporter/exporter.go
@@ -229,7 +229,7 @@ func (e *Exporter) makeRegistry(ctx context.Context, client *mongo.Client, topol
 		registry.MustRegister(rsgsc)
 	}
 
-	if e.opts.EnableShards && requestOpts.EnableShards {
+	if e.opts.EnableShards && nodeType == typeMongos && requestOpts.EnableShards {
 		sc := newShardsCollector(ctx, client, e.opts.Logger, e.opts.CompatibleMode)
 		registry.MustRegister(sc)
 	}
diff --git a/exporter/general_collector.go b/exporter/general_collector.go
@@ -35,7 +35,7 @@ type generalCollector struct {
 func newGeneralCollector(ctx context.Context, client *mongo.Client, logger *logrus.Logger) *generalCollector {
 	return &generalCollector{
 		ctx:  ctx,
-		base: newBaseCollector(client, logger),
+		base: newBaseCollector(client, logger.WithFields(logrus.Fields{"collector": "general"})),
 	}
 }
 
@@ -52,7 +52,7 @@ func (d *generalCollector) collect(ch chan<- prometheus.Metric) {
 	ch <- mongodbUpMetric(d.ctx, d.base.client, d.base.logger)
 }
 
-func mongodbUpMetric(ctx context.Context, client *mongo.Client, log *logrus.Logger) prometheus.Metric {
+func mongodbUpMetric(ctx context.Context, client *mongo.Client, log *logrus.Entry) prometheus.Metric { //nolint:ireturn
 	var value float64
 
 	if client != nil {
diff --git a/exporter/indexstats_collector.go b/exporter/indexstats_collector.go
@@ -41,7 +41,7 @@ type indexstatsCollector struct {
 func newIndexStatsCollector(ctx context.Context, client *mongo.Client, logger *logrus.Logger, discovery, overrideDescendingIndex bool, topology labelsGetter, collections []string) *indexstatsCollector {
 	return &indexstatsCollector{
 		ctx:  ctx,
-		base: newBaseCollector(client, logger),
+		base: newBaseCollector(client, logger.WithFields(logrus.Fields{"collector": "indexstats"})),
 
 		discoveringMode:         discovery,
 		topologyInfo:            topology,
diff --git a/exporter/metrics.go b/exporter/metrics.go
@@ -253,7 +253,6 @@ func makeMetrics(prefix string, m bson.M, labels map[string]string, compatibleMo
 		case map[string]interface{}:
 			res = append(res, makeMetrics(prefix+k, v, labels, compatibleMode)...)
 		case primitive.A:
-			v = []interface{}(v)
 			res = append(res, processSlice(prefix, k, v, labels, compatibleMode)...)
 		case []interface{}:
 			continue
diff --git a/exporter/profile_status_collector.go b/exporter/profile_status_collector.go
@@ -19,7 +19,6 @@ import (
 	"context"
 	"time"
 
-	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/sirupsen/logrus"
 	"go.mongodb.org/mongo-driver/bson"
@@ -41,7 +40,7 @@ func newProfileCollector(ctx context.Context, client *mongo.Client, logger *logr
 ) *profileCollector {
 	return &profileCollector{
 		ctx:            ctx,
-		base:           newBaseCollector(client, logger),
+		base:           newBaseCollector(client, logger.WithFields(logrus.Fields{"collector": "profile"})),
 		compatibleMode: compatible,
 		topologyInfo:   topology,
 		profiletimets:  profileTimeTS,
@@ -65,7 +64,7 @@ func (d *profileCollector) collect(ch chan<- prometheus.Metric) {
 
 	databases, err := databases(d.ctx, client, nil, nil)
 	if err != nil {
-		errors.Wrap(err, "cannot get the database names list")
+		logger.Warnf("cannot get databases: %s", err)
 		return
 	}
 
@@ -79,7 +78,7 @@ func (d *profileCollector) collect(ch chan<- prometheus.Metric) {
 	for _, db := range databases {
 		res, err := client.Database(db).Collection("system.profile").CountDocuments(d.ctx, cmd)
 		if err != nil {
-			errors.Wrapf(err, "cannot read system.profile")
+			logger.Warnf("cannot get profile count for database %s: %s", db, err)
 			break
 		}
 		labels["database"] = db
diff --git a/exporter/replset_status_collector.go b/exporter/replset_status_collector.go
@@ -41,7 +41,7 @@ type replSetGetStatusCollector struct {
 func newReplicationSetStatusCollector(ctx context.Context, client *mongo.Client, logger *logrus.Logger, compatible bool, topology labelsGetter) *replSetGetStatusCollector {
 	return &replSetGetStatusCollector{
 		ctx:  ctx,
-		base: newBaseCollector(client, logger),
+		base: newBaseCollector(client, logger.WithFields(logrus.Fields{"collector": "replset_status"})),
 
 		compatibleMode: compatible,
 		topologyInfo:   topology,
diff --git a/exporter/shards_collector.go b/exporter/shards_collector.go
@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"strings"
 
+	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/sirupsen/logrus"
 	"go.mongodb.org/mongo-driver/bson"
@@ -37,7 +38,7 @@ type shardsCollector struct {
 func newShardsCollector(ctx context.Context, client *mongo.Client, logger *logrus.Logger, compatibleMode bool) *shardsCollector {
 	return &shardsCollector{
 		ctx:        ctx,
-		base:       newBaseCollector(client, logger),
+		base:       newBaseCollector(client, logger.WithFields(logrus.Fields{"collector": "shards"})),
 		compatible: compatibleMode,
 	}
 }
@@ -56,6 +57,26 @@ func (d *shardsCollector) collect(ch chan<- prometheus.Metric) {
 	client := d.base.client
 	logger := d.base.logger
 	prefix := "shards collection chunks"
+	ctx := d.ctx
+
+	metrics := make([]prometheus.Metric, 0)
+	metric, err := chunksTotal(ctx, client)
+	if err != nil {
+		logger.Warnf("cannot create metric for chunks total: %s", err)
+	} else {
+		metrics = append(metrics, metric)
+	}
+
+	ms, err := chunksTotalPerShard(ctx, client)
+	if err != nil {
+		logger.Warnf("cannot create metric for chunks total per shard: %s", err)
+	} else {
+		metrics = append(metrics, ms...)
+	}
+
+	for _, metric := range metrics {
+		ch <- metric
+	}
 
 	databaseNames, err := client.ListDatabaseNames(d.ctx, bson.D{})
 	if err != nil {
@@ -186,4 +207,59 @@ func (d *shardsCollector) getChunksForCollection(row primitive.M) []bson.M {
 	return chunks
 }
 
+func chunksTotal(ctx context.Context, client *mongo.Client) (prometheus.Metric, error) { //nolint:ireturn
+	n, err := client.Database("config").Collection("chunks").CountDocuments(ctx, bson.M{})
+	if err != nil {
+		return nil, errors.Wrap(err, "cannot get total number of chunks")
+	}
+
+	name := "mongodb_mongos_sharding_chunks_total"
+	help := "Total number of chunks"
+
+	d := prometheus.NewDesc(name, help, nil, nil)
+	return prometheus.NewConstMetric(d, prometheus.GaugeValue, float64(n))
+}
+
+func chunksTotalPerShard(ctx context.Context, client *mongo.Client) ([]prometheus.Metric, error) {
+	aggregation := bson.D{
+		{Key: "$group", Value: bson.M{"_id": "$shard", "count": bson.M{"$sum": 1}}},
+	}
+
+	cursor, err := client.Database("config").Collection("chunks").Aggregate(ctx, mongo.Pipeline{aggregation})
+	if err != nil {
+		return nil, errors.Wrap(err, "cannot get $shards cursor for collection config.chunks")
+	}
+
+	var shards []bson.M
+	if err = cursor.All(ctx, &shards); err != nil {
+		return nil, errors.Wrap(err, "cannot get $shards for collection config.chunks")
+	}
+
+	metrics := make([]prometheus.Metric, 0, len(shards))
+
+	for _, shard := range shards {
+		help := "Total number of chunks per shard"
+		id, ok := shard["_id"].(string)
+		if !ok {
+			continue
+		}
+		labels := map[string]string{"shard": id}
+
+		d := prometheus.NewDesc("mongodb_mongos_sharding_shard_chunks_total", help, nil, labels)
+		val, ok := shard["count"].(int32)
+		if !ok {
+			continue
+		}
+
+		metric, err := prometheus.NewConstMetric(d, prometheus.GaugeValue, float64(val))
+		if err != nil {
+			continue
+		}
+
+		metrics = append(metrics, metric)
+	}
+
+	return metrics, nil
+}
+
 var _ prometheus.Collector = (*shardsCollector)(nil)
diff --git a/exporter/top_collector.go b/exporter/top_collector.go
@@ -41,7 +41,7 @@ func newTopCollector(ctx context.Context, client *mongo.Client, logger *logrus.L
 ) *topCollector {
 	return &topCollector{
 		ctx:            ctx,
-		base:           newBaseCollector(client, logger),
+		base:           newBaseCollector(client, logger.WithFields(logrus.Fields{"collector": "top"})),
 		compatibleMode: compatible,
 		topologyInfo:   topology,
 	}
diff --git a/exporter/topology_info.go b/exporter/topology_info.go
@@ -57,7 +57,7 @@ type topologyInfo struct {
 	// by a new connector, able to reconnect if needed. In case of reconnection, we should
 	// call loadLabels to refresh the labels because they might have changed
 	client *mongo.Client
-	logger *logrus.Logger
+	logger *logrus.Entry
 	rw     sync.RWMutex
 	labels map[string]string
 }
@@ -68,7 +68,7 @@ var ErrCannotGetTopologyLabels = fmt.Errorf("cannot get topology labels")
 func newTopologyInfo(ctx context.Context, client *mongo.Client, logger *logrus.Logger) *topologyInfo {
 	ti := &topologyInfo{
 		client: client,
-		logger: logger,
+		logger: logger.WithFields(logrus.Fields{"component": "topology_info"}),
 		labels: make(map[string]string),
 		rw:     sync.RWMutex{},
 	}
diff --git a/exporter/v1_compatibility.go b/exporter/v1_compatibility.go
diff --git a/exporter/v1_compatibility_test.go b/exporter/v1_compatibility_test.go

Original file line number	Diff line number	Diff line change
`@@ -23,8 +23,8 @@ import (`
`23`	`23`	`"github.com/sirupsen/logrus"`
`24`	`24`	`)`
`25`	`25`
`26`		`-func debugResult(log *logrus.Logger, m interface{}) {`
`27`		`- if !log.IsLevelEnabled(logrus.DebugLevel) {`
	`26`	`+func debugResult(log *logrus.Entry, m interface{}) {`
	`27`	`+ if !log.Logger.IsLevelEnabled(logrus.DebugLevel) {`
`28`	`28`	`return`
`29`	`29`	`}`
`30`	`30`
Original file line number	Diff line number	Diff line change
`@@ -55,7 +55,7 @@ func TestDebug(t *testing.T) {`
`55`	`55`	`}`
`56`	`56`	}` + "\n"
`57`	`57`
`58`		`- debugResult(log, m)`
	`58`	`+ debugResult(log.WithField("component", "test"), m)`
`59`	`59`	`assert.NoError(t, w.Close())`
`60`	`60`	`out, _ := io.ReadAll(r)`
`61`	`61`
Original file line number	Diff line number	Diff line change
`@@ -229,7 +229,7 @@ func (e Exporter) makeRegistry(ctx context.Context, client mongo.Client, topol`
`229`	`229`	`registry.MustRegister(rsgsc)`
`230`	`230`	`}`
`231`	`231`
`232`		`- if e.opts.EnableShards && requestOpts.EnableShards {`
	`232`	`+ if e.opts.EnableShards && nodeType == typeMongos && requestOpts.EnableShards {`
`233`	`233`	`sc := newShardsCollector(ctx, client, e.opts.Logger, e.opts.CompatibleMode)`
`234`	`234`	`registry.MustRegister(sc)`
`235`	`235`	`}`
Original file line number	Diff line number	Diff line change
`@@ -35,7 +35,7 @@ type generalCollector struct {`
`35`	`35`	`func newGeneralCollector(ctx context.Context, client mongo.Client, logger logrus.Logger) *generalCollector {`
`36`	`36`	`return &generalCollector{`
`37`	`37`	`ctx: ctx,`
`38`		`- base: newBaseCollector(client, logger),`
	`38`	`+ base: newBaseCollector(client, logger.WithFields(logrus.Fields{"collector": "general"})),`
`39`	`39`	`}`
`40`	`40`	`}`
`41`	`41`
`@@ -52,7 +52,7 @@ func (d *generalCollector) collect(ch chan<- prometheus.Metric) {`
`52`	`52`	`ch <- mongodbUpMetric(d.ctx, d.base.client, d.base.logger)`
`53`	`53`	`}`
`54`	`54`
`55`		`-func mongodbUpMetric(ctx context.Context, client mongo.Client, log logrus.Logger) prometheus.Metric {`
	`55`	`+func mongodbUpMetric(ctx context.Context, client mongo.Client, log logrus.Entry) prometheus.Metric { //nolint:ireturn`
`56`	`56`	`var value float64`
`57`	`57`
`58`	`58`	`if client != nil {`
Original file line number	Diff line number	Diff line change
`@@ -41,7 +41,7 @@ func newTopCollector(ctx context.Context, client mongo.Client, logger logrus.L`
`41`	`41`	`) *topCollector {`
`42`	`42`	`return &topCollector{`
`43`	`43`	`ctx: ctx,`
`44`		`- base: newBaseCollector(client, logger),`
	`44`	`+ base: newBaseCollector(client, logger.WithFields(logrus.Fields{"collector": "top"})),`
`45`	`45`	`compatibleMode: compatible,`
`46`	`46`	`topologyInfo: topology,`
`47`	`47`	`}`