PMM-15198 Indexstats fix.

exporter/collstats_collector.go

@@ -26,7 +26,7 @@
 )
 
 type collstatsCollector struct {
 	ctx  context.Context
 	base *baseCollector
 
 	compatibleMode  bool
@@ -143,14 +143,12 @@
 		debugResult(logger, stats)
 
 		prefix := "collstats"
-		labels := d.topologyInfo.baseLabels()
-		labels["database"] = database
-		labels["collection"] = collection
 
 		for _, metrics := range stats {
-			if shard, ok := metrics["shard"].(string); ok {
-				labels["shard"] = shard
-			}
+			labels := d.topologyInfo.baseLabels()
+			labels["database"] = database
+			labels["collection"] = collection
+			setShardLabel(labels, metrics)
 
 			for _, metric := range makeMetrics(prefix, metrics, labels, d.compatibleMode) {
 				ch <- metric

exporter/collstats_collector_test.go

@@ -50,7 +50,7 @@ func TestCollStatsCollector(t *testing.T) {
 	for i := 0; i < 3; i++ {
 		coll := fmt.Sprintf("testcol_%02d", i)
 		_, err := database.Collection(coll).InsertOne(ctx, bson.M{"f1": 1, "f2": "2"})
-		assert.NoError(t, err)
+		require.NoError(t, err)
 	}
 
 	ti := labelsGetterMock{}
@@ -63,24 +63,24 @@ func TestCollStatsCollector(t *testing.T) {
 	expected := strings.NewReader(`
 # HELP mongodb_collstats_latencyStats_commands_latency collstats.latencyStats.commands.latency
 # TYPE mongodb_collstats_latencyStats_commands_latency untyped
-mongodb_collstats_latencyStats_commands_latency{collection="testcol_00",database="testdb"} 0
-mongodb_collstats_latencyStats_commands_latency{collection="testcol_01",database="testdb"} 0
-mongodb_collstats_latencyStats_commands_latency{collection="testcol_02",database="testdb"} 0
+mongodb_collstats_latencyStats_commands_latency{collection="testcol_00",database="testdb",shard=""} 0
+mongodb_collstats_latencyStats_commands_latency{collection="testcol_01",database="testdb",shard=""} 0
+mongodb_collstats_latencyStats_commands_latency{collection="testcol_02",database="testdb",shard=""} 0
 # HELP mongodb_collstats_latencyStats_transactions_ops collstats.latencyStats.transactions.ops
 # TYPE mongodb_collstats_latencyStats_transactions_ops untyped
-mongodb_collstats_latencyStats_transactions_ops{collection="testcol_00",database="testdb"} 0
-mongodb_collstats_latencyStats_transactions_ops{collection="testcol_01",database="testdb"} 0
-mongodb_collstats_latencyStats_transactions_ops{collection="testcol_02",database="testdb"} 0
+mongodb_collstats_latencyStats_transactions_ops{collection="testcol_00",database="testdb",shard=""} 0
+mongodb_collstats_latencyStats_transactions_ops{collection="testcol_01",database="testdb",shard=""} 0
+mongodb_collstats_latencyStats_transactions_ops{collection="testcol_02",database="testdb",shard=""} 0
 # HELP mongodb_collstats_storageStats_indexSizes collstats.storageStats.indexSizes
 # TYPE mongodb_collstats_storageStats_indexSizes untyped
-mongodb_collstats_storageStats_indexSizes{collection="testcol_00",database="testdb",index_name="_id_"} 4096
-mongodb_collstats_storageStats_indexSizes{collection="testcol_01",database="testdb",index_name="_id_"} 4096
-mongodb_collstats_storageStats_indexSizes{collection="testcol_02",database="testdb",index_name="_id_"} 4096
+mongodb_collstats_storageStats_indexSizes{collection="testcol_00",database="testdb",index_name="_id_",shard=""} 4096
+mongodb_collstats_storageStats_indexSizes{collection="testcol_01",database="testdb",index_name="_id_",shard=""} 4096
+mongodb_collstats_storageStats_indexSizes{collection="testcol_02",database="testdb",index_name="_id_",shard=""} 4096
 # HELP mongodb_collstats_storageStats_capped collstats.storageStats.capped
 # TYPE mongodb_collstats_storageStats_capped untyped
-mongodb_collstats_storageStats_capped{collection="testcol_00",database="testdb"} 0
-mongodb_collstats_storageStats_capped{collection="testcol_01",database="testdb"} 0
-mongodb_collstats_storageStats_capped{collection="testcol_02",database="testdb"} 0` +
+mongodb_collstats_storageStats_capped{collection="testcol_00",database="testdb",shard=""} 0
+mongodb_collstats_storageStats_capped{collection="testcol_01",database="testdb",shard=""} 0
+mongodb_collstats_storageStats_capped{collection="testcol_02",database="testdb",shard=""} 0` +
 		"\n")
 
 	// Filter metrics for 2 reasons:
@@ -97,6 +97,39 @@ mongodb_collstats_storageStats_capped{collection="testcol_02",database="testdb"}
 	assert.NoError(t, err)
 }
 
+// Through mongos, a sharded collection reports one $collStats document per
+// shard. Every series must carry the shard it came from, and a shard must never
+// leak from one document into the next.
+//
+// Not parallel: it enables sharding on the cluster shared with the other tests.
+//
+//nolint:paralleltest
+func TestCollStatsCollectorSharded(t *testing.T) {
+	ctx, cancel := context.WithTimeout(t.Context(), 20*time.Second)
+	defer cancel()
+
+	client := tu.DefaultTestClientMongoS(ctx, t)
+
+	dbName, collName := "testdb_collstats_sharded", "testcol"
+	namespace := dbName + "." + collName
+
+	database := client.Database(dbName)
+	database.Drop(ctx)       //nolint:errcheck
+	defer database.Drop(ctx) //nolint:errcheck
+
+	logger := promslog.New(&promslog.Config{})
+	c := newCollectionStatsCollector(ctx, client, logger, false, labelsGetterMock{}, []string{namespace}, false)
+
+	families := gatherShardedMetrics(ctx, t, client, dbName, collName, c)
+
+	observedShards := make(map[string]struct{})
+	for _, labels := range shardedSeriesLabels(t, families, "mongodb_collstats_", dbName, collName) {
+		observedShards[labels["shard"]] = struct{}{}
+	}
+
+	require.Greater(t, len(observedShards), 1, "collstats metrics were exposed for a single shard only: %v", observedShards)
+}
+
 func TestCollStatsForFakeCountType(t *testing.T) {
 	t.Parallel()
 	ctx, cancel := context.WithTimeout(t.Context(), 3*time.Second)
@@ -143,9 +176,9 @@ func TestCollStatsForFakeCountType(t *testing.T) {
 	expected := strings.NewReader(`
        # HELP mongodb_collstats_storageStats_indexSizes collstats.storageStats.indexSizes
        # TYPE mongodb_collstats_storageStats_indexSizes untyped
-       mongodb_collstats_storageStats_indexSizes{collection="test_collection_account",database="testdb",index_name="_id_"} 4096
-       mongodb_collstats_storageStats_indexSizes{collection="test_collection_account",database="testdb",index_name="test_index_account"} 20480
-       mongodb_collstats_storageStats_indexSizes{collection="test_collection_account",database="testdb",index_name="test_index_count"} 20480
+       mongodb_collstats_storageStats_indexSizes{collection="test_collection_account",database="testdb",index_name="_id_",shard=""} 4096
+       mongodb_collstats_storageStats_indexSizes{collection="test_collection_account",database="testdb",index_name="test_index_account",shard=""} 20480
+       mongodb_collstats_storageStats_indexSizes{collection="test_collection_account",database="testdb",index_name="test_index_count",shard=""} 20480
        `)
 
 	filter := []string{

exporter/common.go

@@ -269,6 +269,17 @@ func splitNamespace(ns string) (string, string) {
 	return parts[0], strings.Join(parts[1:], ".")
 }
 
+// setShardLabel sets the "shard" label from a single $collStats or $indexStats
+// document; documents reporting no shard get an empty value. The label is set
+// unconditionally because descriptors are built from these labels and
+// MustRegister panics when a collector describes one fully-qualified name with
+// two different label sets. Prometheus drops empty labels at ingestion, so
+// deployments without shards keep the same series.
+func setShardLabel(labels map[string]string, doc bson.M) {
+	shard, _ := doc["shard"].(string)
+	labels["shard"] = shard
+}
+
 func fromMapToSlice(databases map[string][]string) []string {
 	var collections []string
 	for db, cols := range databases {

exporter/common_test.go

@@ -22,6 +22,7 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	"go.mongodb.org/mongo-driver/bson"
 	"go.mongodb.org/mongo-driver/mongo"
 
@@ -188,6 +189,49 @@ func TestSplitNamespace(t *testing.T) {
 	}
 }
 
+func TestSetShardLabel(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name string
+		doc  bson.M
+		want map[string]string
+	}{
+		{
+			name: "shard set",
+			doc:  bson.M{"shard": "shard-1"},
+			want: map[string]string{"database": "testdb", "shard": "shard-1"},
+		},
+		{
+			name: "empty shard",
+			doc:  bson.M{"shard": ""},
+			want: map[string]string{"database": "testdb", "shard": ""},
+		},
+		{
+			name: "shard field absent",
+			doc:  bson.M{},
+			want: map[string]string{"database": "testdb", "shard": ""},
+		},
+		{
+			name: "shard field is not a string",
+			doc:  bson.M{"shard": 42},
+			want: map[string]string{"database": "testdb", "shard": ""},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			t.Parallel()
+
+			// A shard from a previous document must not leak into this one.
+			labels := map[string]string{"database": "testdb", "shard": "shard-0"}
+			setShardLabel(labels, test.doc)
+
+			require.Equal(t, test.want, labels)
+		})
+	}
+}
+
 //nolint:paralleltest
 func TestCheckNamespacesForViews(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)

exporter/exporter_test.go

@@ -28,14 +28,27 @@
 	"sync"
 	"testing"
 
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/testutil"
+	dto "github.com/prometheus/client_model/go"
 	"github.com/prometheus/common/promslog"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo"
 
 	"github.com/percona/mongodb_exporter/internal/tu"
 )
 
+const (
+	// minTestShards is the number of shards a cluster needs before a test can
+	// observe metrics coming from more than one of them.
+	minTestShards = 2
+	// shardedTestDocs is large enough that every shard of the test cluster ends up
+	// owning documents of the collection.
+	shardedTestDocs = 100
+)
+
 // Use this for testing because labels like cluster ID are not constant in docker containers
 // so we cannot use the real topology labels in tests.
 type labelsGetterMock struct{}
@@ -48,6 +61,102 @@
 	return nil
 }
 
+// metricLabels flattens the label pairs of a gathered metric into a map.
+func metricLabels(m *dto.Metric) map[string]string {
+	labels := make(map[string]string, len(m.GetLabel()))
+	for _, label := range m.GetLabel() {
+		labels[label.GetName()] = label.GetValue()
+	}
+
+	return labels
+}
+
+// shardTestCollection shards dbName.collName over every shard of the test cluster
+// reached through mongos, so that $collStats and $indexStats report one document
+// per shard. It skips the test only when the cluster itself cannot exercise
+// sharding, meaning it has fewer than two shards.
+func shardTestCollection(ctx context.Context, t *testing.T, client *mongo.Client, dbName, collName string) {
+	t.Helper()
+
+	admin := client.Database("admin")
+
+	var shardList struct {
+		Shards []bson.M `bson:"shards"`
+	}
+	require.NoError(t, admin.RunCommand(ctx, bson.D{{Key: "listShards", Value: 1}}).Decode(&shardList))
+
+	if len(shardList.Shards) < minTestShards {
+		t.Skipf("the test cluster has %d shards, at least %d are needed", len(shardList.Shards), minTestShards)
+	}
+
+	require.NoError(t, admin.RunCommand(ctx, bson.D{{Key: "enableSharding", Value: dbName}}).Err())
+
+	// A hashed shard key on an empty collection presplits the initial chunks and
+	// spreads them over all shards, so every shard owns chunks of the collection.
+	shardCmd := bson.D{
+		{Key: "shardCollection", Value: dbName + "." + collName},
+		{Key: "key", Value: bson.D{{Key: "_id", Value: "hashed"}}},
+	}
+	require.NoError(t, admin.RunCommand(ctx, shardCmd).Err())
+}
+
+// gatherShardedMetrics shards dbName.collName over the test cluster, fills it with documents so
+// that every shard reports on it, and returns what c exposes for it. The collector is expected
+// to be scoped to that one namespace.
+func gatherShardedMetrics(ctx context.Context, t *testing.T, client *mongo.Client,
+	dbName, collName string, c prometheus.Collector,
+) []*dto.MetricFamily {
+	t.Helper()
+
+	shardTestCollection(ctx, t, client, dbName, collName)
+
+	docs := make([]any, 0, shardedTestDocs)
+	for i := range shardedTestDocs {
+		docs = append(docs, bson.M{"f1": i})
+	}
+	_, err := client.Database(dbName).Collection(collName).InsertMany(ctx, docs)
+	require.NoError(t, err)
+
+	// Register runs Describe, which collects everything, and rejects a metric name described
+	// with two different label sets. That is how a shard label set for only some of the
+	// documents would surface.
+	registry := prometheus.NewPedanticRegistry()
+	require.NoError(t, registry.Register(c))
+
+	families, err := registry.Gather()
+	require.NoError(t, err)
+
+	return families
+}
+
+// shardedSeriesLabels returns the labels of every series of dbName.collName exposed by the metric
+// families whose name starts with namePrefix, asserting that each of them carries a shard.
+func shardedSeriesLabels(t *testing.T, families []*dto.MetricFamily,
+	namePrefix, dbName, collName string,
+) []map[string]string {
+	t.Helper()
+
+	var series []map[string]string
+	for _, family := range families {
+		if !strings.HasPrefix(family.GetName(), namePrefix) {
+			continue
+		}
+		for _, metric := range family.GetMetric() {
+			labels := metricLabels(metric)
+			if labels["database"] != dbName || labels["collection"] != collName {
+				continue
+			}
+
+			require.NotEmpty(t, labels["shard"], "series without a shard label: %s %v", family.GetName(), labels)
+			series = append(series, labels)
+		}
+	}
+
+	require.NotEmpty(t, series, "no %s* series for %s.%s", namePrefix, dbName, collName)
+
+	return series
+}
+
 //nolint:funlen
 func TestConnect(t *testing.T) {
 	hostname := "127.0.0.1"
@@ -321,7 +430,7 @@
 				assert.Error(t, err, "Must be unable to connect to %s", tc.URI)
 			}
 
 			e := New(exporterOpts)
 			nodeType, _ := getNodeType(ctx, client)
 			gc := newGeneralCollector(ctx, client, nodeType, e.opts.Logger)
 			r := e.makeRegistry(ctx, client, new(labelsGetterMock), *e.opts)

exporter/indexstats_collector.go

@@ -135,6 +135,7 @@ func (d *indexstatsCollector) collect(ch chan<- prometheus.Metric) {
 			labels["database"] = database
 			labels["collection"] = collection
 			labels["key_name"] = indexName
+			setShardLabel(labels, metric)
 
 			metrics := sanitizeMetrics(metric)
 			for _, metric := range makeMetrics(prefix, metrics, labels, false) {