PMM-12712 Shards collector.

.github/workflows/go.yml

@@ -22,8 +22,10 @@ jobs:
         image:
           - mongo:4.4
           - mongo:5.0
+          - mongo:6.0
           - percona/percona-server-mongodb:4.4
           - percona/percona-server-mongodb:5.0
+          - percona/percona-server-mongodb:6.0
 
     runs-on: ubuntu-latest
 
@@ -39,6 +41,6 @@ jobs:
       - name: Test
         run: |
           TEST_MONGODB_IMAGE=${{ matrix.image }} make test-cluster
-          sleep 10
+          sleep 300
           make test-race
           make test-cluster-clean

docker-compose.yml

@@ -56,6 +56,7 @@ services:
             - MONGO3=mongo-1-3
             - ARBITER=mongo-1-arbiter
             - RS=rs1
+            - VERSION=${TEST_MONGODB_IMAGE}
         entrypoint: [ "/scripts/setup.sh" ]
         networks:
             - rs1
@@ -112,6 +113,7 @@ services:
             - MONGO3=mongo-2-3
             - ARBITER=mongo-2-arbiter
             - RS=rs2
+            - VERSION=${TEST_MONGODB_IMAGE}
         entrypoint: [ "/scripts/setup.sh" ]
         networks:
             - rs2
@@ -162,6 +164,7 @@ services:
             - MONGO3=mongo-cnf-3
             - RS=cnf-serv
             - PORT=27017
+            - VERSION=${TEST_MONGODB_IMAGE}
         entrypoint: [ "/scripts/setup.sh","cnf_servers" ]
         networks:
             - cnf-serv
@@ -205,6 +208,7 @@ services:
             - PORT1=27017
             - PORT2=27017
             - PORT3=27017
+            - VERSION=${TEST_MONGODB_IMAGE}
         entrypoint: [ "/scripts/init-shard.sh" ]
         restart: on-failure:20
 

docker/scripts/init-shard.sh

@@ -1,5 +1,14 @@
 #!/bin/bash 
 
+readarray -d . -t verarr <<< "${VERSION}"
+readarray -d : -t version <<< "${verarr[0]}"
+echo "Mongo version: ${version[1]}"
+
+if ((${version[1]} <= 4))
+then
+  echo -e 'mongo' > /usr/bin/mongosh && chmod +x /usr/bin/mongosh
+fi
+
 mongodb1=`getent hosts ${MONGOS} | awk '{ print $1 }'`
 
 mongodb11=`getent hosts ${MONGO11} | awk '{ print $1 }'`
@@ -17,16 +26,20 @@ mongodb33=`getent hosts ${MONGO33} | awk '{ print $1 }'`
 port=${PORT:-27017}
 
 echo "Waiting for startup.."
-until mongo --host ${mongodb1}:${port} --eval 'quit(db.runCommand({ ping: 1 }).ok ? 0 : 2)' &>/dev/null; do
+until mongosh --host ${mongodb1}:${port} --eval 'quit(db.runCommand({ ping: 1 }).ok ? 0 : 2)' &>/dev/null; do
   printf '.'
   sleep 1
 done
 
 echo "Started.."
 
 echo init-shard.sh time now: `date +"%T" `
-mongo --host ${mongodb1}:${port} <<EOF
+mongosh --host ${mongodb1}:${port} <<EOF
    sh.addShard( "${RS1}/${mongodb11}:${PORT1},${mongodb12}:${PORT2},${mongodb13}:${PORT3}" );
    sh.addShard( "${RS2}/${mongodb21}:${PORT1},${mongodb22}:${PORT2},${mongodb23}:${PORT3}" );
+   use test;
+   db.createCollection("shard");
+   sh.enableSharding("test");
+   sh.shardCollection( "test.shard", { id: "hashed" }, false, { numInitialChunks: 500, collation: { locale: "simple" }} );
    sh.status();
 EOF

docker/scripts/setup.sh

@@ -1,4 +1,12 @@
 #!/bin/bash 
+readarray -d . -t verarr <<< "${VERSION}"
+readarray -d : -t version <<< "${verarr[0]}"
+echo "Mongo version: ${version[1]}"
+
+if ((${version[1]} <= 4))
+then
+  echo -e 'mongo' > /usr/bin/mongosh && chmod +x /usr/bin/mongosh
+fi
 
 mongodb1=`getent hosts ${MONGO1} | awk '{ print $1 }'`
 mongodb2=`getent hosts ${MONGO2} | awk '{ print $1 }'`
@@ -8,7 +16,7 @@ arbiter=`getent hosts ${ARBITER} | awk '{ print $1 }'`
 port=${PORT:-27017}
 
 echo "Waiting for startup.."
-until mongo --host ${mongodb1}:${port} --eval 'quit(db.runCommand({ ping: 1 }).ok ? 0 : 2)' &>/dev/null; do
+until mongosh --host ${mongodb1}:${port} --eval 'quit(db.runCommand({ ping: 1 }).ok ? 0 : 2)' &>/dev/null; do
   printf '.'
   sleep 1
 done
@@ -20,7 +28,7 @@ echo setup.sh time now: `date +"%T" `
 
 function cnf_servers() {
     echo "setup cnf servers on ${MONGO1}(${mongodb1}:${port})"
-    mongo --host ${mongodb1}:${port} <<EOF
+    mongosh --host ${mongodb1}:${port} <<EOF
     var cfg = {
         "_id": "${RS}",
         "version": 1,
@@ -48,7 +56,7 @@ EOF
 
 function general_servers() {
     echo "setup servers on ${MONGO1}(${mongodb1}:${port})"
-    mongo --host ${mongodb1}:${port} <<EOF
+    mongosh --host ${mongodb1}:${port} <<EOF
     var cfg = {
         "_id": "${RS}",
         "protocolVersion": 1,

exporter/exporter.go

@@ -68,6 +68,7 @@ type Opts struct {
 	EnableIndexStats         bool
 	EnableCollStats          bool
 	EnableProfile            bool
+	EnableSharded            bool
 
 	EnableOverrideDescendingIndex bool
 
@@ -163,6 +164,7 @@ func (e *Exporter) makeRegistry(ctx context.Context, client *mongo.Client, topol
 		e.opts.EnableIndexStats = true
 		e.opts.EnableCurrentopMetrics = true
 		e.opts.EnableProfile = true
+		e.opts.EnableSharded = true
 	}
 
 	// arbiter only have isMaster privileges
@@ -175,6 +177,7 @@ func (e *Exporter) makeRegistry(ctx context.Context, client *mongo.Client, topol
 		e.opts.EnableIndexStats = false
 		e.opts.EnableCurrentopMetrics = false
 		e.opts.EnableProfile = false
+		e.opts.EnableSharded = false
 	}
 
 	// If we manually set the collection names we want or auto discovery is set.
@@ -230,6 +233,11 @@ func (e *Exporter) makeRegistry(ctx context.Context, client *mongo.Client, topol
 		registry.MustRegister(rsgsc)
 	}
 
+	if e.opts.EnableSharded && requestOpts.EnableSharded {
+		sc := newShardedCollector(ctx, client, e.opts.Logger, e.opts.CompatibleMode)
+		registry.MustRegister(sc)
+	}
+
 	return registry
 }
 
@@ -304,6 +312,8 @@ func (e *Exporter) Handler() http.Handler {
 				requestOpts.EnableCollStats = true
 			case "profile":
 				requestOpts.EnableProfile = true
+			case "sharded":
+				requestOpts.EnableSharded = true
 			}
 		}
 

exporter/sharded_collector.go

@@ -0,0 +1,189 @@
+// mongodb_exporter
+// Copyright (C) 2017 Percona LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package exporter
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/sirupsen/logrus"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/bson/primitive"
+	"go.mongodb.org/mongo-driver/mongo"
+)
+
+type shardedCollector struct {
+	ctx        context.Context
+	base       *baseCollector
+	compatible bool
+}
+
+// newShardedCollector creates collector collecting metrics about chunks for sharded Mongo.
+func newShardedCollector(ctx context.Context, client *mongo.Client, logger *logrus.Logger, compatibleMode bool) *shardedCollector {
+	return &shardedCollector{
+		ctx:        ctx,
+		base:       newBaseCollector(client, logger),
+		compatible: compatibleMode,
+	}
+}
+
+func (d *shardedCollector) Describe(ch chan<- *prometheus.Desc) {
+	d.base.Describe(d.ctx, ch, d.collect)
+}
+
+func (d *shardedCollector) Collect(ch chan<- prometheus.Metric) {
+	d.base.Collect(ch)
+}
+
+func (d *shardedCollector) collect(ch chan<- prometheus.Metric) {
+	defer measureCollectTime(ch, "mongodb", "sharded")()
+
+	client := d.base.client
+	logger := d.base.logger
+	prefix := "sharded collection chunks"
+
+	databaseNames, err := client.ListDatabaseNames(d.ctx, bson.D{})
+	if err != nil {
+		logger.Errorf("cannot get database names: %s", err)
+	}
+	for _, database := range databaseNames {
+		collections := d.getCollectionsForDBName(database)
+		for _, row := range collections {
+			if len(row) == 0 {
+				continue
+			}
+
+			var ok bool
+			if _, ok = row["_id"]; !ok {
+				continue
+			}
+			var rowID string
+			if rowID, ok = row["_id"].(string); !ok {
+				continue
+			}
+
+			chunks := d.getChunksForCollection(row)
+			for _, c := range chunks {
+				labels, chunks, success := d.getInfoForChunk(c, database, rowID)
+				if !success {
+					continue
+				}
+				for _, metric := range makeMetrics(prefix, primitive.M{"count": chunks}, labels, d.compatible) {
+					ch <- metric
+				}
+			}
+		}
+	}
+}
+
+func (d *shardedCollector) getInfoForChunk(c primitive.M, database, rowID string) (map[string]string, int32, bool) {
+	var ok bool
+	if _, ok = c["dropped"]; ok {
+		if dropped, ok := c["dropped"].(bool); ok && dropped {
+			return nil, 0, false
+		}
+	}
+
+	labels := make(map[string]string)
+	labels["database"] = database
+	labels["collection"] = strings.Replace(rowID, fmt.Sprintf("%s.", database), "", 1)
+
+	if _, ok = c["shard"]; !ok {
+		return nil, 0, ok
+	}
+	var shard string
+	if shard, ok = c["shard"].(string); !ok {
+		return nil, 0, ok
+	}
+	labels["shard"] = shard
+
+	logger := d.base.logger
+	logger.Debug("$sharded metrics for config.chunks")
+	debugResult(logger, primitive.M{database: c})
+
+	if _, ok = c["nChunks"]; !ok {
+		return nil, 0, ok
+	}
+	var chunks int32
+	if chunks, ok = c["nChunks"].(int32); !ok {
+		return nil, 0, ok
+	}
+
+	return labels, chunks, true
+}
+
+func (d *shardedCollector) getCollectionsForDBName(database string) []primitive.M {
+	client := d.base.client
+	logger := d.base.logger
+
+	cursor := client.Database("config").Collection("collections")
+	rs, err := cursor.Find(d.ctx, bson.M{"_id": bson.M{"$regex": fmt.Sprintf("^%s.", database), "$options": "i"}})
+	if err != nil {
+		logger.Errorf("cannot find _id starting with \"%s.\":%s", database, err)
+		return nil
+	}
+
+	var decoded []bson.M
+	err = rs.All(d.ctx, &decoded)
+	if err != nil {
+		logger.Errorf("cannot decode collections:%s", err)
+		return nil
+	}
+
+	return decoded
+}
+
+func (d *shardedCollector) getChunksForCollection(row primitive.M) []bson.M {
+	var chunksMatchPredicate bson.M
+	if _, ok := row["timestamp"]; ok {
+		if uuid, ok := row["uuid"]; ok {
+			chunksMatchPredicate = bson.M{"uuid": uuid}
+		}
+	} else {
+		if id, ok := row["_id"]; ok {
+			chunksMatchPredicate = bson.M{"_id": id}
+		}
+	}
+
+	aggregation := bson.A{
+		bson.M{"$match": chunksMatchPredicate},
+		bson.M{"$group": bson.M{"_id": "$shard", "cnt": bson.M{"$sum": 1}}},
+		bson.M{"$project": bson.M{"_id": 0, "shard": "$_id", "nChunks": "$cnt"}},
+		bson.M{"$sort": bson.M{"shard": 1}},
+	}
+
+	client := d.base.client
+	logger := d.base.logger
+
+	cur, err := client.Database("config").Collection("chunks").Aggregate(context.Background(), aggregation)
+	if err != nil {
+		logger.Errorf("cannot get $sharded cursor for collection config.chunks: %s", err)
+		return nil
+	}
+
+	var chunks []bson.M
+	err = cur.All(context.Background(), &chunks)
+	if err != nil {
+		logger.Errorf("cannot decode $sharded for collection config.chunks: %s", err)
+		return nil
+	}
+
+	return chunks
+}
+
+var _ prometheus.Collector = (*shardedCollector)(nil)