SERVER-68877 Update ServerClusterParameterInitializer to work multitenantly

8 files changed, 447 insertions, 189 deletions

diff --git a/src/mongo/db/catalog/collection_catalog.cpp b/src/mongo/db/catalog/collection_catalog.cpp
index 5c0f8dc72ed..8c3a46914df 100644
--- a/src/mongo/db/catalog/collection_catalog.cpp
+++ b/src/mongo/db/catalog/collection_catalog.cpp
@@ -86,6 +86,9 @@ void assertViewCatalogValid(const ViewsForDatabase& viewsForDb) {
             viewsForDb.valid());
 }
 
+const auto maxUuid = UUID::parse("FFFFFFFF-FFFF-FFFF-FFFF-FFFFFFFFFFFF").getValue();
+const auto minUuid = UUID::parse("00000000-0000-0000-0000-000000000000").getValue();
+
 }  // namespace
 
 class IgnoreExternalViewChangesForDatabase {
@@ -347,7 +350,6 @@ CollectionCatalog::iterator::iterator(OperationContext* opCtx,
                                       const DatabaseName& dbName,
                                       const CollectionCatalog& catalog)
     : _opCtx(opCtx), _dbName(dbName), _catalog(&catalog) {
-    auto minUuid = UUID::parse("00000000-0000-0000-0000-000000000000").getValue();
 
     _mapIter = _catalog->_orderedCollections.lower_bound(std::make_pair(_dbName, minUuid));
 
@@ -1346,7 +1348,6 @@ bool CollectionCatalog::checkIfCollectionSatisfiable(UUID uuid, CollectionInfoFn
 }
 
 std::vector<UUID> CollectionCatalog::getAllCollectionUUIDsFromDb(const DatabaseName& dbName) const {
-    auto minUuid = UUID::parse("00000000-0000-0000-0000-000000000000").getValue();
     auto it = _orderedCollections.lower_bound(std::make_pair(dbName, minUuid));
 
     std::vector<UUID> ret;
@@ -1363,8 +1364,6 @@ std::vector<NamespaceString> CollectionCatalog::getAllCollectionNamesFromDb(
     OperationContext* opCtx, const DatabaseName& dbName) const {
     invariant(opCtx->lockState()->isDbLockedForMode(dbName, MODE_S));
 
-    auto minUuid = UUID::parse("00000000-0000-0000-0000-000000000000").getValue();
-
     std::vector<NamespaceString> ret;
     for (auto it = _orderedCollections.lower_bound(std::make_pair(dbName, minUuid));
          it != _orderedCollections.end() && it->first.first == dbName;
@@ -1376,19 +1375,24 @@ std::vector<NamespaceString> CollectionCatalog::getAllCollectionNamesFromDb(
     return ret;
 }
 
-std::vector<DatabaseName> CollectionCatalog::_getAllDbNamesHelper(DatabaseName firstDbName) const {
-    std::vector<DatabaseName> ret;
-    auto maxUuid = UUID::parse("FFFFFFFF-FFFF-FFFF-FFFF-FFFFFFFFFFFF").getValue();
+Status CollectionCatalog::_iterAllDbNamesHelper(
+    const boost::optional<TenantId>& tenantId,
+    const std::function<Status(const DatabaseName&)>& callback,
+    const std::function<std::pair<DatabaseName, UUID>(const DatabaseName&)>& nextUpperBound) const {
     // _orderedCollections is sorted by <dbName, uuid>. upper_bound will return the iterator to the
     // first element in _orderedCollections greater than <firstDbName, maxUuid>.
-    auto iter = _orderedCollections.upper_bound(std::make_pair(firstDbName, maxUuid));
+    auto iter =
+        _orderedCollections.upper_bound(std::make_pair(DatabaseName(tenantId, ""), maxUuid));
     while (iter != _orderedCollections.end()) {
         auto dbName = iter->first.first;
-        if (firstDbName.tenantId() != boost::none && dbName.tenantId() != firstDbName.tenantId()) {
+        if (tenantId && dbName.tenantId() != tenantId) {
             break;
         }
         if (iter->second->isCommitted()) {
-            ret.push_back(dbName);
+            auto status = callback(dbName);
+            if (!status.isOK()) {
+                return status;
+            }
         } else {
             // If the first collection found for `dbName` is not yet committed, increment the
             // iterator to find the next visible collection (possibly under a different
@@ -1397,18 +1401,42 @@ std::vector<DatabaseName> CollectionCatalog::_getAllDbNamesHelper(DatabaseName f
             continue;
         }
         // Move on to the next database after `dbName`.
-        iter = _orderedCollections.upper_bound(std::make_pair(dbName, maxUuid));
+        iter = _orderedCollections.upper_bound(nextUpperBound(dbName));
     }
-    return ret;
+    return Status::OK();
 }
 
 std::vector<DatabaseName> CollectionCatalog::getAllDbNames() const {
-    return _getAllDbNamesHelper(DatabaseName());
+    return getAllDbNamesForTenant(boost::none);
 }
 
 std::vector<DatabaseName> CollectionCatalog::getAllDbNamesForTenant(
     boost::optional<TenantId> tenantId) const {
-    return _getAllDbNamesHelper(DatabaseName(tenantId, ""));
+    std::vector<DatabaseName> ret;
+    (void)_iterAllDbNamesHelper(
+        tenantId,
+        [&ret](const DatabaseName& dbName) {
+            ret.push_back(dbName);
+            return Status::OK();
+        },
+        [](const DatabaseName& dbName) { return std::make_pair(dbName, maxUuid); });
+    return ret;
+}
+
+std::set<TenantId> CollectionCatalog::getAllTenants() const {
+    std::set<TenantId> ret;
+    (void)_iterAllDbNamesHelper(boost::none,
+                                [&ret](const DatabaseName& dbName) {
+                                    if (const auto& tenantId = dbName.tenantId()) {
+                                        ret.insert(*tenantId);
+                                    }
+                                    return Status::OK();
+                                },
+                                [](const DatabaseName& dbName) {
+                                    return std::make_pair(DatabaseName(dbName.tenantId(), "\xff"),
+                                                          maxUuid);
+                                });
+    return ret;
 }
 
 void CollectionCatalog::setDatabaseProfileSettings(
diff --git a/src/mongo/db/catalog/collection_catalog.h b/src/mongo/db/catalog/collection_catalog.h
index 7b333081b6d..bf561a6f17d 100644
--- a/src/mongo/db/catalog/collection_catalog.h
+++ b/src/mongo/db/catalog/collection_catalog.h
@@ -41,6 +41,7 @@
 #include "mongo/db/storage/durable_catalog_entry.h"
 #include "mongo/db/views/view.h"
 #include "mongo/stdx/unordered_map.h"
+#include "mongo/util/functional.h"
 #include "mongo/util/uuid.h"
 
 namespace mongo {
@@ -468,6 +469,8 @@ public:
      * This function gets all the database names. The result is sorted in alphabetical ascending
      * order.
      *
+     * Callers of this method must hold the global lock in at least MODE_IS.
+     *
      * Unlike DatabaseHolder::getNames(), this does not return databases that are empty.
      */
     std::vector<DatabaseName> getAllDbNames() const;
@@ -476,11 +479,22 @@ public:
      * This function gets all the database names associated with tenantId. The result is sorted in
      * alphabetical ascending order.
      *
+     * Callers of this method must hold the global lock in at least MODE_IS.
+     *
      * Unlike DatabaseHolder::getNames(), this does not return databases that are empty.
      */
     std::vector<DatabaseName> getAllDbNamesForTenant(boost::optional<TenantId> tenantId) const;
 
     /**
+     * This function gets all tenantIds in the database in ascending order.
+     *
+     * Callers of this method must hold the global lock in at least MODE_IS.
+     *
+     * Only returns tenantIds which are attached to at least one non-empty database.
+     */
+    std::set<TenantId> getAllTenants() const;
+
+    /**
      * Sets 'newProfileSettings' as the profiling settings for the database 'dbName'.
      */
     void setDatabaseProfileSettings(const DatabaseName& dbName, ProfileSettings newProfileSettings);
@@ -647,10 +661,18 @@ private:
                                                                   const DatabaseName& dbName) const;
 
     /**
-     * Returns all relevant dbNames using the firstDbName to construct an iterator pointing to the
-     * first desired dbName.
+     * Iterates over databases, and performs a callback on each database. If any callback fails,
+     * returns its error code. If tenantId is set, will iterate only over databases with that
+     * tenantId. nextUpperBound is a callback that controls how we iterate -- given the current
+     * database name, returns a <DatabaseName, UUID> pair which must be strictly less than the next
+     * entry we iterate to.
      */
-    std::vector<DatabaseName> _getAllDbNamesHelper(DatabaseName firstDbName) const;
+    Status _iterAllDbNamesHelper(
+        const boost::optional<TenantId>& tenantId,
+        const std::function<Status(const DatabaseName&)>& callback,
+        const std::function<std::pair<DatabaseName, UUID>(const DatabaseName&)>& nextLowerBound)
+        const;
+
     /**
      * Sets all namespaces used by views for a database. Will uassert if there is a conflicting
      * collection name in the catalog.
diff --git a/src/mongo/idl/SConscript b/src/mongo/idl/SConscript
index c3268b10fd8..9ac6fe00850 100644
--- a/src/mongo/idl/SConscript
+++ b/src/mongo/idl/SConscript
@@ -32,6 +32,7 @@ env.Library(
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/audit',
+        '$BUILD_DIR/mongo/db/catalog/collection_catalog',
         '$BUILD_DIR/mongo/db/dbdirectclient',
         '$BUILD_DIR/mongo/db/repl/replica_set_aware_service',
         '$BUILD_DIR/mongo/db/server_base',
diff --git a/src/mongo/idl/cluster_server_parameter_initializer.cpp b/src/mongo/idl/cluster_server_parameter_initializer.cpp
index 79ce79dc9eb..9361ff661a9 100644
--- a/src/mongo/idl/cluster_server_parameter_initializer.cpp
+++ b/src/mongo/idl/cluster_server_parameter_initializer.cpp
@@ -31,6 +31,8 @@
 
 #include "mongo/base/string_data.h"
 #include "mongo/db/audit.h"
+#include "mongo/db/catalog_raii.h"
+#include "mongo/db/multitenancy_gen.h"
 #include "mongo/db/repl/replica_set_aware_service.h"
 #include "mongo/db/service_context.h"
 #include "mongo/logv2/log.h"
@@ -61,13 +63,15 @@ ClusterServerParameterInitializer* ClusterServerParameterInitializer::get(
 
 void ClusterServerParameterInitializer::updateParameter(OperationContext* opCtx,
                                                         BSONObj doc,
-                                                        StringData mode) {
+                                                        StringData mode,
+                                                        const boost::optional<TenantId>& tenantId) {
     auto nameElem = doc[kIdField];
     if (nameElem.type() != String) {
         LOGV2_DEBUG(6226301,
                     1,
                     "Update with invalid cluster server parameter name",
                     "mode"_attr = mode,
+                    "tenantId"_attr = tenantId,
                     "_id"_attr = nameElem);
         return;
     }
@@ -79,6 +83,7 @@ void ClusterServerParameterInitializer::updateParameter(OperationContext* opCtx,
                     3,
                     "Update to unknown cluster server parameter",
                     "mode"_attr = mode,
+                    "tenantId"_attr = tenantId,
                     "name"_attr = name);
         return;
     }
@@ -89,91 +94,119 @@ void ClusterServerParameterInitializer::updateParameter(OperationContext* opCtx,
                     1,
                     "Update to cluster server parameter has invalid clusterParameterTime",
                     "mode"_attr = mode,
+                    "tenantId"_attr = tenantId,
                     "name"_attr = name,
                     "clusterParameterTime"_attr = cptElem);
         return;
     }
 
     BSONObjBuilder oldValueBob;
-    sp->append(opCtx, &oldValueBob, name.toString(), boost::none);
+    sp->append(opCtx, &oldValueBob, name.toString(), tenantId);
     audit::logUpdateCachedClusterParameter(opCtx->getClient(), oldValueBob.obj(), doc);
 
-    uassertStatusOK(sp->set(doc, boost::none));
+    uassertStatusOK(sp->set(doc, tenantId));
 }
 
 void ClusterServerParameterInitializer::clearParameter(OperationContext* opCtx,
-                                                       ServerParameter* sp) {
-    if (sp->getClusterParameterTime(boost::none) == LogicalTime::kUninitialized) {
+                                                       ServerParameter* sp,
+                                                       const boost::optional<TenantId>& tenantId) {
+    if (sp->getClusterParameterTime(tenantId) == LogicalTime::kUninitialized) {
         // Nothing to clear.
         return;
     }
 
     BSONObjBuilder oldValueBob;
-    sp->append(opCtx, &oldValueBob, sp->name(), boost::none);
+    sp->append(opCtx, &oldValueBob, sp->name(), tenantId);
 
-    uassertStatusOK(sp->reset(boost::none));
+    uassertStatusOK(sp->reset(tenantId));
 
     BSONObjBuilder newValueBob;
-    sp->append(opCtx, &newValueBob, sp->name(), boost::none);
+    sp->append(opCtx, &newValueBob, sp->name(), tenantId);
 
     audit::logUpdateCachedClusterParameter(
         opCtx->getClient(), oldValueBob.obj(), newValueBob.obj());
 }
 
-void ClusterServerParameterInitializer::clearParameter(OperationContext* opCtx, StringData id) {
+void ClusterServerParameterInitializer::clearParameter(OperationContext* opCtx,
+                                                       StringData id,
+                                                       const boost::optional<TenantId>& tenantId) {
     auto* sp = ServerParameterSet::getClusterParameterSet()->getIfExists(id);
     if (!sp) {
         LOGV2_DEBUG(6226303,
                     5,
                     "oplog event deletion of unknown cluster server parameter",
-                    "name"_attr = id);
+                    "name"_attr = id,
+                    "tenantId"_attr = tenantId);
         return;
     }
 
-    clearParameter(opCtx, sp);
+    clearParameter(opCtx, sp, tenantId);
 }
 
-void ClusterServerParameterInitializer::clearAllParameters(OperationContext* opCtx) {
+void ClusterServerParameterInitializer::clearAllTenantParameters(
+    OperationContext* opCtx, const boost::optional<TenantId>& tenantId) {
     const auto& params = ServerParameterSet::getClusterParameterSet()->getMap();
     for (const auto& it : params) {
-        clearParameter(opCtx, it.second);
+        clearParameter(opCtx, it.second, tenantId);
     }
 }
 
-void ClusterServerParameterInitializer::initializeAllParametersFromDisk(OperationContext* opCtx) {
-    doLoadAllParametersFromDisk(
-        opCtx, "initializing"_sd, [this](OperationContext* opCtx, BSONObj doc, StringData mode) {
-            updateParameter(opCtx, doc, mode);
-        });
+void ClusterServerParameterInitializer::initializeAllTenantParametersFromDisk(
+    OperationContext* opCtx, const boost::optional<TenantId>& tenantId) {
+    doLoadAllTenantParametersFromDisk(opCtx,
+                                      "initializing"_sd,
+                                      [this](OperationContext* opCtx,
+                                             const BSONObj& doc,
+                                             StringData mode,
+                                             const boost::optional<TenantId>& tenantId) {
+                                          updateParameter(opCtx, doc, mode, tenantId);
+                                      },
+                                      tenantId);
 }
 
-void ClusterServerParameterInitializer::resynchronizeAllParametersFromDisk(
-    OperationContext* opCtx) {
+void ClusterServerParameterInitializer::resynchronizeAllTenantParametersFromDisk(
+    OperationContext* opCtx, const boost::optional<TenantId>& tenantId) {
     const auto& allParams = ServerParameterSet::getClusterParameterSet()->getMap();
     std::set<std::string> unsetSettings;
     for (const auto& it : allParams) {
         unsetSettings.insert(it.second->name());
     }
 
-    doLoadAllParametersFromDisk(
+    doLoadAllTenantParametersFromDisk(
         opCtx,
         "resynchronizing"_sd,
-        [this, &unsetSettings](OperationContext* opCtx, BSONObj doc, StringData mode) {
+        [this, &unsetSettings](OperationContext* opCtx,
+                               const BSONObj& doc,
+                               StringData mode,
+                               const boost::optional<TenantId>& tenantId) {
             unsetSettings.erase(doc[kIdField].str());
-            updateParameter(opCtx, doc, mode);
-        });
+            updateParameter(opCtx, doc, mode, tenantId);
+        },
+        tenantId);
 
     // For all known settings which were not present in this resync,
     // explicitly clear any value which may be present in-memory.
     for (const auto& setting : unsetSettings) {
-        clearParameter(opCtx, setting);
+        clearParameter(opCtx, setting, tenantId);
     }
 }
 
 void ClusterServerParameterInitializer::onInitialDataAvailable(OperationContext* opCtx,
                                                                bool isMajorityDataAvailable) {
     LOGV2_INFO(6608200, "Initializing cluster server parameters from disk");
-    initializeAllParametersFromDisk(opCtx);
+    if (gMultitenancySupport) {
+        std::set<TenantId> tenantIds;
+        auto catalog = CollectionCatalog::get(opCtx);
+        {
+            Lock::GlobalLock lk(opCtx, MODE_IS);
+            tenantIds = catalog->getAllTenants();
+        }
+
+        for (const auto& tenantId : tenantIds) {
+            initializeAllTenantParametersFromDisk(opCtx, tenantId);
+        }
+    }
+    initializeAllTenantParametersFromDisk(opCtx, boost::none);
 }
 
 }  // namespace mongo
diff --git a/src/mongo/idl/cluster_server_parameter_initializer.h b/src/mongo/idl/cluster_server_parameter_initializer.h
index 15cbd6010c2..14470757ec3 100644
--- a/src/mongo/idl/cluster_server_parameter_initializer.h
+++ b/src/mongo/idl/cluster_server_parameter_initializer.h
@@ -52,22 +52,32 @@ public:
     static ClusterServerParameterInitializer* get(OperationContext* opCtx);
     static ClusterServerParameterInitializer* get(ServiceContext* serviceContext);
 
-    void updateParameter(OperationContext* opCtx, BSONObj doc, StringData mode);
-    void clearParameter(OperationContext* opCtx, ServerParameter* sp);
-    void clearParameter(OperationContext* opCtx, StringData id);
-    void clearAllParameters(OperationContext* opCtx);
+    void updateParameter(OperationContext* opCtx,
+                         BSONObj doc,
+                         StringData mode,
+                         const boost::optional<TenantId>& tenantId);
+    void clearParameter(OperationContext* opCtx,
+                        ServerParameter* sp,
+                        const boost::optional<TenantId>& tenantId);
+    void clearParameter(OperationContext* opCtx,
+                        StringData id,
+                        const boost::optional<TenantId>& tenantId);
+    void clearAllTenantParameters(OperationContext* opCtx,
+                                  const boost::optional<TenantId>& tenantId);
 
     /**
      * Used to initialize in-memory cluster parameter state based on the on-disk contents after
      * startup recovery or initial sync is complete.
      */
-    void initializeAllParametersFromDisk(OperationContext* opCtx);
+    void initializeAllTenantParametersFromDisk(OperationContext* opCtx,
+                                               const boost::optional<TenantId>& tenantId);
 
     /**
      * Used on rollback and rename with drop.
      * Updates settings which are present and clears settings which are not.
      */
-    void resynchronizeAllParametersFromDisk(OperationContext* opCtx);
+    void resynchronizeAllTenantParametersFromDisk(OperationContext* opCtx,
+                                                  const boost::optional<TenantId>& tenantId);
 
     // Virtual methods coming from the ReplicaSetAwareService
     void onStartup(OperationContext* opCtx) override final {}
@@ -85,16 +95,18 @@ public:
 
 private:
     template <typename OnEntry>
-    void doLoadAllParametersFromDisk(OperationContext* opCtx,
-                                     StringData mode,
-                                     OnEntry onEntry) try {
+    void doLoadAllTenantParametersFromDisk(OperationContext* opCtx,
+                                           StringData mode,
+                                           OnEntry onEntry,
+                                           const boost::optional<TenantId>& tenantId) try {
         std::vector<Status> failures;
 
         DBDirectClient client(opCtx);
-        FindCommandRequest findRequest{NamespaceString::kClusterParametersNamespace};
+
+        FindCommandRequest findRequest{NamespaceString::makeClusterParametersNSS(tenantId)};
         client.find(std::move(findRequest), [&](BSONObj doc) {
             try {
-                onEntry(opCtx, doc, mode);
+                onEntry(opCtx, doc, mode, tenantId);
             } catch (const DBException& ex) {
                 failures.push_back(ex.toStatus());
             }
diff --git a/src/mongo/idl/cluster_server_parameter_initializer_test.cpp b/src/mongo/idl/cluster_server_parameter_initializer_test.cpp
index f104f7e21d7..9cddc9d711a 100644
--- a/src/mongo/idl/cluster_server_parameter_initializer_test.cpp
+++ b/src/mongo/idl/cluster_server_parameter_initializer_test.cpp
@@ -27,6 +27,7 @@
  *    it in the license file.
  */
 
+#include "mongo/db/catalog/create_collection.h"
 #include "mongo/db/change_stream_options_manager.h"
 #include "mongo/db/dbdirectclient.h"
 #include "mongo/db/repl/replication_coordinator_mock.h"
@@ -50,21 +51,34 @@ class ClusterServerParameterInitializerTest : public ClusterServerParameterTestB
 public:
     void setUp() final {
         ClusterServerParameterTestBase::setUp();
+        {
+            auto opCtx = cc().makeOperationContext();
+            ASSERT_OK(createCollection(
+                opCtx.get(), CreateCommand(NamespaceString::kClusterParametersNamespace)));
+            ASSERT_OK(createCollection(
+                opCtx.get(), CreateCommand(NamespaceString::makeClusterParametersNSS(kTenantId))));
+        }
 
-        // Insert a document on-disk for ClusterServerParameterTest. This should be loaded in-memory
+        // Insert documents on-disk for ClusterServerParameterTest. This should be loaded in-memory
         // by the initializer during startup recovery and at the end of initial sync.
         Timestamp now(time(nullptr));
-        const auto doc =
-            makeClusterParametersDoc(LogicalTime(now), kInitialIntValue, kInitialStrValue);
+        auto doc = makeClusterParametersDoc(LogicalTime(now), kInitialIntValue, kInitialStrValue);
 
-        upsert(doc);
+        upsert(doc, boost::none);
+
+        doc = makeClusterParametersDoc(
+            LogicalTime(now), kInitialTenantIntValue, kInitialTenantStrValue);
+
+        upsert(doc, kTenantId);
     }
 
     void tearDown() final {
         // Delete all cluster server parameter documents written and refresh in-memory state.
-        remove();
+        remove(boost::none);
+        remove(kTenantId);
         auto opCtx = cc().makeOperationContext();
-        _initializer.resynchronizeAllParametersFromDisk(opCtx.get());
+        _initializer.resynchronizeAllTenantParametersFromDisk(opCtx.get(), boost::none);
+        _initializer.resynchronizeAllTenantParametersFromDisk(opCtx.get(), kTenantId);
     }
     /**
      * Simulates the call to the ClusterServerParameterInitializer at the end of initial sync, when
@@ -97,6 +111,10 @@ TEST_F(ClusterServerParameterInitializerTest, OnInitialSync) {
     ASSERT_EQ(cspTest.getIntValue(), kDefaultIntValue);
     ASSERT_EQ(cspTest.getStrValue(), kDefaultStrValue);
 
+    cspTest = sp->getValue(kTenantId);
+    ASSERT_EQ(cspTest.getIntValue(), kDefaultIntValue);
+    ASSERT_EQ(cspTest.getStrValue(), kDefaultStrValue);
+
     // Indicate that data is available at the end of initial sync and check that the in-memory data
     // is updated.
     doInitialSync();
@@ -105,6 +123,10 @@ TEST_F(ClusterServerParameterInitializerTest, OnInitialSync) {
     cspTest = sp->getValue(boost::none);
     ASSERT_EQ(cspTest.getIntValue(), kInitialIntValue);
     ASSERT_EQ(cspTest.getStrValue(), kInitialStrValue);
+
+    cspTest = sp->getValue(kTenantId);
+    ASSERT_EQ(cspTest.getIntValue(), kInitialTenantIntValue);
+    ASSERT_EQ(cspTest.getStrValue(), kInitialTenantStrValue);
 }
 
 TEST_F(ClusterServerParameterInitializerTest, OnStartupRecovery) {
@@ -123,6 +145,10 @@ TEST_F(ClusterServerParameterInitializerTest, OnStartupRecovery) {
     cspTest = sp->getValue(boost::none);
     ASSERT_EQ(cspTest.getIntValue(), kInitialIntValue);
     ASSERT_EQ(cspTest.getStrValue(), kInitialStrValue);
+
+    cspTest = sp->getValue(kTenantId);
+    ASSERT_EQ(cspTest.getIntValue(), kInitialTenantIntValue);
+    ASSERT_EQ(cspTest.getStrValue(), kInitialTenantStrValue);
 }
 
 }  // namespace
diff --git a/src/mongo/idl/cluster_server_parameter_op_observer.cpp b/src/mongo/idl/cluster_server_parameter_op_observer.cpp
index 44db25c7bff..69f25bed1d0 100644
--- a/src/mongo/idl/cluster_server_parameter_op_observer.cpp
+++ b/src/mongo/idl/cluster_server_parameter_op_observer.cpp
@@ -65,7 +65,8 @@ void ClusterServerParameterOpObserver::onInserts(OperationContext* opCtx,
     }
 
     for (auto it = first; it != last; ++it) {
-        ClusterServerParameterInitializer::get(opCtx)->updateParameter(opCtx, it->doc, kOplog);
+        ClusterServerParameterInitializer::get(opCtx)->updateParameter(
+            opCtx, it->doc, kOplog, coll->ns().dbName().tenantId());
     }
 }
 
@@ -76,7 +77,8 @@ void ClusterServerParameterOpObserver::onUpdate(OperationContext* opCtx,
         return;
     }
 
-    ClusterServerParameterInitializer::get(opCtx)->updateParameter(opCtx, updatedDoc, kOplog);
+    ClusterServerParameterInitializer::get(opCtx)->updateParameter(
+        opCtx, updatedDoc, kOplog, args.nss.dbName().tenantId());
 }
 
 void ClusterServerParameterOpObserver::aboutToDelete(OperationContext* opCtx,
@@ -115,7 +117,8 @@ void ClusterServerParameterOpObserver::onDelete(OperationContext* opCtx,
                                                 const OplogDeleteEntryArgs& args) {
     const auto& docName = aboutToDeleteDoc(opCtx);
     if (!docName.empty()) {
-        ClusterServerParameterInitializer::get(opCtx)->clearParameter(opCtx, docName);
+        ClusterServerParameterInitializer::get(opCtx)->clearParameter(
+            opCtx, docName, tenantIdToDelete(opCtx));
     }
 }
 
@@ -123,7 +126,8 @@ void ClusterServerParameterOpObserver::onDropDatabase(OperationContext* opCtx,
                                                       const DatabaseName& dbName) {
     if (dbName.db() == NamespaceString::kConfigDb) {
         // Entire config DB deleted, reset to default state.
-        ClusterServerParameterInitializer::get(opCtx)->clearAllParameters(opCtx);
+        ClusterServerParameterInitializer::get(opCtx)->clearAllTenantParameters(opCtx,
+                                                                                dbName.tenantId());
     }
 }
 
@@ -135,7 +139,8 @@ repl::OpTime ClusterServerParameterOpObserver::onDropCollection(
     CollectionDropType dropType) {
     if (isConfigNamespace(collectionName)) {
         // Entire collection deleted, reset to default state.
-        ClusterServerParameterInitializer::get(opCtx)->clearAllParameters(opCtx);
+        ClusterServerParameterInitializer::get(opCtx)->clearAllTenantParameters(
+            opCtx, collectionName.dbName().tenantId());
     }
 
     return {};
@@ -150,18 +155,20 @@ void ClusterServerParameterOpObserver::postRenameCollection(
     bool stayTemp) {
     if (isConfigNamespace(fromCollection)) {
         // Same as collection dropped from a config point of view.
-        ClusterServerParameterInitializer::get(opCtx)->clearAllParameters(opCtx);
+        ClusterServerParameterInitializer::get(opCtx)->clearAllTenantParameters(
+            opCtx, fromCollection.dbName().tenantId());
     }
 
     if (isConfigNamespace(toCollection)) {
         // Potentially many documents now set, perform full scan.
         if (dropTargetUUID) {
             // Possibly lost configurations in overwrite.
-            ClusterServerParameterInitializer::get(opCtx)->resynchronizeAllParametersFromDisk(
-                opCtx);
+            ClusterServerParameterInitializer::get(opCtx)->resynchronizeAllTenantParametersFromDisk(
+                opCtx, toCollection.dbName().tenantId());
         } else {
             // Collection did not exist prior to rename.
-            ClusterServerParameterInitializer::get(opCtx)->initializeAllParametersFromDisk(opCtx);
+            ClusterServerParameterInitializer::get(opCtx)->initializeAllTenantParametersFromDisk(
+                opCtx, toCollection.dbName().tenantId());
         }
     }
 }
@@ -177,19 +184,18 @@ void ClusterServerParameterOpObserver::onImportCollection(OperationContext* opCt
     if (!isDryRun && (numRecords > 0) && isConfigNamespace(nss)) {
         // Something was imported, do a full collection scan to sync up.
         // No need to apply rollback rules since nothing will have been deleted.
-        ClusterServerParameterInitializer::get(opCtx)->initializeAllParametersFromDisk(opCtx);
+        ClusterServerParameterInitializer::get(opCtx)->initializeAllTenantParametersFromDisk(
+            opCtx, nss.dbName().tenantId());
     }
 }
 
 void ClusterServerParameterOpObserver::_onReplicationRollback(OperationContext* opCtx,
                                                               const RollbackObserverInfo& rbInfo) {
-    if (rbInfo.rollbackNamespaces.end() !=
-        std::find_if(rbInfo.rollbackNamespaces.begin(),
-                     rbInfo.rollbackNamespaces.end(),
-                     isConfigNamespace)) {
-        // Some kind of rollback happend in the settings collection.
-        // Just reload from disk to be safe.
-        ClusterServerParameterInitializer::get(opCtx)->resynchronizeAllParametersFromDisk(opCtx);
+    for (const auto& nss : rbInfo.rollbackNamespaces) {
+        if (isConfigNamespace(nss)) {
+            ClusterServerParameterInitializer::get(opCtx)->resynchronizeAllTenantParametersFromDisk(
+                opCtx, nss.dbName().tenantId());
+        }
     }
 }
 
diff --git a/src/mongo/idl/cluster_server_parameter_op_observer_test.cpp b/src/mongo/idl/cluster_server_parameter_op_observer_test.cpp
index 87efd29a2a9..b1fd96b3146 100644
--- a/src/mongo/idl/cluster_server_parameter_op_observer_test.cpp
+++ b/src/mongo/idl/cluster_server_parameter_op_observer_test.cpp
@@ -55,6 +55,8 @@ public:
         auto opCtx = makeOperationContext();
         ASSERT_OK(createCollection(opCtx.get(),
                                    CreateCommand(NamespaceString::kClusterParametersNamespace)));
+        ASSERT_OK(createCollection(
+            opCtx.get(), CreateCommand(NamespaceString::makeClusterParametersNSS(kTenantId))));
         for (auto&& nss : kIgnoredNamespaces) {
             ASSERT_OK(createCollection(opCtx.get(), CreateCommand(nss)));
         }
@@ -91,9 +93,9 @@ public:
         observer.onDelete(opCtx.get(), nss, uuid, 1 /* StmtId */, args);
     }
 
-    void doDropDatabase(StringData dbname) {
+    void doDropDatabase(const DatabaseName& dbname) {
         auto opCtx = cc().makeOperationContext();
-        observer.onDropDatabase(opCtx.get(), DatabaseName(boost::none, dbname));
+        observer.onDropDatabase(opCtx.get(), dbname);
     }
 
     void doRenameCollection(const NamespaceString& fromColl, const NamespaceString& toColl) {
@@ -129,28 +131,36 @@ public:
 
     // Asserts that the parameter state does not change for this action.
     template <typename F>
-    void assertIgnored(const NamespaceString& nss, F fn) {
+    void assertIgnored(const NamespaceString& nss,
+                       F fn,
+                       const boost::optional<TenantId>& tenantId) {
         auto* sp =
             ServerParameterSet::getClusterParameterSet()->get<ClusterTestParameter>(kCSPTest);
         ASSERT(sp != nullptr);
 
-        const auto initialCPTime = sp->getClusterParameterTime(boost::none);
-        ClusterServerParameterTest initialCspTest = sp->getValue(boost::none);
+        const auto initialCPTime = sp->getClusterParameterTime(tenantId);
+        ClusterServerParameterTest initialCspTest = sp->getValue(tenantId);
         fn(nss);
-        ClusterServerParameterTest finalCspTest = sp->getValue(boost::none);
+        ClusterServerParameterTest finalCspTest = sp->getValue(tenantId);
 
-        ASSERT_EQ(sp->getClusterParameterTime(boost::none), initialCPTime);
+        ASSERT_EQ(sp->getClusterParameterTime(tenantId), initialCPTime);
         ASSERT_EQ(finalCspTest.getIntValue(), initialCspTest.getIntValue());
         ASSERT_EQ(finalCspTest.getStrValue(), initialCspTest.getStrValue());
     }
 
-    BSONObj initializeState() {
+    std::pair<BSONObj, BSONObj> initializeState() {
         Timestamp now(time(nullptr));
         const auto doc =
             makeClusterParametersDoc(LogicalTime(now), kInitialIntValue, kInitialStrValue);
 
-        upsert(doc);
-        doInserts(NamespaceString::kClusterParametersNamespace, {doc});
+        upsert(doc, boost::none);
+        doInserts(NamespaceString::makeClusterParametersNSS(boost::none), {doc});
+
+        const auto docT = makeClusterParametersDoc(
+            LogicalTime(now), kInitialTenantIntValue, kInitialTenantStrValue);
+
+        upsert(docT, kTenantId);
+        doInserts(NamespaceString::makeClusterParametersNSS(kTenantId), {docT});
 
         auto* sp =
             ServerParameterSet::getClusterParameterSet()->get<ClusterTestParameter>(kCSPTest);
@@ -160,29 +170,58 @@ public:
         ASSERT_EQ(cspTest.getIntValue(), kInitialIntValue);
         ASSERT_EQ(cspTest.getStrValue(), kInitialStrValue);
 
-        return doc;
+        cspTest = sp->getValue(kTenantId);
+        ASSERT_EQ(cspTest.getIntValue(), kInitialTenantIntValue);
+        ASSERT_EQ(cspTest.getStrValue(), kInitialTenantStrValue);
+
+        return {doc, docT};
     }
 
-    // Asserts that a given action is ignore anywhere outside of config.clusterParameters.
+    // Asserts that a given action is ignore anywhere outside of cluster parameter collections.
     template <typename F>
-    void assertIgnoredOtherNamespaces(F fn) {
+    void assertIgnoredOtherNamespaces(F fn, const boost::optional<TenantId>& tenantId) {
         for (const auto& nss : kIgnoredNamespaces) {
-            assertIgnored(nss, fn);
+            assertIgnored(nss, fn, tenantId);
         }
     }
 
-    // Asserts that a given action is ignored anywhere, even on the config.clusterParameters NS.
+    // Asserts that a given action is ignored anywhere, even on cluster parameter collections.
     template <typename F>
-    void assertIgnoredAlways(F fn) {
-        assertIgnoredOtherNamespaces(fn);
-        assertIgnored(NamespaceString::kClusterParametersNamespace, fn);
+    void assertIgnoredAlways(F fn, const boost::optional<TenantId>& tenantId) {
+        assertIgnoredOtherNamespaces(fn, tenantId);
+        assertIgnored(NamespaceString::makeClusterParametersNSS(boost::none), fn, tenantId);
+        assertIgnored(NamespaceString::makeClusterParametersNSS(kTenantId), fn, tenantId);
+    }
+
+    void assertParameterState(int line,
+                              const boost::optional<TenantId>& tenantId,
+                              int intVal,
+                              StringData strVal,
+                              boost::optional<LogicalTime> cpt = boost::none) {
+        auto* sp =
+            ServerParameterSet::getClusterParameterSet()->get<ClusterTestParameter>(kCSPTest);
+        try {
+            if (cpt) {
+                ASSERT_EQ(sp->getClusterParameterTime(tenantId), *cpt);
+            }
+
+            ClusterServerParameterTest cspTest = sp->getValue(tenantId);
+            ASSERT_EQ(cspTest.getIntValue(), intVal);
+            ASSERT_EQ(cspTest.getStrValue(), strVal);
+        } catch (...) {
+            LOGV2_ERROR(6887700, "ASSERT_PARAMETER_STATE failed", "line"_attr = line);
+            throw;
+        }
     }
 
 protected:
     ClusterServerParameterOpObserver observer;
 };
 
+#define ASSERT_PARAMETER_STATE(...) assertParameterState(__LINE__, __VA_ARGS__)
+
 TEST_F(ClusterServerParameterOpObserverTest, OnInsertRecord) {
+    initializeState();
     auto* sp = ServerParameterSet::getClusterParameterSet()->get<ClusterTestParameter>(kCSPTest);
     ASSERT(sp != nullptr);
 
@@ -196,10 +235,8 @@ TEST_F(ClusterServerParameterOpObserverTest, OnInsertRecord) {
     doInserts(NamespaceString::kClusterParametersNamespace,
               {makeClusterParametersDoc(singleLogicalTime, singleIntValue, singleStrValue)});
 
-    ClusterServerParameterTest cspTest = sp->getValue(boost::none);
-    ASSERT_EQ(sp->getClusterParameterTime(boost::none), singleLogicalTime);
-    ASSERT_EQ(cspTest.getIntValue(), singleIntValue);
-    ASSERT_EQ(cspTest.getStrValue(), singleStrValue);
+    ASSERT_PARAMETER_STATE(boost::none, singleIntValue, singleStrValue, singleLogicalTime);
+    ASSERT_PARAMETER_STATE(kTenantId, kInitialTenantIntValue, kInitialTenantStrValue);
 
     // Multi-record insert.
     const auto multiLogicalTime = singleLogicalTime.addTicks(1);
@@ -214,37 +251,54 @@ TEST_F(ClusterServerParameterOpObserverTest, OnInsertRecord) {
                   BSON(ClusterServerParameter::k_idFieldName << "alsoIgnored"),
               });
 
-    cspTest = sp->getValue(boost::none);
-    ASSERT_EQ(sp->getClusterParameterTime(boost::none), multiLogicalTime);
-    ASSERT_EQ(cspTest.getIntValue(), multiIntValue);
-    ASSERT_EQ(cspTest.getStrValue(), multiStrValue);
+    ASSERT_PARAMETER_STATE(boost::none, multiIntValue, multiStrValue, multiLogicalTime);
+    ASSERT_PARAMETER_STATE(kTenantId, kInitialTenantIntValue, kInitialTenantStrValue);
 
     // Insert plausible records to namespaces we don't care about.
-    assertIgnoredOtherNamespaces([this](const auto& nss) {
-        doInserts(nss, {makeClusterParametersDoc(LogicalTime(), 42, "yellow")});
-    });
+    assertIgnoredOtherNamespaces(
+        [this](const auto& nss) {
+            doInserts(nss, {makeClusterParametersDoc(LogicalTime(), 42, "yellow")});
+        },
+        boost::none);
     // Plausible on other NS, multi-insert.
-    assertIgnoredOtherNamespaces([this](const auto& nss) {
-        auto d0 = makeClusterParametersDoc(LogicalTime(), 123, "red");
-        auto d1 = makeClusterParametersDoc(LogicalTime(), 234, "green");
-        auto d2 = makeClusterParametersDoc(LogicalTime(), 345, "blue");
-        doInserts(nss, {d0, d1, d2});
-    });
+    assertIgnoredOtherNamespaces(
+        [this](const auto& nss) {
+            auto d0 = makeClusterParametersDoc(LogicalTime(), 123, "red");
+            auto d1 = makeClusterParametersDoc(LogicalTime(), 234, "green");
+            auto d2 = makeClusterParametersDoc(LogicalTime(), 345, "blue");
+            doInserts(nss, {d0, d1, d2});
+        },
+        boost::none);
 
     // Unknown CSP record ignored on all namespaces.
-    assertIgnoredAlways([this](const auto& nss) {
-        doInserts(nss,
-                  {BSON("_id"
-                        << "ignored")});
-    });
+    assertIgnoredAlways(
+        [this](const auto& nss) {
+            doInserts(nss,
+                      {BSON("_id"
+                            << "ignored")});
+        },
+        boost::none);
     // Unknown CSP, multi-insert.
-    assertIgnoredAlways([this](const auto& nss) {
-        doInserts(nss,
-                  {BSON("_id"
-                        << "ignored"),
-                   BSON("_id"
-                        << "also-ingored")});
-    });
+    assertIgnoredAlways(
+        [this](const auto& nss) {
+            doInserts(nss,
+                      {BSON("_id"
+                            << "ignored"),
+                       BSON("_id"
+                            << "also-ingored")});
+        },
+        boost::none);
+
+    // Insert on separate tenant.
+    const auto tenantLogicalTime = multiLogicalTime.addTicks(1);
+    const auto tenantIntValue = multiIntValue + 1;
+    const auto tenantStrValue = "OnInsertRecord.tenant";
+
+    doInserts(NamespaceString::makeClusterParametersNSS(kTenantId),
+              {makeClusterParametersDoc(tenantLogicalTime, tenantIntValue, tenantStrValue)});
+
+    ASSERT_PARAMETER_STATE(boost::none, multiIntValue, multiStrValue, multiLogicalTime);
+    ASSERT_PARAMETER_STATE(kTenantId, tenantIntValue, tenantStrValue, tenantLogicalTime);
 }
 
 TEST_F(ClusterServerParameterOpObserverTest, OnUpdateRecord) {
@@ -262,72 +316,110 @@ TEST_F(ClusterServerParameterOpObserverTest, OnUpdateRecord) {
     doUpdate(NamespaceString::kClusterParametersNamespace,
              makeClusterParametersDoc(singleLogicalTime, singleIntValue, singleStrValue));
 
-    ClusterServerParameterTest cspTest = sp->getValue(boost::none);
-    ASSERT_EQ(sp->getClusterParameterTime(boost::none), singleLogicalTime);
-    ASSERT_EQ(cspTest.getIntValue(), singleIntValue);
-    ASSERT_EQ(cspTest.getStrValue(), singleStrValue);
+    ASSERT_PARAMETER_STATE(boost::none, singleIntValue, singleStrValue, singleLogicalTime);
+    ASSERT_PARAMETER_STATE(kTenantId, kInitialTenantIntValue, kInitialTenantStrValue);
 
     // Plausible doc in wrong namespace.
-    assertIgnoredOtherNamespaces([this](const auto& nss) {
-        doUpdate(nss, makeClusterParametersDoc(LogicalTime(), 123, "ignored"));
-    });
+    assertIgnoredOtherNamespaces(
+        [this](const auto& nss) {
+            doUpdate(nss, makeClusterParametersDoc(LogicalTime(), 123, "ignored"));
+        },
+        boost::none);
 
     // Non cluster parameter doc.
-    assertIgnoredAlways([this](const auto& nss) {
-        doUpdate(nss, BSON(ClusterServerParameter::k_idFieldName << "ignored"));
-    });
+    assertIgnoredAlways(
+        [this](const auto& nss) {
+            doUpdate(nss, BSON(ClusterServerParameter::k_idFieldName << "ignored"));
+        },
+        boost::none);
+
+    // Update on separate tenant.
+    const auto tenantLogicalTime = singleLogicalTime.addTicks(1);
+    const auto tenantIntValue = singleIntValue + 1;
+    const auto tenantStrValue = "OnInsertRecord.tenant";
+
+    doUpdate(NamespaceString::makeClusterParametersNSS(kTenantId),
+             makeClusterParametersDoc(tenantLogicalTime, tenantIntValue, tenantStrValue));
+
+    ASSERT_PARAMETER_STATE(boost::none, singleIntValue, singleStrValue, singleLogicalTime);
+    ASSERT_PARAMETER_STATE(kTenantId, tenantIntValue, tenantStrValue, tenantLogicalTime);
 }
 
 TEST_F(ClusterServerParameterOpObserverTest, onDeleteRecord) {
     auto* sp = ServerParameterSet::getClusterParameterSet()->get<ClusterTestParameter>(kCSPTest);
     ASSERT(sp != nullptr);
 
-    const auto initialDoc = initializeState();
+    const auto [initialDocB, initialDocT] = initializeState();
+    // Structured bindings cannot be captured, move to variable.
+    const auto initialDoc = std::move(initialDocB);
 
     // Ignore deletes in other namespaces, whether with or without deleted doc.
-    assertIgnoredOtherNamespaces(
-        [this, initialDoc](const auto& nss) { doDelete(nss, initialDoc); });
+    assertIgnoredOtherNamespaces([this, initialDoc](const auto& nss) { doDelete(nss, initialDoc); },
+                                 boost::none);
 
     // Ignore deletes where the _id does not correspond to a known cluster server parameter.
-    assertIgnoredAlways([this](const auto& nss) {
-        doDelete(nss, BSON(ClusterServerParameter::k_idFieldName << "ignored"));
-    });
+    assertIgnoredAlways(
+        [this](const auto& nss) {
+            doDelete(nss, BSON(ClusterServerParameter::k_idFieldName << "ignored"));
+        },
+        boost::none);
 
     // Reset configuration to defaults when we claim to have deleted the doc.
     doDelete(NamespaceString::kClusterParametersNamespace, initialDoc);
-    ClusterServerParameterTest cspTest = sp->getValue(boost::none);
-    ASSERT_EQ(cspTest.getIntValue(), kDefaultIntValue);
-    ASSERT_EQ(cspTest.getStrValue(), kDefaultStrValue);
 
+    ASSERT_PARAMETER_STATE(boost::none, kDefaultIntValue, kDefaultStrValue);
+    ASSERT_PARAMETER_STATE(kTenantId, kInitialTenantIntValue, kInitialTenantStrValue);
 
     // Restore configured state, and delete without including deleteDoc reference.
     initializeState();
     doDelete(NamespaceString::kClusterParametersNamespace, initialDoc, false);
-    cspTest = sp->getValue(boost::none);
-    ASSERT_EQ(cspTest.getIntValue(), kDefaultIntValue);
-    ASSERT_EQ(cspTest.getStrValue(), kDefaultStrValue);
+
+    ASSERT_PARAMETER_STATE(boost::none, kDefaultIntValue, kDefaultStrValue);
+    ASSERT_PARAMETER_STATE(kTenantId, kInitialTenantIntValue, kInitialTenantStrValue);
+
+    // Restore configured state, and delete other tenant with reference.
+    initializeState();
+    doDelete(NamespaceString::makeClusterParametersNSS(kTenantId), initialDocT);
+
+    ASSERT_PARAMETER_STATE(boost::none, kInitialIntValue, kInitialStrValue);
+    ASSERT_PARAMETER_STATE(kTenantId, kDefaultIntValue, kDefaultStrValue);
+
+    // Restore and delete without reference.
+    initializeState();
+    doDelete(NamespaceString::makeClusterParametersNSS(kTenantId), initialDocT, false);
+
+    ASSERT_PARAMETER_STATE(boost::none, kInitialIntValue, kInitialStrValue);
+    ASSERT_PARAMETER_STATE(kTenantId, kDefaultIntValue, kDefaultStrValue);
 }
 
 TEST_F(ClusterServerParameterOpObserverTest, onDropDatabase) {
     initializeState();
 
     // Drop ignorable databases.
-    assertIgnoredOtherNamespaces([this](const auto& nss) {
-        const auto dbname = nss.db();
-        if (dbname != kConfigDB) {
-            doDropDatabase(dbname);
-        }
-    });
+    assertIgnoredOtherNamespaces(
+        [this](const auto& nss) {
+            const auto dbname = nss.dbName();
+            if (dbname.db() != kConfigDB) {
+                doDropDatabase(dbname);
+            }
+        },
+        boost::none);
 
     // Actually drop the config DB.
-    doDropDatabase(kConfigDB);
+    doDropDatabase(DatabaseName(boost::none, kConfigDB));
 
     auto* sp = ServerParameterSet::getClusterParameterSet()->get<ClusterTestParameter>(kCSPTest);
     ASSERT(sp != nullptr);
 
-    ClusterServerParameterTest cspTest = sp->getValue(boost::none);
-    ASSERT_EQ(cspTest.getIntValue(), kDefaultIntValue);
-    ASSERT_EQ(cspTest.getStrValue(), kDefaultStrValue);
+    ASSERT_PARAMETER_STATE(boost::none, kDefaultIntValue, kDefaultStrValue);
+    ASSERT_PARAMETER_STATE(kTenantId, kInitialTenantIntValue, kInitialTenantStrValue);
+
+    // Reinitialize and drop the other tenant's config DB.
+    initializeState();
+    doDropDatabase(DatabaseName(kTenantId, kConfigDB));
+
+    ASSERT_PARAMETER_STATE(boost::none, kInitialIntValue, kInitialStrValue);
+    ASSERT_PARAMETER_STATE(kTenantId, kDefaultIntValue, kDefaultStrValue);
 }
 
 TEST_F(ClusterServerParameterOpObserverTest, onRenameCollection) {
@@ -335,8 +427,10 @@ TEST_F(ClusterServerParameterOpObserverTest, onRenameCollection) {
 
     const NamespaceString kTestFoo("test", "foo");
     // Rename ignorable collections.
-    assertIgnoredOtherNamespaces([&](const auto& nss) { doRenameCollection(nss, kTestFoo); });
-    assertIgnoredOtherNamespaces([&](const auto& nss) { doRenameCollection(kTestFoo, nss); });
+    assertIgnoredOtherNamespaces([&](const auto& nss) { doRenameCollection(nss, kTestFoo); },
+                                 boost::none);
+    assertIgnoredOtherNamespaces([&](const auto& nss) { doRenameCollection(kTestFoo, nss); },
+                                 boost::none);
 
     auto* sp = ServerParameterSet::getClusterParameterSet()->get<ClusterTestParameter>(kCSPTest);
     ASSERT(sp != nullptr);
@@ -344,17 +438,41 @@ TEST_F(ClusterServerParameterOpObserverTest, onRenameCollection) {
     // These renames "work" despite not mutating durable state
     // since the rename away doesn't require a rescan.
 
-    // Rename away (and reset to default)
+    // Upsert a doc for other tenant to make sure it doesn't get rescanned.
+    auto doc = makeClusterParametersDoc(LogicalTime(Timestamp(time(nullptr))), 111, "rename");
+    upsert(doc, kTenantId);
+    // Rename away (and reset that tenant to default)
     doRenameCollection(NamespaceString::kClusterParametersNamespace, kTestFoo);
-    ClusterServerParameterTest cspTest = sp->getValue(boost::none);
-    ASSERT_EQ(cspTest.getIntValue(), kDefaultIntValue);
-    ASSERT_EQ(cspTest.getStrValue(), kDefaultStrValue);
 
-    // Rename in (and restore to initialized state)
+    ASSERT_PARAMETER_STATE(boost::none, kDefaultIntValue, kDefaultStrValue);
+    ASSERT_PARAMETER_STATE(kTenantId, kInitialTenantIntValue, kInitialTenantStrValue);
+
+    // Rename in (and restore that tenant to initialized state)
     doRenameCollection(kTestFoo, NamespaceString::kClusterParametersNamespace);
-    cspTest = sp->getValue(boost::none);
-    ASSERT_EQ(cspTest.getIntValue(), kInitialIntValue);
-    ASSERT_EQ(cspTest.getStrValue(), kInitialStrValue);
+
+    ASSERT_PARAMETER_STATE(boost::none, kInitialIntValue, kInitialStrValue);
+    ASSERT_PARAMETER_STATE(kTenantId, kInitialTenantIntValue, kInitialTenantStrValue);
+
+    // Repeat with other tenant
+    doRenameCollection(NamespaceString::makeClusterParametersNSS(kTenantId), kTestFoo);
+
+    ASSERT_PARAMETER_STATE(boost::none, kInitialIntValue, kInitialStrValue);
+    ASSERT_PARAMETER_STATE(kTenantId, kDefaultIntValue, kDefaultStrValue);
+
+    // Rename in (and restore to initialized state)
+    doRenameCollection(kTestFoo, NamespaceString::makeClusterParametersNSS(kTenantId));
+
+    ASSERT_PARAMETER_STATE(boost::none, kInitialIntValue, kInitialStrValue);
+    ASSERT_PARAMETER_STATE(kTenantId, 111, "rename");
+
+    doc = makeClusterParametersDoc(LogicalTime(Timestamp(time(nullptr))), 222, "rename2");
+    upsert(doc, kTenantId);
+    // Rename from one tenant to another -- should clear from-collection and reload to-collection
+    doRenameCollection(NamespaceString::makeClusterParametersNSS(boost::none),
+                       NamespaceString::makeClusterParametersNSS(kTenantId));
+
+    ASSERT_PARAMETER_STATE(boost::none, kDefaultIntValue, kDefaultStrValue);
+    ASSERT_PARAMETER_STATE(kTenantId, 222, "rename2");
 }
 
 TEST_F(ClusterServerParameterOpObserverTest, onImportCollection) {
@@ -362,19 +480,29 @@ TEST_F(ClusterServerParameterOpObserverTest, onImportCollection) {
 
     const NamespaceString kTestFoo("test", "foo");
     // Import ignorable collections.
-    assertIgnoredOtherNamespaces([&](const auto& nss) { doImportCollection(nss); });
+    assertIgnoredOtherNamespaces([&](const auto& nss) { doImportCollection(nss); }, boost::none);
 
     auto* sp = ServerParameterSet::getClusterParameterSet()->get<ClusterTestParameter>(kCSPTest);
     ASSERT(sp != nullptr);
 
-    // Import the collection (rescan).
     auto doc =
         makeClusterParametersDoc(LogicalTime(Timestamp(time(nullptr))), 333, "onImportCollection");
-    upsert(doc);
+    upsert(doc, boost::none);
+    doc = makeClusterParametersDoc(
+        LogicalTime(Timestamp(time(nullptr))), 444, "onImportCollection.tenant");
+    upsert(doc, kTenantId);
+    // Import the collection -- should rescan only that tenant's cluster parameters.
     doImportCollection(NamespaceString::kClusterParametersNamespace);
-    ClusterServerParameterTest cspTest = sp->getValue(boost::none);
-    ASSERT_EQ(cspTest.getIntValue(), 333);
-    ASSERT_EQ(cspTest.getStrValue(), "onImportCollection");
+
+    ASSERT_PARAMETER_STATE(boost::none, 333, "onImportCollection");
+    ASSERT_PARAMETER_STATE(kTenantId, kInitialTenantIntValue, kInitialTenantStrValue);
+
+    remove(boost::none);
+    // Import the other tenant collection -- should rescan only other tenant's parameters.
+    doImportCollection(NamespaceString::makeClusterParametersNSS(kTenantId));
+
+    ASSERT_PARAMETER_STATE(boost::none, 333, "onImportCollection");
+    ASSERT_PARAMETER_STATE(kTenantId, 444, "onImportCollection.tenant");
 }
 
 TEST_F(ClusterServerParameterOpObserverTest, onReplicationRollback) {
@@ -382,34 +510,36 @@ TEST_F(ClusterServerParameterOpObserverTest, onReplicationRollback) {
 
     const NamespaceString kTestFoo("test", "foo");
     // Import ignorable collections.
-    assertIgnoredOtherNamespaces([&](const auto& nss) { doImportCollection(nss); });
+    assertIgnoredOtherNamespaces([&](const auto& nss) { doReplicationRollback({nss}); },
+                                 boost::none);
 
     auto* sp = ServerParameterSet::getClusterParameterSet()->get<ClusterTestParameter>(kCSPTest);
     ASSERT(sp != nullptr);
 
-    // Trigger rollback of ignorable namespaces.
+    remove(boost::none);
+    remove(kTenantId);
+    // Trigger rollback of ignorable namespaces and ensure no disk reload occurs.
     doReplicationRollback(kIgnoredNamespaces);
+    ASSERT_PARAMETER_STATE(boost::none, kInitialIntValue, kInitialStrValue);
+    ASSERT_PARAMETER_STATE(kTenantId, kInitialTenantIntValue, kInitialTenantStrValue);
 
-    ClusterServerParameterTest cspTest = sp->getValue(boost::none);
-    ASSERT_EQ(cspTest.getIntValue(), kInitialIntValue);
-    ASSERT_EQ(cspTest.getStrValue(), kInitialStrValue);
-
-    // Trigger rollback of relevant namespace.
-    remove();
+    // Trigger rollback of relevant namespace and ensure disk reload occurs only for that tenant.
     doReplicationRollback({NamespaceString::kClusterParametersNamespace});
-    cspTest = sp->getValue(boost::none);
-    ASSERT_EQ(cspTest.getIntValue(), kDefaultIntValue);
-    ASSERT_EQ(cspTest.getStrValue(), kDefaultStrValue);
 
-    // Rollback the rollback.
+    ASSERT_PARAMETER_STATE(boost::none, kDefaultIntValue, kDefaultStrValue);
+    ASSERT_PARAMETER_STATE(kTenantId, kInitialTenantIntValue, kInitialTenantStrValue);
+    // Trigger rollback of other tenant's namespace and ensure disk reload occurs only for that
+    // tenant.
     auto doc = makeClusterParametersDoc(
         LogicalTime(Timestamp(time(nullptr))), 444, "onReplicationRollback");
-    upsert(doc);
-    cspTest = sp->getValue(boost::none);
-    doReplicationRollback({NamespaceString::kClusterParametersNamespace});
-    ASSERT_EQ(cspTest.getIntValue(), kDefaultIntValue);
-    ASSERT_EQ(cspTest.getStrValue(), kDefaultStrValue);
+    upsert(doc, boost::none);
+    doReplicationRollback({NamespaceString::makeClusterParametersNSS(kTenantId)});
+
+    ASSERT_PARAMETER_STATE(boost::none, kDefaultIntValue, kDefaultStrValue);
+    ASSERT_PARAMETER_STATE(kTenantId, kDefaultIntValue, kDefaultStrValue);
 }
 
+#undef ASSERT_PARAMETER_STATE
+
 }  // namespace
 }  // namespace mongo