diff --git a/exporter/collstats_collector.go b/exporter/collstats_collector.go index 32dadc44d..506861211 100644 --- a/exporter/collstats_collector.go +++ b/exporter/collstats_collector.go @@ -143,14 +143,12 @@ func (d *collstatsCollector) collect(ch chan<- prometheus.Metric) { 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 diff --git a/exporter/collstats_collector_test.go b/exporter/collstats_collector_test.go index 491ca2a58..eacac1b94 100644 --- a/exporter/collstats_collector_test.go +++ b/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{ diff --git a/exporter/common.go b/exporter/common.go index 144fb53d9..8b4c22685 100644 --- a/exporter/common.go +++ b/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 { diff --git a/exporter/common_test.go b/exporter/common_test.go index eacd6469a..635630910 100644 --- a/exporter/common_test.go +++ b/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) diff --git a/exporter/exporter_test.go b/exporter/exporter_test.go index be4a6f7fe..4f0202107 100644 --- a/exporter/exporter_test.go +++ b/exporter/exporter_test.go @@ -28,14 +28,27 @@ import ( "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 @@ func (l labelsGetterMock) loadLabels(context.Context) error { 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" diff --git a/exporter/indexstats_collector.go b/exporter/indexstats_collector.go index f1321d736..3ec1cf068 100644 --- a/exporter/indexstats_collector.go +++ b/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) { diff --git a/exporter/indexstats_collector_test.go b/exporter/indexstats_collector_test.go index d53a2cfdd..c16f6ef18 100644 --- a/exporter/indexstats_collector_test.go +++ b/exporter/indexstats_collector_test.go @@ -18,6 +18,7 @@ package exporter import ( "context" "fmt" + "slices" "strings" "testing" "time" @@ -26,6 +27,7 @@ import ( "github.com/prometheus/client_golang/prometheus/testutil" "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/bson/primitive" "go.mongodb.org/mongo-driver/mongo" @@ -50,7 +52,7 @@ func TestIndexStatsCollector(t *testing.T) { collection := fmt.Sprintf("testcol_%02d", i) for j := 0; j < 10; j++ { _, err := database.Collection(collection).InsertOne(ctx, bson.M{"f1": j, "f2": "2"}) - assert.NoError(t, err) + require.NoError(t, err) } mod := mongo.IndexModel{ Keys: bson.M{ @@ -60,7 +62,7 @@ func TestIndexStatsCollector(t *testing.T) { }, } _, err := database.Collection(collection).Indexes().CreateOne(ctx, mod) - assert.NoError(t, err) + require.NoError(t, err) } collection := []string{"testdb.testcol_00", "testdb.testcol_01", "testdb.testcol_02"} @@ -70,12 +72,12 @@ func TestIndexStatsCollector(t *testing.T) { expected := strings.NewReader(` # HELP mongodb_indexstats_accesses_ops indexstats.accesses.ops # TYPE mongodb_indexstats_accesses_ops untyped -mongodb_indexstats_accesses_ops{collection="testcol_00",database="testdb",key_name="_id_"} 0 -mongodb_indexstats_accesses_ops{collection="testcol_00",database="testdb",key_name="idx_01"} 0 -mongodb_indexstats_accesses_ops{collection="testcol_01",database="testdb",key_name="_id_"} 0 -mongodb_indexstats_accesses_ops{collection="testcol_01",database="testdb",key_name="idx_01"} 0 -mongodb_indexstats_accesses_ops{collection="testcol_02",database="testdb",key_name="_id_"} 0 -mongodb_indexstats_accesses_ops{collection="testcol_02",database="testdb",key_name="idx_01"} 0` + +mongodb_indexstats_accesses_ops{collection="testcol_00",database="testdb",key_name="_id_",shard=""} 0 +mongodb_indexstats_accesses_ops{collection="testcol_00",database="testdb",key_name="idx_01",shard=""} 0 +mongodb_indexstats_accesses_ops{collection="testcol_01",database="testdb",key_name="_id_",shard=""} 0 +mongodb_indexstats_accesses_ops{collection="testcol_01",database="testdb",key_name="idx_01",shard=""} 0 +mongodb_indexstats_accesses_ops{collection="testcol_02",database="testdb",key_name="_id_",shard=""} 0 +mongodb_indexstats_accesses_ops{collection="testcol_02",database="testdb",key_name="idx_01",shard=""} 0` + "\n") filter := []string{ @@ -85,6 +87,45 @@ mongodb_indexstats_accesses_ops{collection="testcol_02",database="testdb",key_na assert.NoError(t, err) } +// Through mongos, a sharded collection reports one $indexStats document per +// shard for the same index. Without the shard label they all collapse into one +// series and the duplicates are dropped. +// +// Not parallel: it enables sharding on the cluster shared with the other tests. +// +//nolint:paralleltest +func TestIndexStatsCollectorSharded(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 20*time.Second) + defer cancel() + + client := tu.DefaultTestClientMongoS(ctx, t) + + dbName, collName := "testdb_indexstats_sharded", "testcol" + namespace := dbName + "." + collName + + database := client.Database(dbName) + database.Drop(ctx) //nolint:errcheck + defer database.Drop(ctx) //nolint:errcheck + + c := newIndexStatsCollector(ctx, client, promslog.New(&promslog.Config{}), false, false, labelsGetterMock{}, []string{namespace}) + + families := gatherShardedMetrics(ctx, t, client, dbName, collName, c) + + shardsByIndex := make(map[string][]string) + for _, labels := range shardedSeriesLabels(t, families, "mongodb_indexstats_accesses_ops", dbName, collName) { + indexName := labels["key_name"] + shardsByIndex[indexName] = append(shardsByIndex[indexName], labels["shard"]) + } + + require.Contains(t, shardsByIndex, "_id_") + for indexName, shards := range shardsByIndex { + uniqueShards := slices.Compact(slices.Sorted(slices.Values(shards))) + require.Len(t, uniqueShards, len(shards), + "index %s exposes the same shard more than once: %v", indexName, shards) + require.Greater(t, len(shards), 1, "index %s is exposed for a single shard only: %v", indexName, shards) + } +} + func TestDescendingIndexOverride(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) defer cancel() @@ -101,16 +142,16 @@ func TestDescendingIndexOverride(t *testing.T) { collection := fmt.Sprintf("testcol_%02d", i) for j := 0; j < 10; j++ { _, err := database.Collection(collection).InsertOne(ctx, bson.M{"f1": j, "f2": "2"}) - assert.NoError(t, err) + require.NoError(t, err) } descendingMod := mongo.IndexModel{Keys: bson.M{"f1": -1}} _, err := database.Collection(collection).Indexes().CreateOne(ctx, descendingMod) - assert.NoError(t, err) + require.NoError(t, err) ascendingMod := mongo.IndexModel{Keys: bson.M{"f1": 1}} _, err = database.Collection(collection).Indexes().CreateOne(ctx, ascendingMod) - assert.NoError(t, err) + require.NoError(t, err) } collection := []string{"testdb.testcol_00", "testdb.testcol_01", "testdb.testcol_02"} @@ -120,15 +161,15 @@ func TestDescendingIndexOverride(t *testing.T) { expected := strings.NewReader(` # HELP mongodb_indexstats_accesses_ops indexstats.accesses.ops # TYPE mongodb_indexstats_accesses_ops untyped - mongodb_indexstats_accesses_ops{collection="testcol_00",database="testdb",key_name="_id_"} 0 - mongodb_indexstats_accesses_ops{collection="testcol_00",database="testdb",key_name="f1_1"} 0 - mongodb_indexstats_accesses_ops{collection="testcol_00",database="testdb",key_name="f1_DESC"} 0 - mongodb_indexstats_accesses_ops{collection="testcol_01",database="testdb",key_name="_id_"} 0 - mongodb_indexstats_accesses_ops{collection="testcol_01",database="testdb",key_name="f1_1"} 0 - mongodb_indexstats_accesses_ops{collection="testcol_01",database="testdb",key_name="f1_DESC"} 0 - mongodb_indexstats_accesses_ops{collection="testcol_02",database="testdb",key_name="_id_"} 0 - mongodb_indexstats_accesses_ops{collection="testcol_02",database="testdb",key_name="f1_1"} 0 - mongodb_indexstats_accesses_ops{collection="testcol_02",database="testdb",key_name="f1_DESC"} 0` + "\n") + mongodb_indexstats_accesses_ops{collection="testcol_00",database="testdb",key_name="_id_",shard=""} 0 + mongodb_indexstats_accesses_ops{collection="testcol_00",database="testdb",key_name="f1_1",shard=""} 0 + mongodb_indexstats_accesses_ops{collection="testcol_00",database="testdb",key_name="f1_DESC",shard=""} 0 + mongodb_indexstats_accesses_ops{collection="testcol_01",database="testdb",key_name="_id_",shard=""} 0 + mongodb_indexstats_accesses_ops{collection="testcol_01",database="testdb",key_name="f1_1",shard=""} 0 + mongodb_indexstats_accesses_ops{collection="testcol_01",database="testdb",key_name="f1_DESC",shard=""} 0 + mongodb_indexstats_accesses_ops{collection="testcol_02",database="testdb",key_name="_id_",shard=""} 0 + mongodb_indexstats_accesses_ops{collection="testcol_02",database="testdb",key_name="f1_1",shard=""} 0 + mongodb_indexstats_accesses_ops{collection="testcol_02",database="testdb",key_name="f1_DESC",shard=""} 0` + "\n") filter := []string{ "mongodb_indexstats_accesses_ops",