SERVER-47603 Rewrite db.currentOp() shell helper in terms of $currentOp aggregation stage

13 files changed, 243 insertions, 16 deletions

diff --git a/jstests/auth/auth3.js b/jstests/auth/auth3.js
index 29f61f1c7ee..490c73c93a3 100644
--- a/jstests/auth/auth3.js
+++ b/jstests/auth/auth3.js
@@ -2,16 +2,16 @@
 
 'use strict';
 
-var conn = MongoRunner.runMongod({auth: ""});
+const conn = MongoRunner.runMongod({auth: ""});
 
-var admin = conn.getDB("admin");
-var errorCodeUnauthorized = 13;
+const admin = conn.getDB("admin");
+const errorCodeUnauthorized = 13;
 
 admin.createUser({user: "foo", pwd: "bar", roles: jsTest.adminUserRoles});
 
 print("make sure curop, killop, and unlock fail");
 
-var x = admin.currentOp();
+let x = admin.currentOp();
 assert(!("inprog" in x), tojson(x));
 assert.eq(x.code, errorCodeUnauthorized, tojson(x));
 
diff --git a/jstests/core/awaitdata_getmore_cmd.js b/jstests/core/awaitdata_getmore_cmd.js
index fae96fbdc91..3449a51b3c2 100644
--- a/jstests/core/awaitdata_getmore_cmd.js
+++ b/jstests/core/awaitdata_getmore_cmd.js
@@ -9,6 +9,8 @@
 //   requires_getmore,
 //   uses_multiple_connections,
 //   uses_parallel_shell,
+//   assumes_read_concern_unchanged,
+//   requires_fcv_50
 // ]
 
 (function() {
diff --git a/jstests/core/basic1.js b/jstests/core/basic1.js
index cc2917fb5a3..66c034dd6b0 100644
--- a/jstests/core/basic1.js
+++ b/jstests/core/basic1.js
@@ -1,4 +1,5 @@
-// @tags: [does_not_support_stepdowns]
+// @tags: [does_not_support_stepdowns, assumes_read_concern_unchanged,
+// assumes_read_preference_unchanged ]
 
 t = db.getCollection("basic1");
 t.drop();
diff --git a/jstests/core/currentop.js b/jstests/core/currentop.js
index 41c2e36110f..e1c577b9469 100644
--- a/jstests/core/currentop.js
+++ b/jstests/core/currentop.js
@@ -2,10 +2,13 @@
  * Tests that long-running operations show up in currentOp and report the locks they are holding.
  *
  * @tags: [
+ *   assumes_read_preference_unchanged,
  *   assumes_superuser_permissions,
  *   # fsync command is not available on embedded
  *   incompatible_with_embedded,
  *   uses_parallel_shell,
+ *   # The aggregation stage $currentOp cannot run with a readConcern other than 'local'
+ *   assumes_read_concern_unchanged
  * ]
  */
 
diff --git a/jstests/core/currentop_shell.js b/jstests/core/currentop_shell.js
new file mode 100644
index 00000000000..62d36f38c6d
--- /dev/null
+++ b/jstests/core/currentop_shell.js
@@ -0,0 +1,137 @@
+/**
+ * Tests that the shell helper db.currentOpCursor isn't constrained by the legacy currentOp server
+ * command - ie. the result set isn't limited to 16MB and long operations aren't truncated.
+ *
+ * @tags: [
+ *   uses_parallel_shell,
+ *   # This test uses currentOp to check whether an aggregate command is running. In replica set
+ *   # environments, because currentOp is run against the admin database it is routed to the
+ *   # primary, while the aggregate may be routed to a secondary. If currentOp is running on one
+ *   # node and the expected command is run on another, the latter will not show up in the
+ *   # currentOp results.
+ *   assumes_read_preference_unchanged,
+ *   # The aggregation stage $currentOp cannot run with a readConcern other than 'local'
+ *   assumes_read_concern_unchanged,
+ *   requires_fcv_50
+ * ]
+ */
+
+(function() {
+"use strict";
+
+load("jstests/libs/fixture_helpers.js");  // for FixtureHelpers
+
+const coll = db.currentOp_cursor;
+coll.drop();
+
+for (let i = 0; i < 3; i++) {
+    assert.commandWorked(coll.insert({val: 1}));
+}
+
+// Test that db.currentOpCursor() returns an iterable cursor.
+let res = db.currentOpCursor();
+assert(res.hasNext());
+assert(res.next());
+
+// Test that db.currentOp() interface does not change.
+res = db.currentOp();
+assert("inprog" in res, "Result contains 'inprog' field");
+assert("ok" in res, "Result contains 'ok' field");
+
+// Attempting to access the fsyncLock field from the results throws with an error message.
+let error = assert.throws(() => res.fsyncLock);
+assert(
+    /fsyncLock is no longer included in the currentOp shell helper, run db\.runCommand\({currentOp: 1}\) instead/
+        .test(error));
+
+function shellOp() {
+    function createLargeDoc() {
+        let doc = {};
+        for (let i = 0; i < 100; i++) {
+            doc[i] = "Testing testing 1 2 3...";
+        }
+        return doc;
+    }
+
+    assert.commandFailedWithCode(db.runCommand({
+        aggregate: "currentOp_cursor",
+        pipeline: [{
+            $addFields: {
+                newVal: {$function: {args: [], body: "sleep(300000)", lang: "js"}},
+                bigDoc: createLargeDoc()
+            }
+        }],
+        cursor: {}
+    }),
+                                 ErrorCodes.Interrupted);
+}
+
+function startShellWithOp() {
+    const awaitShell = startParallelShell(shellOp);
+
+    // Confirm that the operation has started in the parallel shell.
+    assert.soon(
+        function() {
+            let aggRes = db.getSiblingDB("admin")
+                             .aggregate([
+                                 {$currentOp: {allUsers: true, localOps: true}},
+                                 {$match: {ns: "test.currentOp_cursor"}}
+                             ])
+                             .toArray();
+            return aggRes.length === 1;
+        },
+        function() {
+            return "Failed to find parallel shell operation in $currentOp output: " +
+                tojson(db.currentOp());
+        });
+    return awaitShell;
+}
+
+// Test that the currentOp server command truncates long operations with a warning logged.
+const serverCommandTest = startShellWithOp();
+res = db.adminCommand({currentOp: true, "ns": "test.currentOp_cursor", $all: true});
+
+if (FixtureHelpers.isMongos(db) && FixtureHelpers.isSharded(coll)) {
+    // Assert currentOp truncation behavior for each shard in the cluster.
+    assert(res.inprog.length >= 1);
+    res.inprog.forEach((result) => {
+        assert.eq(result.op, "getmore");
+        assert(result.cursor.originatingCommand.hasOwnProperty("$truncated"));
+    });
+} else {
+    // Assert currentOp truncation behavior for unsharded collections.
+    assert.eq(res.inprog.length, 1);
+    assert.eq(res.inprog[0].op, "command");
+    assert(res.inprog[0].command.hasOwnProperty("$truncated"));
+}
+
+const log = FixtureHelpers.getPrimaryForNodeHostingDatabase(db).adminCommand({getLog: "global"});
+assert(/will be truncated/.test(log.log));
+
+res.inprog.forEach((op) => {
+    assert.commandWorked(db.killOp(op.opid));
+});
+
+serverCommandTest();
+
+// Test that the db.currentOp() shell helper does not truncate ops.
+const shellHelperTest = startShellWithOp();
+res = db.currentOp({"ns": "test.currentOp_cursor"});
+
+if (FixtureHelpers.isMongos(db) && FixtureHelpers.isSharded(coll)) {
+    assert(res.inprog.length >= 1);
+    res.inprog.forEach((result) => {
+        assert.eq(result.op, "getmore");
+        assert(!result.cursor.originatingCommand.hasOwnProperty("$truncated"));
+    });
+} else {
+    assert.eq(res.inprog.length, 1);
+    assert(!res.inprog[0].command.hasOwnProperty("$truncated"));
+}
+
+res.inprog.forEach((op) => {
+    assert.commandWorked(db.killOp(op.opid));
+});
+
+shellHelperTest();
+})();
diff --git a/jstests/core/fsync.js b/jstests/core/fsync.js
index 77b247ff1f1..c8508e1811e 100644
--- a/jstests/core/fsync.js
+++ b/jstests/core/fsync.js
@@ -49,7 +49,8 @@ assert(!resFail.ok, "fsyncLock command succeeded against DB other than admin.");
 // Uses admin automatically and locks the server for writes.
 var fsyncLockRes = db.fsyncLock();
 assert(fsyncLockRes.ok, "fsyncLock command failed against admin DB");
-assert(db.currentOp().fsyncLock, "Value in db.currentOp incorrect for fsyncLocked server");
+assert(db.getSiblingDB('admin').runCommand({currentOp: 1}).fsyncLock,
+       "Value in currentOp result incorrect for fsyncLocked server");
 
 // Make sure writes are blocked. Spawn a write operation in a separate shell and make sure it
 // is blocked. There is really no way to do that currently, so just check that the write didn't
@@ -64,7 +65,8 @@ assert.eq(1, fsyncLockDB.coll.find({}).itcount());
 // Unlock and make sure the insert succeeded.
 var fsyncUnlockRes = db.fsyncUnlock();
 assert(fsyncUnlockRes.ok, "fsyncUnlock command failed");
-assert(db.currentOp().fsyncLock == null, "fsyncUnlock is not null in db.currentOp");
+assert(db.getSiblingDB('admin').runCommand({currentOp: 1}).fsyncLock == null,
+       "fsyncUnlock is not null in currentOp result");
 
 // Make sure the db is unlocked and the initial write made it through.
 writeOpHandle();
@@ -85,9 +87,9 @@ assert(fsyncUnlockRes.ok, "Second execution of fsyncUnlock command failed");
 fsyncLockRes = db.fsyncLock();
 assert.commandWorked(fsyncLockRes);
 assert(fsyncLockRes.lockCount == 1, tojson(fsyncLockRes));
-let currentOp = db.currentOp();
+let currentOp = db.getSiblingDB('admin').runCommand({currentOp: 1});
 assert.commandWorked(currentOp);
-assert(currentOp.fsyncLock, "Value in db.currentOp incorrect for fsyncLocked server");
+assert(currentOp.fsyncLock, "Value in currentOp result incorrect for fsyncLocked server");
 
 let shellHandle1 =
     startParallelShell("db.getSiblingDB('fsyncLockTestDB').multipleLock.insert({x:1});");
@@ -95,9 +97,9 @@ let shellHandle1 =
 fsyncLockRes = db.fsyncLock();
 assert.commandWorked(fsyncLockRes);
 assert(fsyncLockRes.lockCount == 2, tojson(fsyncLockRes));
-currentOp = db.currentOp();
+currentOp = db.getSiblingDB('admin').runCommand({currentOp: 1});
 assert.commandWorked(currentOp);
-assert(currentOp.fsyncLock, "Value in db.currentOp incorrect for fsyncLocked server");
+assert(currentOp.fsyncLock, "Value in currentOp result incorrect for fsyncLocked server");
 
 let shellHandle2 =
     startParallelShell("db.getSiblingDB('fsyncLockTestDB').multipleLock.insert({x:1});");
diff --git a/jstests/core/killop_drop_collection.js b/jstests/core/killop_drop_collection.js
index f589aff89e8..b09ee1d246c 100644
--- a/jstests/core/killop_drop_collection.js
+++ b/jstests/core/killop_drop_collection.js
@@ -9,6 +9,8 @@
  *   # Uses index building in background
  *   requires_background_index,
  *   uses_parallel_shell,
+ *   assumes_read_concern_unchanged,
+ *   assumes_read_preference_unchanged
  * ]
  */
 (function() {
diff --git a/jstests/core/mr_killop.js b/jstests/core/mr_killop.js
index 925da1118b8..daea8e8aa37 100644
--- a/jstests/core/mr_killop.js
+++ b/jstests/core/mr_killop.js
@@ -6,6 +6,9 @@
 //   uses_map_reduce_with_temp_collections,
 //   uses_multiple_connections,
 //   uses_parallel_shell,
+//   assumes_read_preference_unchanged,
+//   assumes_read_concern_unchanged,
+//   requires_fcv_50
 // ]
 (function() {
 "use strict";
@@ -32,7 +35,8 @@ function getOpCode() {
                 stringifiedCmd.search(source.getName()) >= 0;
         }
 
-        return cmdBody.mapreduce && cmdBody.mapreduce == source.getName();
+        return (cmdBody.mapreduce && cmdBody.mapreduce == source.getName()) ||
+            (cmdBody.isMapReduceCommand && cmdBody.aggregate == source.getName());
     }
 
     for (let i in inProg) {
diff --git a/jstests/libs/parallelTester.js b/jstests/libs/parallelTester.js
index 511ba98b1cd..69ce00b4670 100644
--- a/jstests/libs/parallelTester.js
+++ b/jstests/libs/parallelTester.js
@@ -195,6 +195,9 @@ if (typeof _threadInject != "undefined") {
             // Assumes that other tests are not creating cursors.
             "kill_cursors.js",
 
+            // Assumes that other tests are not starting operations.
+            "currentop_shell.js",
+
             // These tests check global command counters.
             "find_and_modify_metrics.js",
             "update_metrics.js",
diff --git a/jstests/noPassthrough/currentop_query.js b/jstests/noPassthrough/currentop_query.js
index ebdffb9c22c..d4d7aab81c4 100644
--- a/jstests/noPassthrough/currentop_query.js
+++ b/jstests/noPassthrough/currentop_query.js
@@ -636,7 +636,8 @@ function runTests({conn, readMode, currentOp, truncatedOps, localOps}) {
 }
 
 function currentOpCommand(inputDB, filter, truncatedOps, localOps) {
-    return inputDB.currentOp(Object.assign(filter, {$truncateOps: truncatedOps}));
+    return inputDB.getSiblingDB("admin").runCommand(
+        Object.assign({currentOp: true, $truncateOps: truncatedOps}, filter));
 }
 
 function currentOpAgg(inputDB, filter, truncatedOps, localOps) {
diff --git a/jstests/replsets/maintenance_non-blocking.js b/jstests/replsets/maintenance_non-blocking.js
index 19887ae55f6..d2936c7b4a5 100644
--- a/jstests/replsets/maintenance_non-blocking.js
+++ b/jstests/replsets/maintenance_non-blocking.js
@@ -36,7 +36,6 @@ doTest = function() {
 
     print("******* writing to primary ************* ");
     assert.commandWorked(mColl.save({_id: -1}));
-    printjson(sDB.currentOp());
     assert.neq(null, mColl.findOne());
 
     var hello = assert.commandWorked(sColl.runCommand("hello"));
diff --git a/src/mongo/db/curop.cpp b/src/mongo/db/curop.cpp
index 29e63a3a1e9..32dd1a531d4 100644
--- a/src/mongo/db/curop.cpp
+++ b/src/mongo/db/curop.cpp
@@ -674,6 +674,11 @@ void appendAsObjOrString(StringData name,
             objToString[*maxSize - 3] = '.';
             objToString[*maxSize - 2] = '.';
             objToString[*maxSize - 1] = '.';
+            LOGV2_INFO(4760300,
+                       "Gathering currentOp information, operation of size {size} exceeds the size "
+                       "limit of {limit} and will be truncated.",
+                       "size"_attr = objToString.size(),
+                       "limit"_attr = *maxSize);
         }
 
         StringData truncation = StringData(objToString).substr(0, *maxSize);
diff --git a/src/mongo/shell/db.js b/src/mongo/shell/db.js
index bcdcea166a0..53a18150595 100644
--- a/src/mongo/shell/db.js
+++ b/src/mongo/shell/db.js
@@ -814,7 +814,31 @@ var commandUnsupported = function(res) {
 };
 
 DB.prototype.currentOp = function(arg) {
-    var q = {};
+    // TODO CLOUDP-89361: The shell is connected to the Atlas Proxy, which currently does not
+    // support the $currentOp aggregation stage. Remove the legacy server command path once the
+    // proxy can support $currentOp.
+    if (this.serverStatus().hasOwnProperty("atlasVersion")) {
+        return this.currentOpLegacy(arg);
+    }
+
+    try {
+        const results = this.currentOpCursor(arg).toArray();
+        let res = {"inprog": results.length > 0 ? results : [], "ok": 1};
+        Object.defineProperty(res, "fsyncLock", {
+            get: function() {
+                throw Error(
+                    "fsyncLock is no longer included in the currentOp shell helper, run db.runCommand({currentOp: 1}) instead.");
+            }
+        });
+        return res;
+    } catch (e) {
+        return {"ok": 0, "code": e.code, "errmsg": "Error executing $currentOp: " + e.message};
+    }
+};
+DB.prototype.currentOP = DB.prototype.currentOp;
+
+DB.prototype.currentOpLegacy = function(arg) {
+    let q = {};
     if (arg) {
         if (typeof (arg) == "object")
             Object.extend(q, arg);
@@ -838,7 +862,51 @@ DB.prototype.currentOp = function(arg) {
     }
     return res;
 };
-DB.prototype.currentOP = DB.prototype.currentOp;
+
+DB.prototype.currentOpCursor = function(arg) {
+    let q = {};
+    if (arg) {
+        if (typeof (arg) == "object")
+            Object.extend(q, arg);
+        else if (arg)
+            q["$all"] = true;
+    }
+
+    // Convert the incoming currentOp command into an equivalent aggregate command
+    // of the form {aggregate:1, pipeline: [{$currentOp: {idleConnections: $all, allUsers:
+    // !$ownOps, truncateOps: false}}, {$match: {<user-defined filter>}}], cursor:{}}.
+    let pipeline = [];
+
+    let currOpArgs = {};
+    let currOpStage = {"$currentOp": currOpArgs};
+    currOpArgs["allUsers"] = !q["$ownOps"];
+    currOpArgs["idleConnections"] = !!q["$all"];
+    currOpArgs["truncateOps"] = false;
+
+    pipeline.push(currOpStage);
+
+    let matchArgs = {};
+    let matchStage = {"$match": matchArgs};
+    for (const fieldname of Object.keys(q)) {
+        if (fieldname !== "$all" && fieldname !== "$ownOps" && fieldname !== "$truncateOps") {
+            matchArgs[fieldname] = q[fieldname];
+        }
+    }
+
+    pipeline.push(matchStage);
+
+    // The legacy db.currentOp() shell helper ignored any explicitly set read preference and used
+    // the default, with the ability to also run on secondaries. To preserve this behavior we will
+    // temporarily set the session's read preference to "primaryPreferred".
+    const session = this.getSession();
+    const readPreference = session.getOptions().getReadPreference();
+    try {
+        session.getOptions().setReadPreference({mode: "primaryPreferred"});
+        return this.getSiblingDB("admin").aggregate(pipeline);
+    } finally {
+        session.getOptions().setReadPreference(readPreference);
+    }
+};
 
 DB.prototype.killOp = function(op) {
     if (!op)