PMM-9400 Fix filters for PMM-9312 (#402) (#408)

* PMM-9400 Fix filters for PMM-9312

* Added a comment

* Improvement for CR

* Updates for CR

* Fixes for CR

* Changes for CR

* Changes for CR

Author: percona-csalguero

exporter/collstats_collector.go

@@ -46,7 +46,7 @@ func (d *collstatsCollector) Collect(ch chan<- prometheus.Metric) {
 	if d.discoveringMode {
 		namespaces, err := listAllCollections(d.ctx, d.client, d.collections, systemDBs)
 		if err != nil {
-			d.logger.Errorf("cannot auto discover databases and collections")
+			d.logger.Errorf("cannot auto discover databases and collections: %s", err.Error())
 
 			return
 		}

exporter/common.go

@@ -91,42 +91,93 @@ func makeExcludeFilter(exclude []string) *primitive.E {
 	return &primitive.E{Key: "$and", Value: filterExpressions}
 }
 
-// makeDBsFilter creates a filter to list all databases or only the ones in the specified
-// namespaces list. Namespaces have the form of <db>.<collection> and the collection name
-// can have a dot. Example: db1.collection.one -> db: db1, collection: collection.one
-// db1, db2.col2, db3.col.one will produce [db1, db2, db3].
 func makeDBsFilter(filterInNamespaces []string) *primitive.E {
-	dbs := []string{}
-	if len(dbs) == 0 {
+	filterExpressions := []bson.D{}
+
+	nss := removeEmptyStrings(filterInNamespaces)
+	for _, namespace := range nss {
+		parts := strings.Split(namespace, ".")
+		filterExpressions = append(filterExpressions,
+			bson.D{{Key: "name", Value: bson.D{{Key: "$eq", Value: parts[0]}}}},
+		)
+	}
+
+	if len(filterExpressions) == 0 {
 		return nil
 	}
 
-	for _, namespace := range filterInNamespaces {
-		parts := strings.Split(namespace, ".")
-		dbs = append(dbs, parts[0])
+	return &primitive.E{Key: "$or", Value: filterExpressions}
+}
+
+func removeEmptyStrings(items []string) []string {
+	cleanList := []string{}
+
+	for _, item := range items {
+		if item == "" {
+			continue
+		}
+		cleanList = append(cleanList, item)
 	}
 
-	return &primitive.E{Key: "name", Value: primitive.E{Key: "$in", Value: dbs}}
+	return cleanList
+}
+
+func unique(slice []string) []string {
+	keys := make(map[string]bool)
+	list := []string{}
+
+	for _, entry := range slice {
+		if _, ok := keys[entry]; !ok {
+			keys[entry] = true
+			list = append(list, entry)
+		}
+	}
+
+	return list
 }
 
 func listAllCollections(ctx context.Context, client *mongo.Client, filterInNamespaces []string, excludeDBs []string) (map[string][]string, error) {
 	namespaces := make(map[string][]string)
 
-	for _, namespace := range filterInNamespaces {
-		parts := strings.Split(namespace, ".")
-		dbname := parts[0]
+	dbs, err := databases(ctx, client, filterInNamespaces, excludeDBs)
+	if err != nil {
+		return nil, errors.Wrap(err, "cannot make the list of databases to list all collections")
+	}
 
-		colls, err := listCollections(ctx, client, dbname, []string{namespace})
-		if err != nil {
-			return nil, errors.Wrapf(err, "cannot list the collections for %q", dbname)
-		}
+	filterNS := removeEmptyStrings(filterInNamespaces)
+
+	// If there are no specified namespaces to search for collections, it means all dbs should be included.
+	if len(filterNS) == 0 {
+		filterNS = append(filterNS, dbs...)
+	}
+
+	for _, db := range dbs {
+		for _, namespace := range filterNS {
+			parts := strings.Split(namespace, ".")
+			dbname := strings.TrimSpace(parts[0])
 
-		if _, ok := namespaces[dbname]; !ok {
-			namespaces[dbname] = colls
-		} else {
-			namespaces[dbname] = append(namespaces[dbname], colls...)
+			if dbname == "" || dbname != db {
+				continue
+			}
+
+			colls, err := listCollections(ctx, client, db, []string{namespace})
+			if err != nil {
+				return nil, errors.Wrapf(err, "cannot list the collections for %q", db)
+			}
+
+			if _, ok := namespaces[db]; !ok {
+				namespaces[db] = []string{}
+			}
+
+			namespaces[db] = append(namespaces[db], colls...)
 		}
-		sort.Strings(namespaces[dbname]) // make it testeable.
+	}
+
+	// Make it testable.
+	for db, colls := range namespaces {
+		uc := unique(colls)
+		sort.Strings(uc)
+		namespaces[db] = uc
 	}
 
 	return namespaces, nil

exporter/common_test.go

@@ -8,65 +8,127 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo"
 
 	"github.com/percona/mongodb_exporter/internal/tu"
 )
 
-func TestListCollections(t *testing.T) {
-	t.Parallel()
-	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
-	defer cancel()
-
-	client := tu.DefaultTestClient(ctx, t)
-
-	inDBs := []string{"testdb01", "testdb02"}
-	inColls := []string{"col01", "col02", "colxx", "colyy"}
+//nolint:gochecknoglobals
+var (
+	testDBs   = []string{"testdb01", "testdb02"}
+	testColls = []string{"col01", "col02", "colxx", "colyy"}
+)
 
-	defer func() {
-		for _, dbname := range inDBs {
-			client.Database(dbname).Drop(ctx) //nolint:errcheck
-		}
-	}()
+func setupDB(ctx context.Context, t *testing.T, client *mongo.Client) {
+	t.Helper()
 
-	for _, dbname := range inDBs {
-		for _, coll := range inColls {
+	for _, dbname := range testDBs {
+		for _, coll := range testColls {
 			for j := 0; j < 10; j++ {
 				_, err := client.Database(dbname).Collection(coll).InsertOne(ctx, bson.M{"f1": j, "f2": "2"})
 				assert.NoError(t, err)
 			}
 		}
 	}
+}
+
+func cleanupDB(ctx context.Context, client *mongo.Client) {
+	for _, dbname := range testDBs {
+		client.Database(dbname).Drop(ctx) //nolint:errcheck
+	}
+}
 
-	want := []string{"admin", "config", "local", "testdb01", "testdb02"}
-	allDBs, err := databases(ctx, client, nil, nil)
-	assert.NoError(t, err)
-	assert.Equal(t, want, allDBs)
+//nolint:paralleltest
+func TestListDatabases(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
 
-	want = []string{"col01", "col02", "colxx"}
-	inNameSpaces := []string{inDBs[0] + ".col0", inDBs[0] + ".colx"}
-	colls, err := listCollections(ctx, client, inDBs[0], inNameSpaces)
-	sort.Strings(colls)
+	client := tu.DefaultTestClient(ctx, t)
 
-	assert.NoError(t, err)
-	assert.Equal(t, want, colls)
+	setupDB(ctx, t, client)
+	defer cleanupDB(ctx, client)
 
-	// Advanced filtering test
-	wantNS := map[string][]string{
-		"testdb01": {"col01", "col02", "colxx", "colyy"},
-		"testdb02": {"col01", "col02"},
-	}
+	t.Run("Empty filter in list", func(t *testing.T) {
+		want := []string{"testdb01", "testdb02"}
+		allDBs, err := databases(ctx, client, nil, systemDBs)
+		assert.NoError(t, err)
+		assert.Equal(t, want, allDBs)
+	})
 
-	// List all collections in testdb01 (inDBs[0]) but only col01 and col02 from testdb02.
-	filterInNameSpaces := []string{inDBs[0], inDBs[1] + ".col01", inDBs[1] + ".col02"}
-	namespaces, err := listAllCollections(ctx, client, filterInNameSpaces, systemDBs)
-	assert.NoError(t, err)
-	assert.Equal(t, wantNS, namespaces)
+	t.Run("One collection in list", func(t *testing.T) {
+		want := []string{"testdb01"}
+		allDBs, err := databases(ctx, client, []string{"testdb01.col1"}, systemDBs)
+		assert.NoError(t, err)
+		assert.Equal(t, want, allDBs)
+	})
+
+	t.Run("Multiple namespaces in list", func(t *testing.T) {
+		want := []string{"testdb01", "testdb02"}
+		allDBs, err := databases(ctx, client, []string{"testdb01", "testdb02.col2", "testdb02.col1"}, systemDBs)
+		assert.NoError(t, err)
+		assert.Equal(t, want, allDBs)
+	})
+}
+
+//nolint:paralleltest
+func TestListCollections(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+
+	client := tu.DefaultTestClient(ctx, t)
+
+	setupDB(ctx, t, client)
+	defer cleanupDB(ctx, client)
+
+	t.Run("Basic test", func(t *testing.T) {
+		want := []string{"admin", "config", "local", "testdb01", "testdb02"}
+		allDBs, err := databases(ctx, client, nil, nil)
+		assert.NoError(t, err)
+		assert.Equal(t, want, allDBs)
+	})
+
+	t.Run("Filter in databases", func(t *testing.T) {
+		want := []string{"col01", "col02", "colxx"}
+		inNameSpaces := []string{testDBs[0] + ".col0", testDBs[0] + ".colx"}
+		colls, err := listCollections(ctx, client, testDBs[0], inNameSpaces)
+		sort.Strings(colls)
+
+		assert.NoError(t, err)
+		assert.Equal(t, want, colls)
+	})
+
+	t.Run("With namespaces list", func(t *testing.T) {
+		// Advanced filtering test
+		wantNS := map[string][]string{
+			"testdb01": {"col01", "col02", "colxx", "colyy"},
+			"testdb02": {"col01", "col02"},
+		}
+		// List all collections in testdb01 (inDBs[0]) but only col01 and col02 from testdb02.
+		filterInNameSpaces := []string{testDBs[0], testDBs[1] + ".col01", testDBs[1] + ".col02"}
+		namespaces, err := listAllCollections(ctx, client, filterInNameSpaces, systemDBs)
+		assert.NoError(t, err)
+		assert.Equal(t, wantNS, namespaces)
+	})
+
+	t.Run("Empty namespaces list", func(t *testing.T) {
+		wantNS := map[string][]string{
+			"testdb01": {"col01", "col02", "colxx", "colyy"},
+			"testdb02": {"col01", "col02", "colxx", "colyy"},
+		}
+		namespaces, err := listAllCollections(ctx, client, nil, systemDBs)
+		assert.NoError(t, err)
+		assert.Equal(t, wantNS, namespaces)
+	})
 
-	count, err := nonSystemCollectionsCount(ctx, client, nil, nil)
-	assert.NoError(t, err)
-	assert.True(t, count == 8)
+	t.Run("Count basic", func(t *testing.T) {
+		count, err := nonSystemCollectionsCount(ctx, client, nil, nil)
+		assert.NoError(t, err)
+		assert.Equal(t, 8, count)
+	})
 
-	count, err = nonSystemCollectionsCount(ctx, client, nil, []string{inDBs[0] + ".col0", inDBs[0] + ".colx"})
-	assert.NoError(t, err)
-	assert.Equal(t, 6, count)
+	t.Run("Filtered count", func(t *testing.T) {
+		count, err := nonSystemCollectionsCount(ctx, client, nil, []string{testDBs[0] + ".col0", testDBs[0] + ".colx"})
+		assert.NoError(t, err)
+		assert.Equal(t, 6, count)
+	})
 }