New upstream version 6.0.20upstream

1 files changed, 104 insertions, 73 deletions

diff --git a/jstests/core/currentop_shell.js b/jstests/core/currentop_shell.js
index c96ef2507f1..31b084bf132 100644
--- a/jstests/core/currentop_shell.js
+++ b/jstests/core/currentop_shell.js
@@ -2,7 +2,16 @@
  * 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.
  *
+ * Note: On newer branches, this test contains additional cases for the currentOp command (without a
+ * shell helper) and the $currentOp pipeline stage, which are not included here. Those cases would
+ * behave unreliably because of SERVER-92284.
+ *
  * @tags: [
+ *   # The collection may be completely moved to another shard, which results in currentOp not
+ *   # returning the expected command.
+ *   assumes_balancer_off,
+ *   # The test runs commands that are not allowed with security token: getLog.
+ *   not_allowed_with_signed_security_token,
  *   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
@@ -11,6 +20,8 @@
  *   # currentOp results.
  *   assumes_read_preference_unchanged,
  *   no_selinux,
+ *   # Uses $function operator.
+ *   requires_scripting,
  * ]
  */
 
@@ -18,31 +29,45 @@
 "use strict";
 
 load("jstests/libs/fixture_helpers.js");  // for FixtureHelpers
+load('jstests/libs/parallel_shell_helpers.js');
 
 const coll = db.currentOp_cursor;
 coll.drop();
 
-for (let i = 0; i < 3; i++) {
+for (let i = 0; i < 100; 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);
+//
+const cursorFromCurrentOp = db.currentOpCursor();
+assert(cursorFromCurrentOp.hasNext());
+assert(cursorFromCurrentOp.next());
+
+//
+// Test that db.currentOp() returns an object in the expected format.
+//
+const currentOpRes = db.currentOp();
+assert("inprog" in currentOpRes, "Result contains 'inprog' field");
+assert("ok" in currentOpRes, "Result contains 'ok' field");
+
+//
+// Test that attempting to access the fsyncLock field from the results throws with an error message.
+//
+const error = assert.throws(() => currentOpRes.fsyncLock);
 assert(
     /fsyncLock is no longer included in the currentOp shell helper, run db\.runCommand\({currentOp: 1}\) instead/
         .test(error));
 
-function shellOp() {
+//
+// Start a pipeline with a large command object in a parallel shell and then test three different
+// methods of executing "currentOp" queries to ensure that they all observe the operation and that
+// they do or do not truncate its command object (according to each one's specification).
+//
+
+// Starts the query. Intended to b e called from a parallel shell.
+function startLongRunningAggregation(collName, comment) {
     function createLargeDoc() {
         let doc = {};
         for (let i = 0; i < 100; i++) {
@@ -52,90 +77,96 @@ function shellOp() {
     }
 
     assert.commandFailedWithCode(db.runCommand({
-        aggregate: "currentOp_cursor",
+        aggregate: collName,
         pipeline: [{
             $addFields: {
                 newVal: {$function: {args: [], body: "sleep(1000000)", lang: "js"}},
                 bigDoc: createLargeDoc()
             }
         }],
-        comment: TestData.comment,
+        comment: comment,
         cursor: {}
     }),
                                  ErrorCodes.Interrupted);
 }
 
-function startShellWithOp(comment) {
-    TestData.comment = comment;
-    const awaitShell = startParallelShell(shellOp);
-
-    // Confirm that the operation has started in the parallel shell.
+// Repeatedly executes 'getOperationsFunction()' until it returns exactly one operation for each
+// shard in a sharded collection or exactly one operation for an unsharded collection.
+function awaitOperations(getOperationsFunction) {
+    let operations;
     assert.soon(
         function() {
-            let aggRes =
-                db.getSiblingDB("admin")
-                    .aggregate([
-                        {$currentOp: {}},
-                        {$match: {ns: "test.currentOp_cursor", "command.comment": TestData.comment}}
-                    ])
-                    .toArray();
-            return aggRes.length >= 1;
+            const numShards = FixtureHelpers.numberOfShardsForCollection(coll);
+            operations = getOperationsFunction();
+
+            // No shard should have more than one operation matching the query comment. First check
+            // that the total number of operations is no greater than the total number of shards.
+            assert.lte(operations.length, numShards, operations);
+
+            // Also explicitly check that each shard appears no more than once in the list of
+            // operations.
+            const distinctShardNames = new Set(operations.map(op => "shard" in op ? op.shard : ""));
+            assert.eq(operations.length, distinctShardNames.size, {operations, numShards});
+
+            if (operations.length < numShards) {
+                print(`Found ${operations.length} operation(s); waiting until there are ${
+                    numShards} operation(s)`);
+                return false;
+            } else if (operations.some(op => op.op !== "getmore" && "cursor" in op &&
+                                           op.cursor.batchSize === 0)) {
+                print(`Found command with empty 'batchSize' value; waiting for getmore: ${
+                    tojson(operations)}`);
+                return false;
+            }
+
+            return true;
         },
         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("currentOp_server");
-res = db.adminCommand({
-    currentOp: true,
-    $and: [{"ns": "test.currentOp_cursor"}, {"command.comment": "currentOp_server"}]
-});
+    return operations;
+}
 
-if (FixtureHelpers.isMongos(db) && FixtureHelpers.isSharded(coll)) {
-    // Assert currentOp truncation behavior for each shard in the cluster.
-    assert(res.inprog.length >= 1, res);
-    res.inprog.forEach((result) => {
-        assert.eq(result.op, "getmore", result);
-        assert(result.cursor.originatingCommand.hasOwnProperty("$truncated"), result);
-    });
-} else {
-    // Assert currentOp truncation behavior for unsharded collections.
-    assert.eq(res.inprog.length, 1, res);
-    assert.eq(res.inprog[0].op, "command", res);
-    assert(res.inprog[0].command.hasOwnProperty("$truncated"), res);
+function getCommandFromCurrentOpEntry(entry) {
+    if (entry.op === "command" && "command" in entry) {
+        return entry.command;
+    } else if (entry.op === "getmore" && "cursor" in entry &&
+               "originatingCommand" in entry.cursor) {
+        return entry.cursor.originatingCommand;
+    } else {
+        assert(false, entry);
+    }
 }
 
-const log = FixtureHelpers.getPrimaryForNodeHostingDatabase(db).adminCommand({getLog: "global"});
-assert(/will be truncated/.test(log.log));
+const comment = "long_running_aggregation";
+const awaitShell =
+    startParallelShell(funWithArgs(startLongRunningAggregation, coll.getName(), comment));
 
-res.inprog.forEach((op) => {
-    assert.commandWorked(db.killOp(op.opid));
-});
+const filter = {
+    ns: coll.getFullName(),
+    "command.comment": comment,
 
-serverCommandTest();
-
-// Test that the db.currentOp() shell helper does not truncate ops.
-const shellHelperTest = startShellWithOp("currentOp_shell");
-res = db.currentOp({"ns": "test.currentOp_cursor", "command.comment": "currentOp_shell"});
-
-if (FixtureHelpers.isMongos(db) && FixtureHelpers.isSharded(coll)) {
-    assert(res.inprog.length >= 1, res);
-    res.inprog.forEach((result) => {
-        assert.eq(result.op, "getmore", result);
-        assert(!result.cursor.originatingCommand.hasOwnProperty("$truncated"), result);
-    });
-} else {
-    assert.eq(res.inprog.length, 1, res);
-    assert(!res.inprog[0].command.hasOwnProperty("$truncated"), res);
-}
+    // On the replica set endpoint, currentOp reports both router and shard operations. So filter
+    // out one of them.
+    role: TestData.testingReplicaSetEndpoint ? "ClusterRole{router}" : {$exists: false}
+};
 
-res.inprog.forEach((op) => {
-    assert.commandWorked(db.killOp(op.opid));
+// The 'currentOp' shell helper should _not_ truncate the command.
+const operationsViaCurrentOpShellHelper = awaitOperations(function() {
+    return db.currentOp(filter).inprog;
 });
-
-shellHelperTest();
+assert(operationsViaCurrentOpShellHelper.every(op => {
+    const command = getCommandFromCurrentOpEntry(op);
+    return !("$truncated" in command) && command.aggregate == coll.getName();
+}),
+       operationsViaCurrentOpShellHelper);
+
+// Finish the test by killing the long-running aggregation pipeline and joining the parallel shell
+// that launched it.
+for (let op of operationsViaCurrentOpShellHelper) {
+    assert.commandWorked(db.killOp(op.opid));
+}
+awaitShell();
 })();