SERVER-31098 Wrong ns in system.profile for aggregation query

(cherry picked from commit 1862b00862a6ea9c3fdd08bba52ea6a016eccbc5)
(cherry picked from commit 93f8e8b2df10c2ec28090d9bbfd533b45f3b58e2)

9 files changed, 142 insertions, 39 deletions

diff --git a/buildscripts/resmokeconfig/suites/aggregation_sharded_collections_passthrough.yml b/buildscripts/resmokeconfig/suites/aggregation_sharded_collections_passthrough.yml
index ec75451ddd7..85b7a90db67 100644
--- a/buildscripts/resmokeconfig/suites/aggregation_sharded_collections_passthrough.yml
+++ b/buildscripts/resmokeconfig/suites/aggregation_sharded_collections_passthrough.yml
@@ -31,6 +31,9 @@ selector:
   - jstests/aggregation/bugs/lookup_unwind_killcursor.js
   # TODO: Remove after SERVER-23229 is fixed.
   - jstests/aggregation/bugs/groupMissing.js
+
+  # Uses the profiler, which is not supported through mongos.
+  - jstests/aggregation/sources/lookup/profile_lookup.js
   exclude_with_any_tags:
   # Tests tagged with the following will fail because they assume collections are not sharded.
   - assumes_no_implicit_collection_creation_after_drop
diff --git a/jstests/aggregation/sources/lookup/profile_lookup.js b/jstests/aggregation/sources/lookup/profile_lookup.js
new file mode 100644
index 00000000000..859b63a64a0
--- /dev/null
+++ b/jstests/aggregation/sources/lookup/profile_lookup.js
@@ -0,0 +1,40 @@
+// @tags: [does_not_support_stepdowns, requires_profiling]
+//
+// Tests that profiled $lookups contain the correct namespace and that Top is updated accordingly.
+
+(function() {
+    "use strict";
+
+    const localColl = db.local;
+    const foreignColl = db.foreign;
+    localColl.drop();
+    foreignColl.drop();
+
+    assert.writeOK(localColl.insert([{a: 1}, {b: 1}, {a: 2}]));
+    assert.writeOK(foreignColl.insert({a: 1}));
+
+    db.system.profile.drop();
+    db.setProfilingLevel(2);
+
+    const oldTop = db.adminCommand("top");
+
+    localColl.aggregate(
+        [{$lookup: {from: foreignColl.getName(), as: "res", localField: "a", foreignField: "a"}}]);
+
+    db.setProfilingLevel(0);
+
+    // Confirm that namespace is the local rather than foreign collection.
+    const profileDoc = db.system.profile.findOne();
+    assert.eq("test.local", profileDoc.ns);
+
+    // Confirm that the local collection had one command added to Top.
+    const newTop = db.adminCommand("top");
+    assert.eq(1,
+              newTop.totals[localColl.getFullName()].commands.count -
+                  oldTop.totals[localColl.getFullName()].commands.count);
+
+    // Confirm that for each document in local, the foreign collection had one entry added to Top.
+    assert.eq(3,
+              newTop.totals[foreignColl.getFullName()].commands.count -
+                  oldTop.totals[foreignColl.getFullName()].commands.count);
+}());
diff --git a/src/mongo/db/commands/getmore_cmd.cpp b/src/mongo/db/commands/getmore_cmd.cpp
index ce817497760..3715d7cf1ef 100644
--- a/src/mongo/db/commands/getmore_cmd.cpp
+++ b/src/mongo/db/commands/getmore_cmd.cpp
@@ -199,8 +199,11 @@ public:
                     ? request.nss.getTargetNSForGloballyManagedNamespace()
                     : request.nss) {
                 const boost::optional<int> dbProfilingLevel = boost::none;
-                statsTracker.emplace(
-                    opCtx, *nssForCurOp, Top::LockType::NotLocked, dbProfilingLevel);
+                statsTracker.emplace(opCtx,
+                                     *nssForCurOp,
+                                     Top::LockType::NotLocked,
+                                     AutoStatsTracker::LogMode::kUpdateTopAndCurop,
+                                     dbProfilingLevel);
             }
         } else {
             readLock.emplace(opCtx, request.nss);
@@ -208,6 +211,7 @@ public:
             statsTracker.emplace(opCtx,
                                  request.nss,
                                  Top::LockType::ReadLocked,
+                                 AutoStatsTracker::LogMode::kUpdateTopAndCurop,
                                  readLock->getDb() ? readLock->getDb()->getProfilingLevel()
                                                    : doNotChangeProfilingLevel);
 
diff --git a/src/mongo/db/commands/killcursors_cmd.cpp b/src/mongo/db/commands/killcursors_cmd.cpp
index 62e911ab72c..ceb64f2c99c 100644
--- a/src/mongo/db/commands/killcursors_cmd.cpp
+++ b/src/mongo/db/commands/killcursors_cmd.cpp
@@ -75,8 +75,11 @@ private:
                     ? nss.getTargetNSForGloballyManagedNamespace()
                     : nss) {
                 const boost::optional<int> dbProfilingLevel = boost::none;
-                statsTracker.emplace(
-                    opCtx, *nssForCurOp, Top::LockType::NotLocked, dbProfilingLevel);
+                statsTracker.emplace(opCtx,
+                                     *nssForCurOp,
+                                     Top::LockType::NotLocked,
+                                     AutoStatsTracker::LogMode::kUpdateTopAndCurop,
+                                     dbProfilingLevel);
             }
         }
 
diff --git a/src/mongo/db/commands/run_aggregate.cpp b/src/mongo/db/commands/run_aggregate.cpp
index 3aef5cf95ab..a89f5ce6c20 100644
--- a/src/mongo/db/commands/run_aggregate.cpp
+++ b/src/mongo/db/commands/run_aggregate.cpp
@@ -393,7 +393,11 @@ Status runAggregate(OperationContext* opCtx,
         // If this is a collectionless aggregation with no foreign namespaces, we don't want to
         // acquire any locks. Otherwise, lock the collection or view.
         if (nss.isCollectionlessAggregateNS() && pipelineInvolvedNamespaces.empty()) {
-            statsTracker.emplace(opCtx, nss, Top::LockType::NotLocked, 0);
+            statsTracker.emplace(opCtx,
+                                 nss,
+                                 Top::LockType::NotLocked,
+                                 AutoStatsTracker::LogMode::kUpdateTopAndCurop,
+                                 0);
         } else {
             ctx.emplace(opCtx, nss);
         }
diff --git a/src/mongo/db/db_raii.cpp b/src/mongo/db/db_raii.cpp
index 4a3e141bdbc..190c5435dc1 100644
--- a/src/mongo/db/db_raii.cpp
+++ b/src/mongo/db/db_raii.cpp
@@ -44,26 +44,29 @@ namespace mongo {
 AutoStatsTracker::AutoStatsTracker(OperationContext* opCtx,
                                    const NamespaceString& nss,
                                    Top::LockType lockType,
+                                   LogMode logMode,
                                    boost::optional<int> dbProfilingLevel)
-    : _opCtx(opCtx), _lockType(lockType) {
-    if (!dbProfilingLevel) {
+    : _opCtx(opCtx), _lockType(lockType), _nss(nss) {
+    if (!dbProfilingLevel && logMode == LogMode::kUpdateTopAndCurop) {
         // No profiling level was determined, attempt to read the profiling level from the Database
         // object.
-        AutoGetDb autoDb(_opCtx, nss.db(), MODE_IS);
+        AutoGetDb autoDb(_opCtx, _nss.db(), MODE_IS);
         if (autoDb.getDb()) {
             dbProfilingLevel = autoDb.getDb()->getProfilingLevel();
         }
     }
 
     stdx::lock_guard<Client> clientLock(*_opCtx->getClient());
-    CurOp::get(_opCtx)->enter_inlock(nss.ns().c_str(), dbProfilingLevel);
+    if (logMode == LogMode::kUpdateTopAndCurop) {
+        CurOp::get(_opCtx)->enter_inlock(_nss.ns().c_str(), dbProfilingLevel);
+    }
 }
 
 AutoStatsTracker::~AutoStatsTracker() {
     auto curOp = CurOp::get(_opCtx);
     Top::get(_opCtx->getServiceContext())
         .record(_opCtx,
-                curOp->getNS(),
+                _nss.ns(),
                 curOp->getLogicalOp(),
                 _lockType,
                 durationCount<Microseconds>(curOp->elapsedTimeExcludingPauses()),
@@ -108,6 +111,7 @@ AutoGetCollectionForRead::AutoGetCollectionForRead(OperationContext* opCtx,
     // Note: this can yield.
     _ensureMajorityCommittedSnapshotIsValid(nss, opCtx);
 }
+
 void AutoGetCollectionForRead::_ensureMajorityCommittedSnapshotIsValid(const NamespaceString& nss,
                                                                        OperationContext* opCtx) {
     while (true) {
@@ -148,12 +152,14 @@ AutoGetCollectionForReadCommand::AutoGetCollectionForReadCommand(
     OperationContext* opCtx,
     const NamespaceString& nss,
     AutoGetCollection::ViewMode viewMode,
-    Lock::DBLock lock) {
+    Lock::DBLock lock,
+    AutoStatsTracker::LogMode logMode) {
     _autoCollForRead.emplace(opCtx, nss, viewMode, std::move(lock));
     const int doNotChangeProfilingLevel = 0;
     _statsTracker.emplace(opCtx,
                           nss,
                           Top::LockType::ReadLocked,
+                          logMode,
                           _autoCollForRead->getDb() ? _autoCollForRead->getDb()->getProfilingLevel()
                                                     : doNotChangeProfilingLevel);
 
@@ -164,33 +170,45 @@ AutoGetCollectionForReadCommand::AutoGetCollectionForReadCommand(
 }
 
 AutoGetCollectionForReadCommand::AutoGetCollectionForReadCommand(
-    OperationContext* opCtx, const NamespaceString& nss, AutoGetCollection::ViewMode viewMode)
+    OperationContext* opCtx,
+    const NamespaceString& nss,
+    AutoGetCollection::ViewMode viewMode,
+    AutoStatsTracker::LogMode logMode)
     : AutoGetCollectionForReadCommand(
-          opCtx, nss, viewMode, Lock::DBLock(opCtx, nss.db(), MODE_IS)) {}
+          opCtx, nss, viewMode, Lock::DBLock(opCtx, nss.db(), MODE_IS), logMode) {}
 
 AutoGetCollectionOrViewForReadCommand::AutoGetCollectionOrViewForReadCommand(
     OperationContext* opCtx, const NamespaceString& nss)
-    : AutoGetCollectionForReadCommand(opCtx, nss, AutoGetCollection::ViewMode::kViewsPermitted),
+    : AutoGetCollectionForReadCommand(opCtx,
+                                      nss,
+                                      AutoGetCollection::ViewMode::kViewsPermitted,
+                                      AutoStatsTracker::LogMode::kUpdateTopAndCurop),
       _view(_autoCollForRead->getDb() && !getCollection()
                 ? _autoCollForRead->getDb()->getViewCatalog()->lookup(opCtx, nss.ns())
                 : nullptr) {}
 
 AutoGetCollectionOrViewForReadCommand::AutoGetCollectionOrViewForReadCommand(
     OperationContext* opCtx, const NamespaceString& nss, Lock::DBLock lock)
-    : AutoGetCollectionForReadCommand(
-          opCtx, nss, AutoGetCollection::ViewMode::kViewsPermitted, std::move(lock)),
+    : AutoGetCollectionForReadCommand(opCtx,
+                                      nss,
+                                      AutoGetCollection::ViewMode::kViewsPermitted,
+                                      std::move(lock),
+                                      AutoStatsTracker::LogMode::kUpdateTopAndCurop),
       _view(_autoCollForRead->getDb() && !getCollection()
                 ? _autoCollForRead->getDb()->getViewCatalog()->lookup(opCtx, nss.ns())
                 : nullptr) {}
 
-AutoGetCollectionForReadCommand::AutoGetCollectionForReadCommand(OperationContext* opCtx,
-                                                                 const StringData dbName,
-                                                                 const UUID& uuid) {
+AutoGetCollectionForReadCommand::AutoGetCollectionForReadCommand(
+    OperationContext* opCtx,
+    const StringData dbName,
+    const UUID& uuid,
+    AutoStatsTracker::LogMode logMode) {
     _autoCollForRead.emplace(opCtx, dbName, uuid);
     if (_autoCollForRead->getCollection()) {
         _statsTracker.emplace(opCtx,
                               _autoCollForRead->getCollection()->ns(),
                               Top::LockType::ReadLocked,
+                              logMode,
                               _autoCollForRead->getDb()->getProfilingLevel());
 
         // We have both the DB and collection locked, which is the prerequisite to do a stable shard
diff --git a/src/mongo/db/db_raii.h b/src/mongo/db/db_raii.h
index cbb7704a082..27549af6a40 100644
--- a/src/mongo/db/db_raii.h
+++ b/src/mongo/db/db_raii.h
@@ -39,22 +39,35 @@
 namespace mongo {
 
 /**
- * RAII-style class which automatically tracks the operation namespace in CurrentOp and records the
- * operation via Top upon destruction.
+ * RAII-style class which can update the diagnostic state on the operation's CurOp object and record
+ * the operation via Top upon destruction. Can be configured to only update the Top counters if
+ * desired.
  */
 class AutoStatsTracker {
     MONGO_DISALLOW_COPYING(AutoStatsTracker);
 
 public:
     /**
-     * Sets the namespace of the CurOp object associated with 'opCtx' to be 'nss' and starts the
-     * CurOp timer. 'lockType' describes which type of lock is held by this operation, and will be
-     * used for reporting via Top. If 'dbProfilingLevel' is not given, this constructor will acquire
-     * and then drop a database lock in order to determine the database's profiling level.
+     * Describes which diagnostics to update during the lifetime of this object.
+     */
+    enum class LogMode {
+        kUpdateTop,  // Increments the Top counter for this operation type and this namespace upon
+                     // destruction.
+        kUpdateTopAndCurop,  // In addition to incrementing the Top counter, adjusts state on the
+                             // CurOp object associated with the OperationContext. Updates the
+                             // namespace to be 'nss', starts a timer for the operation (if it
+                             // hasn't started already), and figures out and records the profiling
+                             // level of the operation.
+    };
+
+    /**
+     * If 'logMode' is 'kUpdateTopAndCurop', sets up and records state on the CurOp object attached
+     * to 'opCtx', as described above.
      */
     AutoStatsTracker(OperationContext* opCtx,
                      const NamespaceString& nss,
                      Top::LockType lockType,
+                     LogMode logMode,
                      boost::optional<int> dbProfilingLevel);
 
     /**
@@ -65,6 +78,7 @@ public:
 private:
     OperationContext* _opCtx;
     Top::LockType _lockType;
+    const NamespaceString _nss;
 };
 
 /**
@@ -142,19 +156,26 @@ class AutoGetCollectionForReadCommand {
     MONGO_DISALLOW_COPYING(AutoGetCollectionForReadCommand);
 
 public:
-    AutoGetCollectionForReadCommand(OperationContext* opCtx, const NamespaceString& nss)
+    AutoGetCollectionForReadCommand(
+        OperationContext* opCtx,
+        const NamespaceString& nss,
+        AutoStatsTracker::LogMode logMode = AutoStatsTracker::LogMode::kUpdateTopAndCurop)
         : AutoGetCollectionForReadCommand(
-              opCtx, nss, AutoGetCollection::ViewMode::kViewsForbidden) {}
+              opCtx, nss, AutoGetCollection::ViewMode::kViewsForbidden, logMode) {}
 
-    AutoGetCollectionForReadCommand(OperationContext* opCtx,
-                                    const NamespaceString& nss,
-                                    Lock::DBLock lock)
+    AutoGetCollectionForReadCommand(
+        OperationContext* opCtx,
+        const NamespaceString& nss,
+        Lock::DBLock lock,
+        AutoStatsTracker::LogMode logMode = AutoStatsTracker::LogMode::kUpdateTopAndCurop)
         : AutoGetCollectionForReadCommand(
-              opCtx, nss, AutoGetCollection::ViewMode::kViewsForbidden, std::move(lock)) {}
+              opCtx, nss, AutoGetCollection::ViewMode::kViewsForbidden, std::move(lock), logMode) {}
 
-    AutoGetCollectionForReadCommand(OperationContext* opCtx,
-                                    const StringData dbName,
-                                    const UUID& uuid);
+    AutoGetCollectionForReadCommand(
+        OperationContext* opCtx,
+        const StringData dbName,
+        const UUID& uuid,
+        AutoStatsTracker::LogMode logMode = AutoStatsTracker::LogMode::kUpdateTopAndCurop);
 
     Database* getDb() const {
         return _autoCollForRead->getDb();
@@ -167,12 +188,14 @@ public:
 protected:
     AutoGetCollectionForReadCommand(OperationContext* opCtx,
                                     const NamespaceString& nss,
-                                    AutoGetCollection::ViewMode viewMode);
+                                    AutoGetCollection::ViewMode viewMode,
+                                    AutoStatsTracker::LogMode logMode);
 
     AutoGetCollectionForReadCommand(OperationContext* opCtx,
                                     const NamespaceString& nss,
                                     AutoGetCollection::ViewMode viewMode,
-                                    Lock::DBLock lock);
+                                    Lock::DBLock lock,
+                                    AutoStatsTracker::LogMode logMode);
 
     // '_autoCollForRead' may need to be reset by AutoGetCollectionOrViewForReadCommand, so needs to
     // be a boost::optional.
diff --git a/src/mongo/db/pipeline/pipeline_d.cpp b/src/mongo/db/pipeline/pipeline_d.cpp
index e2a3526984a..78f86e09233 100644
--- a/src/mongo/db/pipeline/pipeline_d.cpp
+++ b/src/mongo/db/pipeline/pipeline_d.cpp
@@ -236,14 +236,17 @@ public:
 
         boost::optional<AutoGetCollectionForReadCommand> autoColl;
         if (expCtx->uuid) {
-            autoColl.emplace(expCtx->opCtx, expCtx->ns.db(), *expCtx->uuid);
+            autoColl.emplace(expCtx->opCtx,
+                             expCtx->ns.db(),
+                             *expCtx->uuid,
+                             AutoStatsTracker::LogMode::kUpdateTop);
             if (autoColl->getCollection() == nullptr) {
                 // The UUID doesn't exist anymore.
                 return {ErrorCodes::NamespaceNotFound,
                         "No namespace with UUID " + expCtx->uuid->toString()};
             }
         } else {
-            autoColl.emplace(expCtx->opCtx, expCtx->ns);
+            autoColl.emplace(expCtx->opCtx, expCtx->ns, AutoStatsTracker::LogMode::kUpdateTop);
         }
 
         // makePipeline() is only called to perform secondary aggregation requests and expects the
diff --git a/src/mongo/db/query/find.cpp b/src/mongo/db/query/find.cpp
index c2740453e6f..d59bcf4dee6 100644
--- a/src/mongo/db/query/find.cpp
+++ b/src/mongo/db/query/find.cpp
@@ -277,7 +277,11 @@ Message getMore(OperationContext* opCtx,
             const auto profilingLevel = autoDb.getDb()
                 ? boost::optional<int>{autoDb.getDb()->getProfilingLevel()}
                 : boost::none;
-            statsTracker.emplace(opCtx, *nssForCurOp, Top::LockType::NotLocked, profilingLevel);
+            statsTracker.emplace(opCtx,
+                                 *nssForCurOp,
+                                 Top::LockType::NotLocked,
+                                 AutoStatsTracker::LogMode::kUpdateTopAndCurop,
+                                 profilingLevel);
             auto view = autoDb.getDb()
                 ? autoDb.getDb()->getViewCatalog()->lookup(opCtx, nssForCurOp->ns())
                 : nullptr;
@@ -295,6 +299,7 @@ Message getMore(OperationContext* opCtx,
         statsTracker.emplace(opCtx,
                              nss,
                              Top::LockType::ReadLocked,
+                             AutoStatsTracker::LogMode::kUpdateTopAndCurop,
                              readLock->getDb() ? readLock->getDb()->getProfilingLevel()
                                                : doNotChangeProfilingLevel);
         Collection* collection = readLock->getCollection();