SERVER-50863 Collect resource consumption metrics for specified commands

23 files changed, 373 insertions, 17 deletions

diff --git a/src/mongo/db/SConscript b/src/mongo/db/SConscript
index 701e82a657f..0d27639c637 100644
--- a/src/mongo/db/SConscript
+++ b/src/mongo/db/SConscript
@@ -1249,6 +1249,7 @@ env.Library(
         'catalog/database_holder',
         'commands/server_status_core',
         'kill_sessions',
+        'stats/resource_consumption_metrics',
     ],
 )
 
diff --git a/src/mongo/db/commands.h b/src/mongo/db/commands.h
index 06803dbad84..a70dcad61b8 100644
--- a/src/mongo/db/commands.h
+++ b/src/mongo/db/commands.h
@@ -409,6 +409,13 @@ public:
     }
 
     /**
+      Returns true if this command collects operation resource consumption metrics.
+     */
+    virtual bool collectsResourceConsumptionMetrics() const {
+        return false;
+    }
+
+    /**
      * Return true if the command requires auth.
      */
     virtual bool requiresAuth() const {
diff --git a/src/mongo/db/commands/collection_to_capped.cpp b/src/mongo/db/commands/collection_to_capped.cpp
index 1ac73e3da17..175f0629ffe 100644
--- a/src/mongo/db/commands/collection_to_capped.cpp
+++ b/src/mongo/db/commands/collection_to_capped.cpp
@@ -149,6 +149,11 @@ public:
     virtual bool supportsWriteConcern(const BSONObj& cmd) const override {
         return true;
     }
+
+    bool collectsResourceConsumptionMetrics() const override {
+        return true;
+    }
+
     std::string help() const override {
         return "{ convertToCapped:<fromCollectionName>, size:<sizeInBytes> }";
     }
diff --git a/src/mongo/db/commands/count_cmd.cpp b/src/mongo/db/commands/count_cmd.cpp
index a450e2cbebe..2014df3dcc7 100644
--- a/src/mongo/db/commands/count_cmd.cpp
+++ b/src/mongo/db/commands/count_cmd.cpp
@@ -75,6 +75,10 @@ public:
         return false;
     }
 
+    bool collectsResourceConsumptionMetrics() const override {
+        return true;
+    }
+
     bool canIgnorePrepareConflicts() const override {
         return true;
     }
diff --git a/src/mongo/db/commands/create_indexes.cpp b/src/mongo/db/commands/create_indexes.cpp
index 0f3b83667cd..c3bf793ca6b 100644
--- a/src/mongo/db/commands/create_indexes.cpp
+++ b/src/mongo/db/commands/create_indexes.cpp
@@ -737,6 +737,10 @@ public:
         return true;
     }
 
+    bool collectsResourceConsumptionMetrics() const override {
+        return true;
+    }
+
     AllowedOnSecondary secondaryAllowed(ServiceContext*) const override {
         return AllowedOnSecondary::kNever;
     }
diff --git a/src/mongo/db/commands/dbcommands.cpp b/src/mongo/db/commands/dbcommands.cpp
index 06634e1e68b..7a7a57d1f0c 100644
--- a/src/mongo/db/commands/dbcommands.cpp
+++ b/src/mongo/db/commands/dbcommands.cpp
@@ -129,6 +129,10 @@ public:
         return true;
     }
 
+    bool collectsResourceConsumptionMetrics() const override {
+        return true;
+    }
+
     CmdDropDatabase() : BasicCommand("dropDatabase") {}
 
     bool run(OperationContext* opCtx,
@@ -235,6 +239,10 @@ public:
         return true;
     }
 
+    bool collectsResourceConsumptionMetrics() const override {
+        return true;
+    }
+
     virtual bool errmsgRun(OperationContext* opCtx,
                            const string& dbname,
                            const BSONObj& cmdObj,
@@ -291,6 +299,10 @@ public:
         return true;
     }
 
+    bool collectsResourceConsumptionMetrics() const override {
+        return true;
+    }
+
     std::string help() const override {
         return str::stream()
             << "explicitly creates a collection or view\n"
@@ -658,6 +670,11 @@ public:
     virtual bool supportsWriteConcern(const BSONObj& cmd) const override {
         return true;
     }
+
+    bool collectsResourceConsumptionMetrics() const override {
+        return true;
+    }
+
     std::string help() const override {
         return "Sets collection options.\n"
                "Example: { collMod: 'foo', viewOn: 'bar'} "
diff --git a/src/mongo/db/commands/dbcommands_d.cpp b/src/mongo/db/commands/dbcommands_d.cpp
index ff50ee21f25..944a63b39bd 100644
--- a/src/mongo/db/commands/dbcommands_d.cpp
+++ b/src/mongo/db/commands/dbcommands_d.cpp
@@ -223,6 +223,10 @@ public:
         return false;
     }
 
+    bool collectsResourceConsumptionMetrics() const override {
+        return true;
+    }
+
     virtual std::string parseNs(const std::string& dbname, const BSONObj& cmdObj) const {
         std::string collectionName;
         if (const auto rootElt = cmdObj["root"]) {
diff --git a/src/mongo/db/commands/distinct.cpp b/src/mongo/db/commands/distinct.cpp
index 0b332721625..6ab423127d9 100644
--- a/src/mongo/db/commands/distinct.cpp
+++ b/src/mongo/db/commands/distinct.cpp
@@ -84,6 +84,10 @@ public:
         return false;
     }
 
+    bool collectsResourceConsumptionMetrics() const override {
+        return true;
+    }
+
     bool canIgnorePrepareConflicts() const override {
         return true;
     }
diff --git a/src/mongo/db/commands/drop_indexes.cpp b/src/mongo/db/commands/drop_indexes.cpp
index e8bd2670229..eebf122c494 100644
--- a/src/mongo/db/commands/drop_indexes.cpp
+++ b/src/mongo/db/commands/drop_indexes.cpp
@@ -79,6 +79,11 @@ public:
     virtual bool supportsWriteConcern(const BSONObj& cmd) const override {
         return true;
     }
+
+    bool collectsResourceConsumptionMetrics() const override {
+        return true;
+    }
+
     std::string help() const override {
         return "drop indexes for a collection";
     }
diff --git a/src/mongo/db/commands/find_and_modify.cpp b/src/mongo/db/commands/find_and_modify.cpp
index 6ee903eb59b..ea36a9dcf1c 100644
--- a/src/mongo/db/commands/find_and_modify.cpp
+++ b/src/mongo/db/commands/find_and_modify.cpp
@@ -244,6 +244,10 @@ public:
         return true;
     }
 
+    bool collectsResourceConsumptionMetrics() const override {
+        return true;
+    }
+
     void addRequiredPrivileges(const std::string& dbname,
                                const BSONObj& cmdObj,
                                std::vector<Privilege>* out) const override {
diff --git a/src/mongo/db/commands/find_cmd.cpp b/src/mongo/db/commands/find_cmd.cpp
index 3f4d7d840df..2cae02d0a10 100644
--- a/src/mongo/db/commands/find_cmd.cpp
+++ b/src/mongo/db/commands/find_cmd.cpp
@@ -178,6 +178,10 @@ public:
         return true;
     }
 
+    bool collectsResourceConsumptionMetrics() const override {
+        return true;
+    }
+
     class Invocation final : public CommandInvocation {
     public:
         Invocation(const FindCmd* definition, const OpMsgRequest& request, StringData dbName)
diff --git a/src/mongo/db/commands/getmore_cmd.cpp b/src/mongo/db/commands/getmore_cmd.cpp
index bc4fafc8070..0285be5ab5c 100644
--- a/src/mongo/db/commands/getmore_cmd.cpp
+++ b/src/mongo/db/commands/getmore_cmd.cpp
@@ -746,6 +746,10 @@ public:
         return FindCommon::kMaxBytesToReturnToClientAtOnce + 1024u;
     }
 
+    bool collectsResourceConsumptionMetrics() const override {
+        return true;
+    }
+
     /**
      * A getMore command increments the getMore counter, not the command counter.
      */
diff --git a/src/mongo/db/commands/list_collections.cpp b/src/mongo/db/commands/list_collections.cpp
index d080f8313db..ba05d17ff45 100644
--- a/src/mongo/db/commands/list_collections.cpp
+++ b/src/mongo/db/commands/list_collections.cpp
@@ -228,6 +228,10 @@ public:
         return false;
     }
 
+    bool collectsResourceConsumptionMetrics() const override {
+        return true;
+    }
+
     std::string help() const final {
         return "list collections for this db";
     }
diff --git a/src/mongo/db/commands/list_indexes.cpp b/src/mongo/db/commands/list_indexes.cpp
index e793ddf2f6f..bd2eab3b3fc 100644
--- a/src/mongo/db/commands/list_indexes.cpp
+++ b/src/mongo/db/commands/list_indexes.cpp
@@ -111,6 +111,10 @@ public:
         return false;
     }
 
+    bool collectsResourceConsumptionMetrics() const override {
+        return true;
+    }
+
     std::string help() const override {
         return "list indexes for a collection";
     }
diff --git a/src/mongo/db/commands/pipeline_command.cpp b/src/mongo/db/commands/pipeline_command.cpp
index 6b542b421ae..738d161ce65 100644
--- a/src/mongo/db/commands/pipeline_command.cpp
+++ b/src/mongo/db/commands/pipeline_command.cpp
@@ -86,6 +86,10 @@ public:
         return true;
     }
 
+    bool collectsResourceConsumptionMetrics() const override {
+        return true;
+    }
+
     class Invocation final : public CommandInvocation {
     public:
         Invocation(Command* cmd,
diff --git a/src/mongo/db/commands/rename_collection_cmd.cpp b/src/mongo/db/commands/rename_collection_cmd.cpp
index 35c19fd028d..fbad05deeb2 100644
--- a/src/mongo/db/commands/rename_collection_cmd.cpp
+++ b/src/mongo/db/commands/rename_collection_cmd.cpp
@@ -67,6 +67,11 @@ public:
     virtual bool supportsWriteConcern(const BSONObj& cmd) const override {
         return true;
     }
+
+    bool collectsResourceConsumptionMetrics() const override {
+        return true;
+    }
+
     virtual Status checkAuthForCommand(Client* client,
                                        const std::string& dbname,
                                        const BSONObj& cmdObj) const {
diff --git a/src/mongo/db/commands/txn_cmds.cpp b/src/mongo/db/commands/txn_cmds.cpp
index df7aa9787dc..49ce490a353 100644
--- a/src/mongo/db/commands/txn_cmds.cpp
+++ b/src/mongo/db/commands/txn_cmds.cpp
@@ -77,6 +77,10 @@ public:
         return true;
     }
 
+    bool collectsResourceConsumptionMetrics() const override {
+        return true;
+    }
+
     std::string help() const override {
         return "Commits a transaction";
     }
@@ -181,6 +185,10 @@ public:
         return true;
     }
 
+    bool collectsResourceConsumptionMetrics() const override {
+        return true;
+    }
+
     ReadConcernSupportResult supportsReadConcern(const BSONObj& cmdObj,
                                                  repl::ReadConcernLevel level) const override {
         // abortTransaction commences running inside a transaction (even though the transaction will
diff --git a/src/mongo/db/commands/write_commands/write_commands.cpp b/src/mongo/db/commands/write_commands/write_commands.cpp
index bacfbd5dfd8..a016aef5de2 100644
--- a/src/mongo/db/commands/write_commands/write_commands.cpp
+++ b/src/mongo/db/commands/write_commands/write_commands.cpp
@@ -220,6 +220,10 @@ private:
         return false;
     }
 
+    bool collectsResourceConsumptionMetrics() const override {
+        return true;
+    }
+
     ReadWriteType getReadWriteType() const final {
         return ReadWriteType::kWrite;
     }
diff --git a/src/mongo/db/query/find.cpp b/src/mongo/db/query/find.cpp
index 38c82127276..fe77d388ff2 100644
--- a/src/mongo/db/query/find.cpp
+++ b/src/mongo/db/query/find.cpp
@@ -60,6 +60,7 @@
 #include "mongo/db/s/collection_sharding_state.h"
 #include "mongo/db/server_options.h"
 #include "mongo/db/service_context.h"
+#include "mongo/db/stats/resource_consumption_metrics.h"
 #include "mongo/db/stats/top.h"
 #include "mongo/db/storage/storage_options.h"
 #include "mongo/db/views/view_catalog.h"
@@ -227,6 +228,12 @@ Message getMore(OperationContext* opCtx,
 
     const NamespaceString nss(ns);
 
+    ResourceConsumption::ScopedMetricsCollector scopedMetrics(opCtx);
+    if (ResourceConsumption::shouldCollectMetricsForDatabase(nss.db())) {
+        auto& opMetrics = ResourceConsumption::MetricsCollector::get(opCtx);
+        opMetrics.setDbName(nss.db().toString());
+    }
+
     // Cursors come in one of two flavors:
     //
     // - Cursors which read from a single collection, such as those generated via the find command.
@@ -566,6 +573,12 @@ bool runQuery(OperationContext* opCtx,
             nss.isValid());
     invariant(!nss.isCommand());
 
+    ResourceConsumption::ScopedMetricsCollector scopedMetrics(opCtx);
+    if (ResourceConsumption::shouldCollectMetricsForDatabase(nss.db())) {
+        auto& opMetrics = ResourceConsumption::MetricsCollector::get(opCtx);
+        opMetrics.setDbName(nss.db().toString());
+    }
+
     // Set CurOp information.
     const auto upconvertedQuery = upconvertQueryEntry(q.query, nss, q.ntoreturn, q.ntoskip);
 
diff --git a/src/mongo/db/service_entry_point_common.cpp b/src/mongo/db/service_entry_point_common.cpp
index f01218bba70..77dda9f44a5 100644
--- a/src/mongo/db/service_entry_point_common.cpp
+++ b/src/mongo/db/service_entry_point_common.cpp
@@ -81,6 +81,7 @@
 #include "mongo/db/session_catalog_mongod.h"
 #include "mongo/db/stats/api_version_metrics.h"
 #include "mongo/db/stats/counters.h"
+#include "mongo/db/stats/resource_consumption_metrics.h"
 #include "mongo/db/stats/server_read_concern_metrics.h"
 #include "mongo/db/stats/top.h"
 #include "mongo/db/transaction_participant.h"
@@ -1037,6 +1038,12 @@ void execCommandDatabase(OperationContext* opCtx,
             str::stream() << "Invalid database name: '" << dbname << "'",
             NamespaceString::validDBName(dbname, NamespaceString::DollarInDbNameBehavior::Allow));
 
+        ResourceConsumption::ScopedMetricsCollector scopedMetrics(
+            opCtx, command->collectsResourceConsumptionMetrics());
+        if (ResourceConsumption::shouldCollectMetricsForDatabase(dbname)) {
+            auto& opMetrics = ResourceConsumption::MetricsCollector::get(opCtx);
+            opMetrics.setDbName(dbname);
+        }
 
         const auto allowTransactionsOnConfigDatabase =
             (serverGlobalParams.clusterRole == ClusterRole::ConfigServer ||
diff --git a/src/mongo/db/stats/resource_consumption_metrics.cpp b/src/mongo/db/stats/resource_consumption_metrics.cpp
index a5bb969042a..712704aafb0 100644
--- a/src/mongo/db/stats/resource_consumption_metrics.cpp
+++ b/src/mongo/db/stats/resource_consumption_metrics.cpp
@@ -29,17 +29,63 @@
 
 #include "mongo/db/stats/resource_consumption_metrics.h"
 
+#include "mongo/db/stats/operation_resource_consumption_gen.h"
+
 namespace mongo {
 namespace {
-const OperationContext::Decoration<ResourceConsumption::Metrics> getOperationMetrics =
-    OperationContext::declareDecoration<ResourceConsumption::Metrics>();
+const OperationContext::Decoration<ResourceConsumption::MetricsCollector> getMetricsCollector =
+    OperationContext::declareDecoration<ResourceConsumption::MetricsCollector>();
 const ServiceContext::Decoration<ResourceConsumption> getGlobalResourceConsumption =
     ServiceContext::declareDecoration<ResourceConsumption>();
 }  // namespace
 
 
-ResourceConsumption::Metrics& ResourceConsumption::Metrics::get(OperationContext* opCtx) {
-    return getOperationMetrics(opCtx);
+ResourceConsumption::MetricsCollector& ResourceConsumption::MetricsCollector::get(
+    OperationContext* opCtx) {
+    return getMetricsCollector(opCtx);
+}
+
+ResourceConsumption::ScopedMetricsCollector::ScopedMetricsCollector(OperationContext* opCtx,
+                                                                    bool commandCollectsMetrics)
+    : _opCtx(opCtx) {
+
+    // Nesting is allowed but does nothing. Lower-level ScopedMetricsCollectors should not influence
+    // the top-level Collector's behavior.
+    auto& metrics = MetricsCollector::get(opCtx);
+    _topLevel = !metrics.isInScope();
+    if (!_topLevel) {
+        return;
+    }
+
+    if (!commandCollectsMetrics || !gMeasureOperationResourceConsumption) {
+        metrics.beginScopedNotCollecting();
+        return;
+    }
+
+    metrics.beginScopedCollecting();
+}
+
+ResourceConsumption::ScopedMetricsCollector::~ScopedMetricsCollector() {
+    if (!_topLevel) {
+        return;
+    }
+
+    auto& collector = MetricsCollector::get(_opCtx);
+    bool wasCollecting = collector.endScopedCollecting();
+    if (!wasCollecting) {
+        return;
+    }
+
+    if (collector.getDbName().empty()) {
+        return;
+    }
+
+    if (!gAggregateOperationResourceConsumption) {
+        return;
+    }
+
+    auto& globalResourceConsumption = ResourceConsumption::get(_opCtx);
+    globalResourceConsumption.add(collector);
 }
 
 ResourceConsumption& ResourceConsumption::get(ServiceContext* svcCtx) {
@@ -50,10 +96,10 @@ ResourceConsumption& ResourceConsumption::get(OperationContext* opCtx) {
     return getGlobalResourceConsumption(opCtx->getServiceContext());
 }
 
-void ResourceConsumption::add(const Metrics& metrics) {
-    invariant(!metrics.getDbName().empty());
+void ResourceConsumption::add(const MetricsCollector& collector) {
+    invariant(!collector.getDbName().empty());
     stdx::unique_lock<Mutex> lk(_mutex);
-    _metrics[metrics.getDbName()] += metrics;
+    _metrics[collector.getDbName()] += collector.getMetrics();
 }
 
 ResourceConsumption::MetricsMap ResourceConsumption::getMetrics() const {
diff --git a/src/mongo/db/stats/resource_consumption_metrics.h b/src/mongo/db/stats/resource_consumption_metrics.h
index 257039ca17f..3b0f14bb36f 100644
--- a/src/mongo/db/stats/resource_consumption_metrics.h
+++ b/src/mongo/db/stats/resource_consumption_metrics.h
@@ -32,6 +32,7 @@
 #include <map>
 #include <string>
 
+#include "mongo/db/namespace_string.h"
 #include "mongo/db/operation_context.h"
 #include "mongo/platform/mutex.h"
 
@@ -46,13 +47,10 @@ public:
     static ResourceConsumption& get(ServiceContext* svcCtx);
 
     /**
-     * Metrics maintains non-thread-safe, per-operation resource consumption metrics for a specific
-     * database.
+     * Metrics maintains a set of resource consumption metrics.
      */
     class Metrics {
     public:
-        static Metrics& get(OperationContext* opCtx);
-
         /**
          * Adds other Metrics to this one.
          */
@@ -61,6 +59,53 @@ public:
             add(other);
             return *this;
         }
+    };
+
+    /**
+     * MetricsCollector maintains non-thread-safe, per-operation resource consumption metrics for a
+     * specific database.
+     */
+    class MetricsCollector {
+    public:
+        static MetricsCollector& get(OperationContext* opCtx);
+
+        /**
+         * When called, resource consumption metrics should be recorded and added to the global
+         * structure.
+         */
+        void beginScopedCollecting() {
+            invariant(!isInScope());
+            _collecting = ScopedCollectionState::kInScopeCollecting;
+        }
+
+        /**
+         * When called, sets state that a ScopedMetricsCollector is in scope, but is not recording
+         * metrics. This is to support nesting Scope objects and preventing lower levels from
+         * overriding this behavior.
+         */
+        void beginScopedNotCollecting() {
+            invariant(!isInScope());
+            _collecting = ScopedCollectionState::kInScopeNotCollecting;
+        }
+
+        /**
+         * When called, resource consumption metrics should not be recorded. Returns whether this
+         * Collector was in a collecting state.
+         */
+        bool endScopedCollecting() {
+            bool wasCollecting = isCollecting();
+            _collecting = ScopedCollectionState::kInactive;
+            return wasCollecting;
+        }
+
+        bool isCollecting() const {
+            return _collecting == ScopedCollectionState::kInScopeCollecting;
+        }
+
+        bool isInScope() const {
+            return _collecting == ScopedCollectionState::kInScopeCollecting ||
+                _collecting == ScopedCollectionState::kInScopeNotCollecting;
+        }
 
         /**
          * Set the database name associated with these metrics.
@@ -73,17 +118,71 @@ public:
             return _dbName;
         }
 
+        /**
+         * To observe the stored Metrics, the dbName must be set. This prevents "losing" collected
+         * Metrics due to the Collector stopping without being associated with any database yet.
+         */
+        Metrics& getMetrics() {
+            invariant(!_dbName.empty(), "observing Metrics before a dbName has been set");
+            return _metrics;
+        }
+
+        const Metrics& getMetrics() const {
+            invariant(!_dbName.empty(), "observing Metrics before a dbName has been set");
+            return _metrics;
+        }
+
     private:
+        /**
+         * Represents the ScopedMetricsCollector state.
+         */
+        enum class ScopedCollectionState {
+            // No ScopedMetricsCollector is in scope
+            kInactive,
+            // A ScopedMetricsCollector is in scope but not collecting metrics
+            kInScopeNotCollecting,
+            // A ScopedMetricsCollector is in scope and collecting metrics
+            kInScopeCollecting
+        };
+        ScopedCollectionState _collecting = ScopedCollectionState::kInactive;
         std::string _dbName;
+        Metrics _metrics;
+    };
+
+    /**
+     * When instantiated with commandCollectsMetrics=true, enables operation resource consumption
+     * collection. When destructed, appends collected metrics to the global structure.
+     */
+    class ScopedMetricsCollector {
+    public:
+        ScopedMetricsCollector(OperationContext* opCtx, bool commandCollectsMetrics);
+        ScopedMetricsCollector(OperationContext* opCtx) : ScopedMetricsCollector(opCtx, true) {}
+        ~ScopedMetricsCollector();
+
+    private:
+        bool _topLevel;
+        OperationContext* _opCtx;
     };
 
     /**
-     * Adds a Metrics object to an existing Metrics object in the map, keyed by database name. If no
-     * Metrics exist for the database, the value is initialized with the provided Metrics.
+     * Returns whether the database's metrics should be collected.
+     */
+    static bool shouldCollectMetricsForDatabase(StringData dbName) {
+        if (dbName == NamespaceString::kAdminDb || dbName == NamespaceString::kConfigDb ||
+            dbName == NamespaceString::kLocalDb) {
+            return false;
+        }
+        return true;
+    }
+
+    /**
+     * Adds a MetricsCollector's Metrics to an existing Metrics object in the map, keyed by
+     * database name. If no Metrics exist for the database, the value is initialized with the
+     * provided MetricsCollector's Metrics.
      *
-     * The Metrics database name must not be an empty string.
+     * The MetricsCollector's database name must not be an empty string.
      */
-    void add(const Metrics& metrics);
+    void add(const MetricsCollector& metrics);
 
     /**
      * Returns a copy of the Metrics map.
diff --git a/src/mongo/db/stats/resource_consumption_metrics_test.cpp b/src/mongo/db/stats/resource_consumption_metrics_test.cpp
index 093170ec199..58f95bf1448 100644
--- a/src/mongo/db/stats/resource_consumption_metrics_test.cpp
+++ b/src/mongo/db/stats/resource_consumption_metrics_test.cpp
@@ -30,6 +30,7 @@
 #include "mongo/platform/basic.h"
 
 #include "mongo/db/service_context_test_fixture.h"
+#include "mongo/db/stats/operation_resource_consumption_gen.h"
 #include "mongo/db/stats/resource_consumption_metrics.h"
 #include "mongo/unittest/unittest.h"
 
@@ -39,6 +40,8 @@ class ResourceConsumptionMetricsTest : public ServiceContextTest {
 public:
     void setUp() {
         _opCtx = makeOperationContext();
+        gMeasureOperationResourceConsumption = true;
+        gAggregateOperationResourceConsumption = true;
     }
 
     typedef std::pair<ServiceContext::UniqueClient, ServiceContext::UniqueOperationContext>
@@ -58,10 +61,12 @@ TEST_F(ResourceConsumptionMetricsTest, Add) {
 
     auto [client2, opCtx2] = makeClientAndCtx("opCtx2");
 
-    auto& operationMetrics1 = ResourceConsumption::Metrics::get(_opCtx.get());
-    auto& operationMetrics2 = ResourceConsumption::Metrics::get(opCtx2.get());
+    auto& operationMetrics1 = ResourceConsumption::MetricsCollector::get(_opCtx.get());
+    auto& operationMetrics2 = ResourceConsumption::MetricsCollector::get(opCtx2.get());
 
+    operationMetrics1.beginScopedCollecting();
     operationMetrics1.setDbName("db1");
+    operationMetrics2.beginScopedCollecting();
     operationMetrics2.setDbName("db2");
     globalResourceConsumption.add(operationMetrics1);
     globalResourceConsumption.add(operationMetrics1);
@@ -73,4 +78,98 @@ TEST_F(ResourceConsumptionMetricsTest, Add) {
     ASSERT_EQ(globalMetrics.count("db2"), 1);
     ASSERT_EQ(globalMetrics.count("db3"), 0);
 }
+
+TEST_F(ResourceConsumptionMetricsTest, ScopedMetricsCollector) {
+    auto& globalResourceConsumption = ResourceConsumption::get(getServiceContext());
+    auto& operationMetrics = ResourceConsumption::MetricsCollector::get(_opCtx.get());
+
+    // Collect, but don't set a dbName
+    {
+        const bool collectMetrics = true;
+        ResourceConsumption::ScopedMetricsCollector scope(_opCtx.get(), collectMetrics);
+        ASSERT_TRUE(operationMetrics.isCollecting());
+    }
+
+    ASSERT_FALSE(operationMetrics.isCollecting());
+
+    auto metricsCopy = globalResourceConsumption.getMetrics();
+    ASSERT_EQ(metricsCopy.size(), 0);
+
+    // Don't collect
+    {
+        const bool collectMetrics = false;
+        ResourceConsumption::ScopedMetricsCollector scope(_opCtx.get(), collectMetrics);
+        operationMetrics.setDbName("db1");
+        ASSERT_FALSE(operationMetrics.isCollecting());
+    }
+
+    ASSERT_FALSE(operationMetrics.isCollecting());
+
+    metricsCopy = globalResourceConsumption.getMetrics();
+    ASSERT_EQ(metricsCopy.count("db1"), 0);
+
+    // Collect
+    {
+        ResourceConsumption::ScopedMetricsCollector scope(_opCtx.get());
+        operationMetrics.setDbName("db1");
+    }
+
+    metricsCopy = globalResourceConsumption.getMetrics();
+    ASSERT_EQ(metricsCopy.count("db1"), 1);
+
+    // Collect on a different database
+    {
+        ResourceConsumption::ScopedMetricsCollector scope(_opCtx.get());
+        operationMetrics.setDbName("db2");
+    }
+
+    metricsCopy = globalResourceConsumption.getMetrics();
+    ASSERT_EQ(metricsCopy.count("db1"), 1);
+    ASSERT_EQ(metricsCopy.count("db2"), 1);
+}
+
+TEST_F(ResourceConsumptionMetricsTest, NestedScopedMetricsCollector) {
+    auto& globalResourceConsumption = ResourceConsumption::get(getServiceContext());
+    auto& operationMetrics = ResourceConsumption::MetricsCollector::get(_opCtx.get());
+
+    // Collect, nesting does not override that behavior.
+    {
+        ResourceConsumption::ScopedMetricsCollector scope(_opCtx.get());
+        operationMetrics.setDbName("db1");
+
+        {
+            const bool collectMetrics = false;
+            ResourceConsumption::ScopedMetricsCollector scope(_opCtx.get(), collectMetrics);
+            ASSERT_TRUE(operationMetrics.isCollecting());
+
+            {
+                ResourceConsumption::ScopedMetricsCollector scope(_opCtx.get());
+                ASSERT_TRUE(operationMetrics.isCollecting());
+            }
+        }
+    }
+
+    auto metricsCopy = globalResourceConsumption.getMetrics();
+    ASSERT_EQ(metricsCopy.count("db1"), 1);
+
+    // Don't collect, nesting does not override that behavior.
+    {
+        const bool collectMetrics = false;
+        ResourceConsumption::ScopedMetricsCollector scope(_opCtx.get(), collectMetrics);
+        operationMetrics.setDbName("db2");
+
+        {
+            ResourceConsumption::ScopedMetricsCollector scope(_opCtx.get());
+            ASSERT_FALSE(operationMetrics.isCollecting());
+
+            {
+                ResourceConsumption::ScopedMetricsCollector scope(_opCtx.get(), collectMetrics);
+                ASSERT_FALSE(operationMetrics.isCollecting());
+            }
+        }
+    }
+
+    metricsCopy = globalResourceConsumption.getMetrics();
+    ASSERT_EQ(metricsCopy.count("db2"), 0);
+}
 }  // namespace mongo