PMM-12712 Shards collector.

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,105 @@
+// 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"
+	"strings"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/sirupsen/logrus"
+	"go.mongodb.org/mongo-driver/bson"
+	"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
+
+	aggregation := bson.D{
+		{Key: "$group", Value: bson.M{"_id": "$shard", "count": bson.M{"$sum": 1}, "ns": bson.M{"$first": "$ns"}}},
+	}
+	cur, err := client.Database("config").Collection("chunks").Aggregate(context.Background(), mongo.Pipeline{aggregation})
+	if err != nil {
+		logger.Errorf("cannot get $sharded cursor for collection config.chunks: %s", err)
+	}
+
+	var chunks []bson.M
+	err = cur.All(context.Background(), &chunks)
+	if err != nil {
+		logger.Errorf("cannot get $sharded for collection config.chunks: %s", err)
+	}
+
+	logger.Debug("$sharded metrics for config.chunks")
+	debugResult(logger, chunks)
+
+	for _, c := range chunks {
+		var ok bool
+		var id, namespace string
+		if id, ok = c["_id"].(string); !ok {
+			logger.Warning("$sharded chunk with wrong ID found")
+			continue
+		}
+		if namespace, ok = c["ns"].(string); !ok {
+			logger.Warning("not valid namespace in $sharded chunk")
+			continue
+		}
+
+		split := strings.Split(namespace, ".")
+		database := split[0]
+		collection := ""
+		if len(split) >= 2 { //nolint:gomnd
+			collection = strings.Join(split[1:], ".")
+		}
+
+		prefix := "sharded collection chunks"
+		labels := make(map[string]string)
+		labels["database"] = database
+		labels["collection"] = collection
+		labels["shard"] = id
+
+		for _, metric := range makeMetrics(prefix, c, labels, d.compatible) {
+			ch <- metric
+		}
+	}
+}
+
+var _ prometheus.Collector = (*shardedCollector)(nil)

exporter/sharded_collector_test.go

@@ -0,0 +1,49 @@
+// 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"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus/testutil"
+	"github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/assert"
+
+	"github.com/percona/mongodb_exporter/internal/tu"
+)
+
+//nolint:paralleltest
+func TestShardedCollector(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+
+	client := tu.DefaultTestClientMongoS(ctx, t)
+	c := newShardedCollector(ctx, client, logrus.New(), false)
+
+	expected := strings.NewReader(`
+	# HELP mongodb_sharded_collection_chunks_count sharded collection chunks.
+	# TYPE mongodb_sharded_collection_chunks_count counter
+	mongodb_sharded_collection_chunks_count{collection="system.sessions",database="config",shard="rs1"} 512
+    mongodb_sharded_collection_chunks_count{collection="system.sessions",database="config",shard="rs2"} 512` + "\n")
+	filter := []string{
+		"mongodb_sharded_collection_chunks_count",
+	}
+	err := testutil.CollectAndCompare(c, expected, filter...)
+	assert.NoError(t, err)
+}

internal/tu/testutils.go

@@ -71,6 +71,15 @@
 	return TestClient(ctx, port, t)
 }
 
+// DefaultTestClientMongoS returns the mongos MongoDB connection used for tests. It is a direct
+// connection to the mongos server.
+func DefaultTestClientMongoS(ctx context.Context, t *testing.T) *mongo.Client {
+	port, err := PortForContainer("mongos")
+	require.NoError(t, err)
+
+	return TestClient(ctx, port, t)
+}
+
 // GetImageNameForDefault returns image name and version of running
 // default test mongo container.
 func GetImageNameForDefault() (string, string, error) {

main.go

@@ -57,6 +57,7 @@
 	EnableIndexStats         bool `name:"collector.indexstats" help:"Enable collecting metrics from $indexStats"`
 	EnableCollStats          bool `name:"collector.collstats" help:"Enable collecting metrics from $collStats"`
 	EnableProfile            bool `name:"collector.profile" help:"Enable collecting metrics from profile"`
+	EnableSharded            bool `help:"Enable collecting metrics from sharded sharded Mongo about chunks" name:"collector.sharded"`
 
 	EnableOverrideDescendingIndex bool `name:"metrics.overridedescendingindex" help:"Enable descending index name override to replace -1 with _DESC"`
 
@@ -152,6 +153,7 @@
 		EnableIndexStats:         opts.EnableIndexStats,
 		EnableCollStats:          opts.EnableCollStats,
 		EnableProfile:            opts.EnableProfile,
+		EnableSharded:            opts.EnableSharded,
 
 		EnableOverrideDescendingIndex: opts.EnableOverrideDescendingIndex,