SERVER-47803 Move the database level profile setting from Database into CollectionCatalog

26 files changed, 227 insertions, 187 deletions

diff --git a/src/mongo/db/catalog/SConscript b/src/mongo/db/catalog/SConscript
index aedda2ed6af..d0e60f796af 100644
--- a/src/mongo/db/catalog/SConscript
+++ b/src/mongo/db/catalog/SConscript
@@ -277,6 +277,7 @@ env.Library(
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/concurrency/write_conflict_exception',
+        '$BUILD_DIR/mongo/db/server_options_core',
         '$BUILD_DIR/mongo/idl/server_parameter',
     ]
 )
diff --git a/src/mongo/db/catalog/collection_catalog.cpp b/src/mongo/db/catalog/collection_catalog.cpp
index c6e4973f259..eed2b4f28ee 100644
--- a/src/mongo/db/catalog/collection_catalog.cpp
+++ b/src/mongo/db/catalog/collection_catalog.cpp
@@ -36,6 +36,7 @@
 #include "mongo/db/catalog/uncommitted_collections.h"
 #include "mongo/db/concurrency/lock_manager_defs.h"
 #include "mongo/db/concurrency/write_conflict_exception.h"
+#include "mongo/db/server_options.h"
 #include "mongo/db/storage/recovery_unit.h"
 #include "mongo/logv2/log.h"
 #include "mongo/util/assert_util.h"
@@ -428,6 +429,26 @@ std::vector<std::string> CollectionCatalog::getAllDbNames() const {
     return ret;
 }
 
+void CollectionCatalog::setDatabaseProfileLevel(StringData dbName, int newProfileLevel) {
+    stdx::lock_guard<Latch> lock(_profileLevelsLock);
+    _databaseProfileLevels[dbName] = newProfileLevel;
+}
+
+int CollectionCatalog::getDatabaseProfileLevel(StringData dbName) const {
+    stdx::lock_guard<Latch> lock(_profileLevelsLock);
+    auto it = _databaseProfileLevels.find(dbName);
+    if (it != _databaseProfileLevels.end()) {
+        return it->second;
+    }
+
+    return serverGlobalParams.defaultProfile;
+}
+
+void CollectionCatalog::clearDatabaseProfileLevel(StringData dbName) {
+    stdx::lock_guard<Latch> lock(_profileLevelsLock);
+    _databaseProfileLevels.erase(dbName);
+}
+
 void CollectionCatalog::registerCollection(CollectionUUID uuid, std::unique_ptr<Collection>* coll) {
     auto ns = (*coll)->ns();
     stdx::lock_guard<Latch> lock(_catalogLock);
diff --git a/src/mongo/db/catalog/collection_catalog.h b/src/mongo/db/catalog/collection_catalog.h
index 1a57e6f55f5..2fd90339a29 100644
--- a/src/mongo/db/catalog/collection_catalog.h
+++ b/src/mongo/db/catalog/collection_catalog.h
@@ -219,6 +219,23 @@ public:
     std::vector<std::string> getAllDbNames() const;
 
     /**
+     * Sets 'newProfileLevel' as the profiling level for the database 'dbName'.
+     */
+    void setDatabaseProfileLevel(StringData dbName, int newProfileLevel);
+
+    /**
+     * Fetches the profiling level for database 'dbName'.
+     *
+     * Returns the server's default database profile level if the database does not exist.
+     */
+    int getDatabaseProfileLevel(StringData dbName) const;
+
+    /**
+     * Clears the database profile level entry for 'dbName'.
+     */
+    void clearDatabaseProfileLevel(StringData dbName);
+
+    /**
      * Puts the catalog in closed state. In this state, the lookupNSSByUUID method will fall back
      * to the pre-close state to resolve queries for currently unknown UUIDs. This allows processes,
      * like authorization and replication, which need to do lookups outside of database locks, to
@@ -283,10 +300,12 @@ private:
         mongo::stdx::unordered_map<CollectionUUID, NamespaceString, CollectionUUID::Hash>>
         _shadowCatalog;
 
-    using CollectionCatalogMap = mongo::stdx::
-        unordered_map<CollectionUUID, std::unique_ptr<Collection>, CollectionUUID::Hash>;
+    using CollectionCatalogMap =
+        stdx::unordered_map<CollectionUUID, std::unique_ptr<Collection>, CollectionUUID::Hash>;
     using OrderedCollectionMap = std::map<std::pair<std::string, CollectionUUID>, Collection*>;
-    using NamespaceCollectionMap = mongo::stdx::unordered_map<NamespaceString, Collection*>;
+    using NamespaceCollectionMap = stdx::unordered_map<NamespaceString, Collection*>;
+    using DatabaseProfileLevelMap = StringMap<int>;
+
     CollectionCatalogMap _catalog;
     OrderedCollectionMap _orderedCollections;  // Ordered by <dbName, collUUID> pair
     NamespaceCollectionMap _collections;
@@ -313,5 +332,15 @@ private:
 
     // Mapping from ResourceId to a set of strings that contains collection and database namespaces.
     std::map<ResourceId, std::set<std::string>> _resourceInformation;
+
+    // Protects _databaseProfileLevels.
+    mutable Mutex _profileLevelsLock = MONGO_MAKE_LATCH("CollectionCatalog::_profileLevelsLock");
+
+    /**
+     * Contains non-default database profile levels. New collections, current collections and views
+     * must all be able to access the correct profile level for the database in which they reside.
+     * Simple database name to integer profile level map. Access protected by the _catalogLock.
+     */
+    DatabaseProfileLevelMap _databaseProfileLevels;
 };
 }  // namespace mongo
diff --git a/src/mongo/db/catalog/collection_catalog_test.cpp b/src/mongo/db/catalog/collection_catalog_test.cpp
index 4cf75ec7962..4a8e5785d4a 100644
--- a/src/mongo/db/catalog/collection_catalog_test.cpp
+++ b/src/mongo/db/catalog/collection_catalog_test.cpp
@@ -41,7 +41,8 @@
 #include "mongo/unittest/death_test.h"
 #include "mongo/unittest/unittest.h"
 
-using namespace mongo;
+namespace mongo {
+namespace {
 
 /**
  * A test fixture that creates a CollectionCatalog and Collection* pointer to store in it.
@@ -314,8 +315,6 @@ protected:
     CollectionCatalog catalog;
 };
 
-namespace {
-
 TEST_F(CollectionCatalogResourceTest, RemoveAllResources) {
     catalog.deregisterAllCollections();
 
@@ -617,6 +616,25 @@ TEST_F(CollectionCatalogTest, GetAllCollectionNamesAndGetAllDbNames) {
     catalog.deregisterAllCollections();
 }
 
+// Test setting and fetching the profile level for a database.
+TEST_F(CollectionCatalogTest, DatabaseProfileLevel) {
+    std::string testDBNameFirst = "testdbfirst";
+    std::string testDBNameSecond = "testdbsecond";
+
+    // Requesting a profile level that is not in the _databaseProfileLevel map should return the
+    // default server-wide setting
+    ASSERT_EQ(catalog.getDatabaseProfileLevel(testDBNameFirst), serverGlobalParams.defaultProfile);
+
+    // Setting the default profile level should have not change the result.
+    catalog.setDatabaseProfileLevel(testDBNameFirst, serverGlobalParams.defaultProfile);
+    ASSERT_EQ(catalog.getDatabaseProfileLevel(testDBNameFirst), serverGlobalParams.defaultProfile);
+
+    // Changing the profile level should make fetching it different.
+    catalog.setDatabaseProfileLevel(testDBNameSecond, serverGlobalParams.defaultProfile + 1);
+    ASSERT_EQ(catalog.getDatabaseProfileLevel(testDBNameSecond),
+              serverGlobalParams.defaultProfile + 1);
+}
+
 TEST_F(CollectionCatalogTest, GetAllCollectionNamesAndGetAllDbNamesWithUncommittedCollections) {
     NamespaceString aColl("dbA", "collA");
     NamespaceString b1Coll("dbB", "collB1");
@@ -789,3 +807,4 @@ TEST_F(ForEachCollectionFromDbTest, ForEachCollectionFromDbWithPredicate) {
 }
 
 }  // namespace
+}  // namespace mongo
diff --git a/src/mongo/db/catalog/create_collection.cpp b/src/mongo/db/catalog/create_collection.cpp
index f19d74c40ed..0b6f1568948 100644
--- a/src/mongo/db/catalog/create_collection.cpp
+++ b/src/mongo/db/catalog/create_collection.cpp
@@ -86,11 +86,12 @@ Status _createView(OperationContext* opCtx,
 
         WriteUnitOfWork wunit(opCtx);
 
-        AutoStatsTracker statsTracker(opCtx,
-                                      nss,
-                                      Top::LockType::NotLocked,
-                                      AutoStatsTracker::LogMode::kUpdateTopAndCurOp,
-                                      db->getProfilingLevel());
+        AutoStatsTracker statsTracker(
+            opCtx,
+            nss,
+            Top::LockType::NotLocked,
+            AutoStatsTracker::LogMode::kUpdateTopAndCurOp,
+            CollectionCatalog::get(opCtx).getDatabaseProfileLevel(nss.db()));
 
         // If the view creation rolls back, ensure that the Top entry created for the view is
         // deleted.
@@ -135,11 +136,12 @@ Status _createCollection(OperationContext* opCtx,
 
         WriteUnitOfWork wunit(opCtx);
 
-        AutoStatsTracker statsTracker(opCtx,
-                                      nss,
-                                      Top::LockType::NotLocked,
-                                      AutoStatsTracker::LogMode::kUpdateTopAndCurOp,
-                                      autoDb.getDb()->getProfilingLevel());
+        AutoStatsTracker statsTracker(
+            opCtx,
+            nss,
+            Top::LockType::NotLocked,
+            AutoStatsTracker::LogMode::kUpdateTopAndCurOp,
+            CollectionCatalog::get(opCtx).getDatabaseProfileLevel(nss.db()));
 
         // If the collection creation rolls back, ensure that the Top entry created for the
         // collection is deleted.
diff --git a/src/mongo/db/catalog/database.h b/src/mongo/db/catalog/database.h
index 1de4fa3e09e..27982bbf772 100644
--- a/src/mongo/db/catalog/database.h
+++ b/src/mongo/db/catalog/database.h
@@ -87,16 +87,6 @@ public:
     virtual void clearTmpCollections(OperationContext* const opCtx) const = 0;
 
     /**
-     * Sets a new profiling level for the database and returns the outcome.
-     *
-     * @param opCtx Operation context which to use for creating the profiling collection.
-     * @param newLevel New profiling level to use.
-     */
-    virtual Status setProfilingLevel(OperationContext* const opCtx, const int newLevel) = 0;
-
-    virtual int getProfilingLevel() const = 0;
-
-    /**
      * Sets the 'drop-pending' state of this Database.
      * This is done at the beginning of a dropDatabase operation and is used to reject subsequent
      * collection creation requests on this database.
diff --git a/src/mongo/db/catalog/database_holder_impl.cpp b/src/mongo/db/catalog/database_holder_impl.cpp
index 7d72ed98cef..ccd202e95ba 100644
--- a/src/mongo/db/catalog/database_holder_impl.cpp
+++ b/src/mongo/db/catalog/database_holder_impl.cpp
@@ -204,6 +204,8 @@ void DatabaseHolderImpl::dropDb(OperationContext* opCtx, Database* db) {
         Top::get(serviceContext).collectionDropped(coll->ns());
     }
 
+    // Clean up the in-memory database state.
+    CollectionCatalog::get(opCtx).clearDatabaseProfileLevel(name);
     close(opCtx, name);
 
     auto const storageEngine = serviceContext->getStorageEngine();
diff --git a/src/mongo/db/catalog/database_impl.cpp b/src/mongo/db/catalog/database_impl.cpp
index bfc4ff75ca0..fb277d08ecc 100644
--- a/src/mongo/db/catalog/database_impl.cpp
+++ b/src/mongo/db/catalog/database_impl.cpp
@@ -136,7 +136,6 @@ DatabaseImpl::DatabaseImpl(const StringData name)
     auto viewCatalog = std::make_unique<ViewCatalog>(std::move(durableViewCatalog));
 
     ViewCatalog::set(this, std::move(viewCatalog));
-    _profile.store(serverGlobalParams.defaultProfile);
 }
 
 void DatabaseImpl::init(OperationContext* const opCtx) const {
@@ -211,39 +210,6 @@ void DatabaseImpl::clearTmpCollections(OperationContext* opCtx) const {
     catalog::forEachCollectionFromDb(opCtx, name(), MODE_X, callback, predicate);
 }
 
-Status DatabaseImpl::setProfilingLevel(OperationContext* opCtx, int newLevel) {
-    auto currLevel = _profile.load();
-
-    if (currLevel == newLevel) {
-        return Status::OK();
-    }
-
-    if (newLevel == 0) {
-        _profile.store(0);
-        return Status::OK();
-    }
-
-    if (newLevel < 0 || newLevel > 2) {
-        return Status(ErrorCodes::BadValue, "profiling level has to be >=0 and <= 2");
-    }
-
-    // Can't support profiling without supporting capped collections.
-    if (!opCtx->getServiceContext()->getStorageEngine()->supportsCappedCollections()) {
-        return Status(ErrorCodes::CommandNotSupported,
-                      "the storage engine doesn't support profiling.");
-    }
-
-    Status status = createProfileCollection(opCtx, this);
-
-    if (!status.isOK()) {
-        return status;
-    }
-
-    _profile.store(newLevel);
-
-    return Status::OK();
-}
-
 void DatabaseImpl::setDropPending(OperationContext* opCtx, bool dropPending) {
     auto mode = dropPending ? MODE_X : MODE_IX;
     invariant(opCtx->lockState()->isDbLockedForMode(name(), mode));
@@ -341,7 +307,7 @@ Status DatabaseImpl::dropCollection(OperationContext* opCtx,
 
     if (nss.isSystem()) {
         if (nss.isSystemDotProfile()) {
-            if (_profile.load() != 0)
+            if (CollectionCatalog::get(opCtx).getDatabaseProfileLevel(_name) != 0)
                 return Status(ErrorCodes::IllegalOperation,
                               "turn off profiling before dropping system.profile collection");
         } else if (!(nss.isSystemDotViews() || nss.isHealthlog() ||
diff --git a/src/mongo/db/catalog/database_impl.h b/src/mongo/db/catalog/database_impl.h
index 3992a645e04..2cb451a2778 100644
--- a/src/mongo/db/catalog/database_impl.h
+++ b/src/mongo/db/catalog/database_impl.h
@@ -45,18 +45,6 @@ public:
 
     void clearTmpCollections(OperationContext* opCtx) const final;
 
-    /**
-     * Sets a new profiling level for the database and returns the outcome.
-     *
-     * @param opCtx Operation context which to use for creating the profiling collection.
-     * @param newLevel New profiling level to use.
-     */
-    Status setProfilingLevel(OperationContext* opCtx, int newLevel) final;
-
-    int getProfilingLevel() const final {
-        return _profile.load();
-    }
-
     void setDropPending(OperationContext* opCtx, bool dropPending) final;
 
     bool isDropPending(OperationContext* opCtx) const final;
@@ -161,8 +149,6 @@ private:
 
     const NamespaceString _viewsName;  // "dbname.system.views"
 
-    AtomicWord<int> _profile{0};  // 0=off
-
     // If '_dropPending' is true, this Database is in the midst of a two-phase drop. No new
     // collections may be created in this Database.
     // This variable may only be read/written while the database is locked in MODE_X.
diff --git a/src/mongo/db/catalog/drop_collection.cpp b/src/mongo/db/catalog/drop_collection.cpp
index 47c71c3feac..2f96e840b0b 100644
--- a/src/mongo/db/catalog/drop_collection.cpp
+++ b/src/mongo/db/catalog/drop_collection.cpp
@@ -94,11 +94,12 @@ Status _dropView(OperationContext* opCtx,
         hangDuringDropCollection.pauseWhileSet();
     }
 
-    AutoStatsTracker statsTracker(opCtx,
-                                  collectionName,
-                                  Top::LockType::NotLocked,
-                                  AutoStatsTracker::LogMode::kUpdateCurOp,
-                                  db->getProfilingLevel());
+    AutoStatsTracker statsTracker(
+        opCtx,
+        collectionName,
+        Top::LockType::NotLocked,
+        AutoStatsTracker::LogMode::kUpdateCurOp,
+        CollectionCatalog::get(opCtx).getDatabaseProfileLevel(collectionName.db()));
 
     if (opCtx->writesAreReplicated() &&
         !repl::ReplicationCoordinator::get(opCtx)->canAcceptWritesFor(opCtx, collectionName)) {
@@ -147,11 +148,12 @@ Status _abortIndexBuildsAndDropCollection(OperationContext* opCtx,
         hangDuringDropCollection.pauseWhileSet();
     }
 
-    AutoStatsTracker statsTracker(opCtx,
-                                  startingNss,
-                                  Top::LockType::NotLocked,
-                                  AutoStatsTracker::LogMode::kUpdateCurOp,
-                                  autoDb->getDb()->getProfilingLevel());
+    AutoStatsTracker statsTracker(
+        opCtx,
+        startingNss,
+        Top::LockType::NotLocked,
+        AutoStatsTracker::LogMode::kUpdateCurOp,
+        CollectionCatalog::get(opCtx).getDatabaseProfileLevel(startingNss.db()));
 
     IndexBuildsCoordinator* indexBuildsCoord = IndexBuildsCoordinator::get(opCtx);
     const UUID collectionUUID = coll->uuid();
@@ -244,11 +246,12 @@ Status _dropCollection(OperationContext* opCtx,
         hangDuringDropCollection.pauseWhileSet();
     }
 
-    AutoStatsTracker statsTracker(opCtx,
-                                  collectionName,
-                                  Top::LockType::NotLocked,
-                                  AutoStatsTracker::LogMode::kUpdateCurOp,
-                                  db->getProfilingLevel());
+    AutoStatsTracker statsTracker(
+        opCtx,
+        collectionName,
+        Top::LockType::NotLocked,
+        AutoStatsTracker::LogMode::kUpdateCurOp,
+        CollectionCatalog::get(opCtx).getDatabaseProfileLevel(collectionName.db()));
 
     WriteUnitOfWork wunit(opCtx);
 
diff --git a/src/mongo/db/catalog/rename_collection.cpp b/src/mongo/db/catalog/rename_collection.cpp
index 24b9f9cbb6b..e87efd91d74 100644
--- a/src/mongo/db/catalog/rename_collection.cpp
+++ b/src/mongo/db/catalog/rename_collection.cpp
@@ -325,11 +325,12 @@ Status renameCollectionWithinDB(OperationContext* opCtx,
     Collection* const targetColl =
         CollectionCatalog::get(opCtx).lookupCollectionByNamespace(opCtx, target);
 
-    AutoStatsTracker statsTracker(opCtx,
-                                  source,
-                                  Top::LockType::NotLocked,
-                                  AutoStatsTracker::LogMode::kUpdateCurOp,
-                                  db->getProfilingLevel());
+    AutoStatsTracker statsTracker(
+        opCtx,
+        source,
+        Top::LockType::NotLocked,
+        AutoStatsTracker::LogMode::kUpdateCurOp,
+        CollectionCatalog::get(opCtx).getDatabaseProfileLevel(source.db()));
 
     if (!targetColl) {
         return renameCollectionDirectly(opCtx, db, sourceColl->uuid(), source, target, options);
@@ -365,11 +366,12 @@ Status renameCollectionWithinDBForApplyOps(OperationContext* opCtx,
     Collection* const sourceColl =
         CollectionCatalog::get(opCtx).lookupCollectionByNamespace(opCtx, source);
 
-    AutoStatsTracker statsTracker(opCtx,
-                                  source,
-                                  Top::LockType::NotLocked,
-                                  AutoStatsTracker::LogMode::kUpdateCurOp,
-                                  db->getProfilingLevel());
+    AutoStatsTracker statsTracker(
+        opCtx,
+        source,
+        Top::LockType::NotLocked,
+        AutoStatsTracker::LogMode::kUpdateCurOp,
+        CollectionCatalog::get(opCtx).getDatabaseProfileLevel(source.db()));
 
     return writeConflictRetry(opCtx, "renameCollection", target.ns(), [&] {
         Collection* targetColl =
@@ -482,12 +484,13 @@ Status renameBetweenDBs(OperationContext* opCtx,
     if (!sourceDB)
         return Status(ErrorCodes::NamespaceNotFound, "source namespace does not exist");
 
-    boost::optional<AutoStatsTracker> statsTracker(boost::in_place_init,
-                                                   opCtx,
-                                                   source,
-                                                   Top::LockType::NotLocked,
-                                                   AutoStatsTracker::LogMode::kUpdateCurOp,
-                                                   sourceDB->getProfilingLevel());
+    boost::optional<AutoStatsTracker> statsTracker(
+        boost::in_place_init,
+        opCtx,
+        source,
+        Top::LockType::NotLocked,
+        AutoStatsTracker::LogMode::kUpdateCurOp,
+        CollectionCatalog::get(opCtx).getDatabaseProfileLevel(source.db()));
 
     Collection* const sourceColl =
         CollectionCatalog::get(opCtx).lookupCollectionByNamespace(opCtx, source);
diff --git a/src/mongo/db/commands/create_indexes.cpp b/src/mongo/db/commands/create_indexes.cpp
index 2f48fbac2ee..66a2121380a 100644
--- a/src/mongo/db/commands/create_indexes.cpp
+++ b/src/mongo/db/commands/create_indexes.cpp
@@ -550,12 +550,11 @@ bool runCreateIndexesWithCoordinator(OperationContext* opCtx,
 
     // Use AutoStatsTracker to update Top.
     boost::optional<AutoStatsTracker> statsTracker;
-    const boost::optional<int> dbProfilingLevel = boost::none;
     statsTracker.emplace(opCtx,
                          ns,
                          Top::LockType::WriteLocked,
                          AutoStatsTracker::LogMode::kUpdateTopAndCurOp,
-                         dbProfilingLevel);
+                         CollectionCatalog::get(opCtx).getDatabaseProfileLevel(ns.db()));
 
     auto buildUUID = UUID::gen();
     ReplIndexBuildState::IndexCatalogStats stats;
diff --git a/src/mongo/db/commands/dbcommands.cpp b/src/mongo/db/commands/dbcommands.cpp
index 9310a2654b1..514dc83c026 100644
--- a/src/mongo/db/commands/dbcommands.cpp
+++ b/src/mongo/db/commands/dbcommands.cpp
@@ -753,7 +753,8 @@ public:
             {
                 stdx::lock_guard<Client> lk(*opCtx->getClient());
                 // TODO: OldClientContext legacy, needs to be removed
-                CurOp::get(opCtx)->enter_inlock(dbname.c_str(), db->getProfilingLevel());
+                CurOp::get(opCtx)->enter_inlock(
+                    dbname.c_str(), CollectionCatalog::get(opCtx).getDatabaseProfileLevel(dbname));
             }
 
             db->getStats(opCtx, &result, scale);
diff --git a/src/mongo/db/commands/dbcommands_d.cpp b/src/mongo/db/commands/dbcommands_d.cpp
index c6c1eccb48d..2405659d110 100644
--- a/src/mongo/db/commands/dbcommands_d.cpp
+++ b/src/mongo/db/commands/dbcommands_d.cpp
@@ -47,6 +47,7 @@
 #include "mongo/db/auth/user_name.h"
 #include "mongo/db/background.h"
 #include "mongo/db/catalog/coll_mod.h"
+#include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/catalog/create_collection.h"
 #include "mongo/db/catalog/database_holder.h"
 #include "mongo/db/catalog/drop_collection.h"
@@ -105,6 +106,41 @@ MONGO_FAIL_POINT_DEFINE(waitInFilemd5DuringManualYield);
 
 namespace {
 
+Status _setProfilingLevel(OperationContext* opCtx, Database* db, StringData dbName, int newLevel) {
+    invariant(db);
+
+    auto currLevel = CollectionCatalog::get(opCtx).getDatabaseProfileLevel(dbName);
+
+    if (currLevel == newLevel) {
+        return Status::OK();
+    }
+
+    if (newLevel == 0) {
+        CollectionCatalog::get(opCtx).setDatabaseProfileLevel(dbName, newLevel);
+        return Status::OK();
+    }
+
+    if (newLevel < 0 || newLevel > 2) {
+        return Status(ErrorCodes::BadValue, "profiling level has to be >=0 and <= 2");
+    }
+
+    // Can't support profiling without supporting capped collections.
+    if (!opCtx->getServiceContext()->getStorageEngine()->supportsCappedCollections()) {
+        return Status(ErrorCodes::CommandNotSupported,
+                      "the storage engine doesn't support profiling.");
+    }
+
+    Status status = createProfileCollection(opCtx, db);
+    if (!status.isOK()) {
+        return status;
+    }
+
+    CollectionCatalog::get(opCtx).setDatabaseProfileLevel(dbName, newLevel);
+
+    return Status::OK();
+}
+
+
 /**
  * Sets the profiling level, logging/profiling threshold, and logging/profiling sample rate for the
  * given database.
@@ -131,7 +167,8 @@ protected:
         AutoGetDb ctx(opCtx, dbName, dbMode);
         Database* db = ctx.getDb();
 
-        auto oldLevel = (db ? db->getProfilingLevel() : serverGlobalParams.defaultProfile);
+        // Fetches the database profiling level or the server default if the db does not exist.
+        auto oldLevel = CollectionCatalog::get(opCtx).getDatabaseProfileLevel(dbName);
 
         if (!readOnly) {
             if (!db) {
@@ -140,7 +177,8 @@ protected:
                 auto databaseHolder = DatabaseHolder::get(opCtx);
                 db = databaseHolder->openDb(opCtx, dbName);
             }
-            uassertStatusOK(db->setProfilingLevel(opCtx, profilingLevel));
+
+            uassertStatusOK(_setProfilingLevel(opCtx, db, dbName, profilingLevel));
         }
 
         return oldLevel;
diff --git a/src/mongo/db/commands/find_and_modify.cpp b/src/mongo/db/commands/find_and_modify.cpp
index 4f99b56cd2e..2c86f93243d 100644
--- a/src/mongo/db/commands/find_and_modify.cpp
+++ b/src/mongo/db/commands/find_and_modify.cpp
@@ -454,12 +454,10 @@ public:
         AutoGetCollection autoColl(opCtx, nsString, MODE_IX);
 
         {
-            boost::optional<int> dbProfilingLevel;
-            if (autoColl.getDb())
-                dbProfilingLevel = autoColl.getDb()->getProfilingLevel();
-
             stdx::lock_guard<Client> lk(*opCtx->getClient());
-            CurOp::get(opCtx)->enter_inlock(nsString.ns().c_str(), dbProfilingLevel);
+            CurOp::get(opCtx)->enter_inlock(
+                nsString.ns().c_str(),
+                CollectionCatalog::get(opCtx).getDatabaseProfileLevel(nsString.db()));
         }
 
         assertCanWrite(opCtx, nsString);
@@ -514,11 +512,10 @@ public:
         Database* db = autoColl.ensureDbExists();
 
         {
-            boost::optional<int> dbProfilingLevel;
-            dbProfilingLevel = db->getProfilingLevel();
-
             stdx::lock_guard<Client> lk(*opCtx->getClient());
-            CurOp::get(opCtx)->enter_inlock(nsString.ns().c_str(), dbProfilingLevel);
+            CurOp::get(opCtx)->enter_inlock(
+                nsString.ns().c_str(),
+                CollectionCatalog::get(opCtx).getDatabaseProfileLevel(nsString.db()));
         }
 
         assertCanWrite(opCtx, nsString);
diff --git a/src/mongo/db/commands/getmore_cmd.cpp b/src/mongo/db/commands/getmore_cmd.cpp
index 1e41387d0dd..133ecf102a9 100644
--- a/src/mongo/db/commands/getmore_cmd.cpp
+++ b/src/mongo/db/commands/getmore_cmd.cpp
@@ -414,12 +414,12 @@ public:
             }
             if (cursorPin->lockPolicy() == ClientCursorParams::LockPolicy::kLocksInternally) {
                 if (!_request.nss.isCollectionlessCursorNamespace()) {
-                    const boost::optional<int> dbProfilingLevel = boost::none;
-                    statsTracker.emplace(opCtx,
-                                         _request.nss,
-                                         Top::LockType::NotLocked,
-                                         AutoStatsTracker::LogMode::kUpdateTopAndCurOp,
-                                         dbProfilingLevel);
+                    statsTracker.emplace(
+                        opCtx,
+                        _request.nss,
+                        Top::LockType::NotLocked,
+                        AutoStatsTracker::LogMode::kUpdateTopAndCurOp,
+                        CollectionCatalog::get(opCtx).getDatabaseProfileLevel(_request.nss.db()));
                 }
             } else {
                 invariant(cursorPin->lockPolicy() ==
@@ -444,13 +444,12 @@ public:
                 // Otherwise, these two namespaces will match.
                 readLock.emplace(opCtx, cursorPin->getExecutor()->nss());
 
-                const int doNotChangeProfilingLevel = 0;
-                statsTracker.emplace(opCtx,
-                                     _request.nss,
-                                     Top::LockType::ReadLocked,
-                                     AutoStatsTracker::LogMode::kUpdateTopAndCurOp,
-                                     readLock->getDb() ? readLock->getDb()->getProfilingLevel()
-                                                       : doNotChangeProfilingLevel);
+                statsTracker.emplace(
+                    opCtx,
+                    _request.nss,
+                    Top::LockType::ReadLocked,
+                    AutoStatsTracker::LogMode::kUpdateTopAndCurOp,
+                    CollectionCatalog::get(opCtx).getDatabaseProfileLevel(_request.nss.db()));
 
                 // Check whether we are allowed to read from this node after acquiring our locks.
                 uassertStatusOK(repl::ReplicationCoordinator::get(opCtx)->checkCanServeReadsFor(
diff --git a/src/mongo/db/commands/killcursors_cmd.cpp b/src/mongo/db/commands/killcursors_cmd.cpp
index 52fdf7a6803..065cc40d0c9 100644
--- a/src/mongo/db/commands/killcursors_cmd.cpp
+++ b/src/mongo/db/commands/killcursors_cmd.cpp
@@ -66,12 +66,11 @@ private:
                        CursorId id) const final {
         boost::optional<AutoStatsTracker> statsTracker;
         if (!nss.isCollectionlessCursorNamespace()) {
-            const boost::optional<int> dbProfilingLevel = boost::none;
             statsTracker.emplace(opCtx,
                                  nss,
                                  Top::LockType::NotLocked,
                                  AutoStatsTracker::LogMode::kUpdateTopAndCurOp,
-                                 dbProfilingLevel);
+                                 CollectionCatalog::get(opCtx).getDatabaseProfileLevel(nss.db()));
         }
 
         auto cursorManager = CursorManager::get(opCtx);
diff --git a/src/mongo/db/curop.cpp b/src/mongo/db/curop.cpp
index 301d3cdd5c8..8a9ef9be44d 100644
--- a/src/mongo/db/curop.cpp
+++ b/src/mongo/db/curop.cpp
@@ -412,12 +412,10 @@ void CurOp::ensureStarted() {
     }
 }
 
-void CurOp::enter_inlock(const char* ns, boost::optional<int> dbProfileLevel) {
+void CurOp::enter_inlock(const char* ns, int dbProfileLevel) {
     ensureStarted();
     _ns = ns;
-    if (dbProfileLevel) {
-        raiseDbProfileLevel(*dbProfileLevel);
-    }
+    raiseDbProfileLevel(dbProfileLevel);
 }
 
 void CurOp::raiseDbProfileLevel(int dbProfileLevel) {
diff --git a/src/mongo/db/curop.h b/src/mongo/db/curop.h
index 54762580f79..db9fd2c1350 100644
--- a/src/mongo/db/curop.h
+++ b/src/mongo/db/curop.h
@@ -365,7 +365,7 @@ public:
         return _originatingCommand;
     }
 
-    void enter_inlock(const char* ns, boost::optional<int> dbProfileLevel);
+    void enter_inlock(const char* ns, int dbProfileLevel);
 
     /**
      * Sets the type of the current network operation.
diff --git a/src/mongo/db/db_raii.cpp b/src/mongo/db/db_raii.cpp
index 53dda3ca490..f15adadabf3 100644
--- a/src/mongo/db/db_raii.cpp
+++ b/src/mongo/db/db_raii.cpp
@@ -59,25 +59,13 @@ AutoStatsTracker::AutoStatsTracker(OperationContext* opCtx,
                                    const NamespaceString& nss,
                                    Top::LockType lockType,
                                    LogMode logMode,
-                                   boost::optional<int> dbProfilingLevel,
+                                   int dbProfilingLevel,
                                    Date_t deadline)
     : _opCtx(opCtx), _lockType(lockType), _nss(nss), _logMode(logMode) {
     if (_logMode == LogMode::kUpdateTop) {
         return;
     }
 
-    if (!dbProfilingLevel) {
-        // No profiling level was determined, attempt to read the profiling level from the Database
-        // object. Since we are only reading the in-memory profiling level out of the database
-        // object (which is configured on a per-node basis and not replicated or persisted), we
-        // never need to conflict with secondary batch application.
-        ShouldNotConflictWithSecondaryBatchApplicationBlock noConflict(opCtx->lockState());
-        AutoGetDb autoDb(_opCtx, _nss.db(), MODE_IS, deadline);
-        if (autoDb.getDb()) {
-            dbProfilingLevel = autoDb.getDb()->getProfilingLevel();
-        }
-    }
-
     stdx::lock_guard<Client> clientLock(*_opCtx->getClient());
     CurOp::get(_opCtx)->enter_inlock(_nss.ns().c_str(), dbProfilingLevel);
 }
@@ -256,13 +244,13 @@ AutoGetCollectionForReadCommand::AutoGetCollectionForReadCommand(
     Date_t deadline,
     AutoStatsTracker::LogMode logMode)
     : _autoCollForRead(opCtx, nsOrUUID, viewMode, deadline),
-      _statsTracker(opCtx,
-                    _autoCollForRead.getNss(),
-                    Top::LockType::ReadLocked,
-                    logMode,
-                    _autoCollForRead.getDb() ? _autoCollForRead.getDb()->getProfilingLevel()
-                                             : kDoNotChangeProfilingLevel,
-                    deadline) {
+      _statsTracker(
+          opCtx,
+          _autoCollForRead.getNss(),
+          Top::LockType::ReadLocked,
+          logMode,
+          CollectionCatalog::get(opCtx).getDatabaseProfileLevel(_autoCollForRead.getNss().db()),
+          deadline) {
 
     if (!_autoCollForRead.getView()) {
         auto* const css = CollectionShardingState::get(opCtx, _autoCollForRead.getNss());
@@ -294,7 +282,8 @@ OldClientContext::OldClientContext(OperationContext* opCtx, const std::string& n
     }
 
     stdx::lock_guard<Client> lk(*_opCtx->getClient());
-    currentOp->enter_inlock(ns.c_str(), _db->getProfilingLevel());
+    currentOp->enter_inlock(ns.c_str(),
+                            CollectionCatalog::get(opCtx).getDatabaseProfileLevel(_db->name()));
 }
 
 OldClientContext::~OldClientContext() {
diff --git a/src/mongo/db/db_raii.h b/src/mongo/db/db_raii.h
index 51208643c6a..d1e9e85cb4e 100644
--- a/src/mongo/db/db_raii.h
+++ b/src/mongo/db/db_raii.h
@@ -68,7 +68,7 @@ public:
                      const NamespaceString& nss,
                      Top::LockType lockType,
                      LogMode logMode,
-                     boost::optional<int> dbProfilingLevel,
+                     int dbProfilingLevel,
                      Date_t deadline = Date_t::max());
 
     /**
diff --git a/src/mongo/db/ops/write_ops_exec.cpp b/src/mongo/db/ops/write_ops_exec.cpp
index 061658ca4a1..ce348f9c144 100644
--- a/src/mongo/db/ops/write_ops_exec.cpp
+++ b/src/mongo/db/ops/write_ops_exec.cpp
@@ -396,7 +396,8 @@ bool insertBatchAndHandleErrors(OperationContext* opCtx,
             makeCollection(opCtx, wholeOp.getNamespace());
         }
 
-        curOp.raiseDbProfileLevel(collection->getDb()->getProfilingLevel());
+        curOp.raiseDbProfileLevel(
+            CollectionCatalog::get(opCtx).getDatabaseProfileLevel(wholeOp.getNamespace().db()));
         assertCanWrite_inlock(opCtx, wholeOp.getNamespace());
 
         CurOpFailpointHelpers::waitWhileFailPointEnabled(
@@ -661,7 +662,7 @@ static SingleWriteResult performSingleUpdateOp(OperationContext* opCtx,
     auto& curOp = *CurOp::get(opCtx);
 
     if (collection->getDb()) {
-        curOp.raiseDbProfileLevel(collection->getDb()->getProfilingLevel());
+        curOp.raiseDbProfileLevel(CollectionCatalog::get(opCtx).getDatabaseProfileLevel(ns.db()));
     }
 
     assertCanWrite_inlock(opCtx, ns);
@@ -896,7 +897,7 @@ static SingleWriteResult performSingleDeleteOp(OperationContext* opCtx,
     AutoGetCollection collection(opCtx, ns, fixLockModeForSystemDotViewsChanges(ns, MODE_IX));
 
     if (collection.getDb()) {
-        curOp.raiseDbProfileLevel(collection.getDb()->getProfilingLevel());
+        curOp.raiseDbProfileLevel(CollectionCatalog::get(opCtx).getDatabaseProfileLevel(ns.db()));
     }
 
     assertCanWrite_inlock(opCtx, ns);
diff --git a/src/mongo/db/query/find.cpp b/src/mongo/db/query/find.cpp
index 87a85b628d4..5e28db4d8c8 100644
--- a/src/mongo/db/query/find.cpp
+++ b/src/mongo/db/query/find.cpp
@@ -296,14 +296,11 @@ Message getMore(OperationContext* opCtx,
     if (cursorPin->lockPolicy() == ClientCursorParams::LockPolicy::kLocksInternally) {
         if (!nss.isCollectionlessCursorNamespace()) {
             AutoGetDb autoDb(opCtx, nss.db(), MODE_IS);
-            const auto profilingLevel = autoDb.getDb()
-                ? boost::optional<int>{autoDb.getDb()->getProfilingLevel()}
-                : boost::none;
             statsTracker.emplace(opCtx,
                                  nss,
                                  Top::LockType::NotLocked,
                                  AutoStatsTracker::LogMode::kUpdateTopAndCurOp,
-                                 profilingLevel);
+                                 CollectionCatalog::get(opCtx).getDatabaseProfileLevel(nss.db()));
             auto view = autoDb.getDb() ? ViewCatalog::get(autoDb.getDb())->lookup(opCtx, nss.ns())
                                        : nullptr;
             uassert(
@@ -316,13 +313,11 @@ Message getMore(OperationContext* opCtx,
         }
     } else {
         readLock.emplace(opCtx, nss);
-        const int doNotChangeProfilingLevel = 0;
         statsTracker.emplace(opCtx,
                              nss,
                              Top::LockType::ReadLocked,
                              AutoStatsTracker::LogMode::kUpdateTopAndCurOp,
-                             readLock->getDb() ? readLock->getDb()->getProfilingLevel()
-                                               : doNotChangeProfilingLevel);
+                             CollectionCatalog::get(opCtx).getDatabaseProfileLevel(nss.db()));
 
         // This checks to make sure the operation is allowed on a replicated node.  Since we are not
         // passing in a query object (necessary to check SlaveOK query option), we allow reads
diff --git a/src/mongo/db/run_op_kill_cursors.cpp b/src/mongo/db/run_op_kill_cursors.cpp
index 4b0733ef9d6..db251bee5c0 100644
--- a/src/mongo/db/run_op_kill_cursors.cpp
+++ b/src/mongo/db/run_op_kill_cursors.cpp
@@ -59,12 +59,11 @@ bool killCursorIfAuthorized(OperationContext* opCtx, CursorId id) {
 
     boost::optional<AutoStatsTracker> statsTracker;
     if (!nss.isCollectionlessCursorNamespace()) {
-        const boost::optional<int> dbProfilingLevel = boost::none;
         statsTracker.emplace(opCtx,
                              nss,
                              Top::LockType::NotLocked,
                              AutoStatsTracker::LogMode::kUpdateTopAndCurOp,
-                             dbProfilingLevel);
+                             CollectionCatalog::get(opCtx).getDatabaseProfileLevel(nss.db()));
     }
 
     AuthorizationSession* as = AuthorizationSession::get(opCtx->getClient());
diff --git a/src/mongo/db/storage/wiredtiger/oplog_stones_server_status_section.cpp b/src/mongo/db/storage/wiredtiger/oplog_stones_server_status_section.cpp
index fef46b4348e..71d2421f5e5 100644
--- a/src/mongo/db/storage/wiredtiger/oplog_stones_server_status_section.cpp
+++ b/src/mongo/db/storage/wiredtiger/oplog_stones_server_status_section.cpp
@@ -57,13 +57,15 @@ public:
         AutoGetOplog oplogRead(opCtx, OplogAccessMode::kRead);
         auto oplog = oplogRead.getCollection();
         if (oplog) {
-            const auto localDb = DatabaseHolder::get(opCtx)->getDb(opCtx, "local");
+            const auto localDb =
+                DatabaseHolder::get(opCtx)->getDb(opCtx, NamespaceString::kLocalDb);
             invariant(localDb);
-            AutoStatsTracker statsTracker(opCtx,
-                                          NamespaceString::kRsOplogNamespace,
-                                          Top::LockType::ReadLocked,
-                                          AutoStatsTracker::LogMode::kUpdateTop,
-                                          localDb->getProfilingLevel());
+            AutoStatsTracker statsTracker(
+                opCtx,
+                NamespaceString::kRsOplogNamespace,
+                Top::LockType::ReadLocked,
+                AutoStatsTracker::LogMode::kUpdateTop,
+                CollectionCatalog::get(opCtx).getDatabaseProfileLevel(NamespaceString::kLocalDb));
             oplog->getRecordStore()->getOplogTruncateStats(builder);
         }
         return builder.obj();
diff --git a/src/mongo/db/transaction_history_iterator.cpp b/src/mongo/db/transaction_history_iterator.cpp
index 07f794238a4..7fae8ea2da1 100644
--- a/src/mongo/db/transaction_history_iterator.cpp
+++ b/src/mongo/db/transaction_history_iterator.cpp
@@ -77,14 +77,15 @@ BSONObj findOneOplogEntry(OperationContext* opCtx,
     std::unique_ptr<CanonicalQuery> cq = std::move(statusWithCQ.getValue());
 
     AutoGetOplog oplogRead(opCtx, OplogAccessMode::kRead);
-    const auto localDb = DatabaseHolder::get(opCtx)->getDb(opCtx, "local");
+    const auto localDb = DatabaseHolder::get(opCtx)->getDb(opCtx, NamespaceString::kLocalDb);
     invariant(localDb);
-    AutoStatsTracker statsTracker(opCtx,
-                                  NamespaceString::kRsOplogNamespace,
-                                  Top::LockType::ReadLocked,
-                                  AutoStatsTracker::LogMode::kUpdateTop,
-                                  localDb->getProfilingLevel(),
-                                  Date_t::max());
+    AutoStatsTracker statsTracker(
+        opCtx,
+        NamespaceString::kRsOplogNamespace,
+        Top::LockType::ReadLocked,
+        AutoStatsTracker::LogMode::kUpdateTop,
+        CollectionCatalog::get(opCtx).getDatabaseProfileLevel(NamespaceString::kLocalDb),
+        Date_t::max());
 
     auto exec = uassertStatusOK(
         getExecutorFind(opCtx, oplogRead.getCollection(), std::move(cq), permitYield));