SERVER-28350 cluster_pipeline_command.cpp::explain constructs command object with unwrapped readPref but doesn't use

4 files changed, 146 insertions, 11 deletions

diff --git a/buildscripts/resmokeconfig/suites/sharding_last_stable_mongos_and_mixed_shards.yml b/buildscripts/resmokeconfig/suites/sharding_last_stable_mongos_and_mixed_shards.yml
index 1b2fd9cb6b3..c725958cc00 100644
--- a/buildscripts/resmokeconfig/suites/sharding_last_stable_mongos_and_mixed_shards.yml
+++ b/buildscripts/resmokeconfig/suites/sharding_last_stable_mongos_and_mixed_shards.yml
@@ -17,6 +17,7 @@ selector:
     # Enable when 3.6 becomes last-stable.
     - jstests/sharding/views.js
     - jstests/sharding/view_rewrite.js
+    - jstests/sharding/explain_agg_read_pref.js
     # New feature in v3.6 mongos
     - jstests/sharding/logical_time_metadata.js
     # New feature in v3.6 mongos and mongod.
diff --git a/jstests/sharding/explain_agg_read_pref.js b/jstests/sharding/explain_agg_read_pref.js
new file mode 100644
index 00000000000..a90a7bce131
--- /dev/null
+++ b/jstests/sharding/explain_agg_read_pref.js
@@ -0,0 +1,130 @@
+/**
+ * Tests that readPref applies on an explain for an aggregation command.
+ */
+(function() {
+    "use strict";
+
+    load("jstests/libs/profiler.js");  // For profilerHasSingleMatchingEntryOrThrow.
+
+    const st = new ShardingTest({
+        name: "agg_explain_readPref",
+        shards: 2,
+        other: {
+            rs0: {
+                nodes: [
+                    {rsConfig: {priority: 1, tags: {"tag": "primary"}}},
+                    {rsConfig: {priority: 0, tags: {"tag": "secondary"}}}
+                ]
+            },
+            rs1: {
+                nodes: [
+                    {rsConfig: {priority: 1, tags: {"tag": "primary"}}},
+                    {rsConfig: {priority: 0, tags: {"tag": "secondary"}}}
+                ]
+            },
+            enableBalancer: false
+        }
+    });
+
+    const mongos = st.s;
+    const config = mongos.getDB("config");
+    const mongosDB = mongos.getDB("agg_explain_readPref");
+    assert.commandWorked(mongosDB.dropDatabase());
+
+    const coll = mongosDB.getCollection("coll");
+
+    assert.commandWorked(config.adminCommand({enableSharding: mongosDB.getName()}));
+    st.ensurePrimaryShard(mongosDB.getName(), "agg_explain_readPref-rs0");
+    const rs0Primary = st.rs0.getPrimary();
+    const rs0Secondary = st.rs0.getSecondary();
+    const rs1Primary = st.rs1.getPrimary();
+    const rs1Secondary = st.rs1.getSecondary();
+
+    for (let i = 0; i < 10; ++i) {
+        assert.writeOK(coll.insert({a: i}));
+    }
+
+    //
+    // Confirms that aggregations with explain run against mongos are executed against a tagged
+    // secondary or primary, as per readPreference setting.
+    //
+    function confirmReadPreference(primary, secondary) {
+        assert.commandWorked(secondary.setProfilingLevel(2));
+        assert.commandWorked(primary.setProfilingLevel(2));
+
+        // [<pref>, <tags>, <target>, <comment>]
+        [['primary', [{}], primary, "primary"],
+         ['primaryPreferred', [{tag: 'secondary'}], primary, "primaryPreferred"],
+         ['secondary', [{}], secondary, "secondary"],
+         ['secondary', [{tag: 'secondary'}], secondary, "secondaryTag"],
+         ['secondaryPreferred', [{tag: 'secondary'}], secondary, "secondaryPreferred"],
+         ['secondaryPreferred', [{tag: 'primary'}], primary, "secondaryPreferredTagPrimary"]]
+            .forEach(function(args) {
+                const pref = args[0], tagSets = args[1], target = args[2], name = args[3];
+
+                //
+                // Tests that explain within an aggregate command and an explicit $readPreference
+                // targets the correct node in the replica set given by 'target'.
+                //
+                let comment = name + "_explain_within_query";
+                assert.commandWorked(mongosDB.runCommand({
+                    query: {
+                        aggregate: "coll",
+                        pipeline: [],
+                        comment: comment,
+                        cursor: {},
+                        explain: true
+                    },
+                    $readPreference: {mode: pref, tags: tagSets}
+                }));
+
+                profilerHasSingleMatchingEntryOrThrow(target, {
+                    "ns": coll.getFullName(),
+                    "command.explain.aggregate": coll.getName(),
+                    "command.explain.comment": comment,
+                    "command.explain.$queryOptions.$readPreference.mode": pref
+                });
+
+                //
+                // Tests that an aggregation command wrapped in an explain with explicit
+                // $queryOptions targets the correct node in the replica set given by 'target'.
+                //
+                comment = name + "_explain_wrapped_agg";
+                assert.commandWorked(mongosDB.runCommand({
+                    explain: {aggregate: "coll", pipeline: [], comment: comment, cursor: {}},
+                    $queryOptions: {$readPreference: {mode: pref, tags: tagSets}}
+                }));
+
+                profilerHasSingleMatchingEntryOrThrow(target, {
+                    "ns": coll.getFullName(),
+                    "command.explain.aggregate": coll.getName(),
+                    "command.explain.comment": comment,
+                    "command.explain.$queryOptions.$readPreference.mode": pref
+                });
+            });
+
+        assert.commandWorked(secondary.setProfilingLevel(0));
+        assert.commandWorked(primary.setProfilingLevel(0));
+    }
+
+    //
+    // Test aggregate explains run against an unsharded collection.
+    //
+    confirmReadPreference(rs0Primary.getDB(mongosDB.getName()),
+                          rs0Secondary.getDB(mongosDB.getName()));
+
+    //
+    // Test aggregate explains run against a sharded collection.
+    //
+    assert.commandWorked(coll.createIndex({a: 1}));
+    assert.commandWorked(config.adminCommand({shardCollection: coll.getFullName(), key: {a: 1}}));
+    assert.commandWorked(mongos.adminCommand({split: coll.getFullName(), middle: {a: 6}}));
+    assert.commandWorked(mongosDB.adminCommand(
+        {moveChunk: coll.getFullName(), find: {a: 25}, to: "agg_explain_readPref-rs1"}));
+
+    // Sharded tests are run against the non-primary shard for the "agg_explain_readPref" db.
+    confirmReadPreference(rs1Primary.getDB(mongosDB.getName()),
+                          rs1Secondary.getDB(mongosDB.getName()));
+
+    st.stop();
+})();
diff --git a/src/mongo/s/commands/cluster_aggregate.cpp b/src/mongo/s/commands/cluster_aggregate.cpp
index b52beed565f..d17a80a1e70 100644
--- a/src/mongo/s/commands/cluster_aggregate.cpp
+++ b/src/mongo/s/commands/cluster_aggregate.cpp
@@ -66,10 +66,13 @@ namespace {
 //
 // produces the corresponding explain command:
 //
-//   {explain: {aggregate: "myCollection", pipline: [], ...}, verbosity: ...}
+//   {explain: {aggregate: "myCollection", pipline: [], ...}, $queryOptions: {...}, verbosity: ...}
 Document wrapAggAsExplain(Document aggregateCommand, ExplainOptions::Verbosity verbosity) {
     MutableDocument explainCommandBuilder;
     explainCommandBuilder["explain"] = Value(aggregateCommand);
+    // Downstream host targeting code expects queryOptions at the top level of the command object.
+    explainCommandBuilder[QueryRequest::kUnwrappedReadPrefField] =
+        Value(aggregateCommand[QueryRequest::kUnwrappedReadPrefField]);
 
     // Add explain command options.
     for (auto&& explainOption : ExplainOptions::toBSON(verbosity)) {
diff --git a/src/mongo/s/commands/cluster_pipeline_cmd.cpp b/src/mongo/s/commands/cluster_pipeline_cmd.cpp
index 460363bb1be..4f4431b52ec 100644
--- a/src/mongo/s/commands/cluster_pipeline_cmd.cpp
+++ b/src/mongo/s/commands/cluster_pipeline_cmd.cpp
@@ -78,7 +78,7 @@ public:
                      BSONObj& cmdObj,
                      std::string& errmsg,
                      BSONObjBuilder& result) {
-        const NamespaceString nss(parseNsCollectionRequired(dbname, cmdObj));
+        NamespaceString nss(parseNsCollectionRequired(dbname, cmdObj));
 
         auto request = AggregationRequest::parseFromBSON(nss, cmdObj);
         if (!request.isOK()) {
@@ -100,28 +100,29 @@ public:
                    ExplainOptions::Verbosity verbosity,
                    const rpc::ServerSelectionMetadata& serverSelectionMetadata,
                    BSONObjBuilder* out) const override {
-        const NamespaceString nss(parseNsCollectionRequired(dbName, cmdObj));
-
-        auto request = AggregationRequest::parseFromBSON(nss, cmdObj, verbosity);
-        if (!request.isOK()) {
-            return request.getStatus();
-        }
 
         // Add the server selection metadata to the aggregate command in the "unwrapped" format that
         // runAggregate() expects: {aggregate: ..., $queryOptions: {$readPreference: ...}}.
         BSONObjBuilder aggCmdBuilder;
         aggCmdBuilder.appendElements(cmdObj);
         if (auto readPref = serverSelectionMetadata.getReadPreference()) {
-            auto readPrefObj = readPref->toBSON();
             aggCmdBuilder.append(QueryRequest::kUnwrappedReadPrefField,
-                                 BSON("$readPreference" << readPrefObj));
+                                 BSON("$readPreference" << readPref->toBSON()));
+        }
+        BSONObj aggCmd = aggCmdBuilder.obj();
+
+        NamespaceString nss(parseNsCollectionRequired(dbName, aggCmd));
+
+        auto request = AggregationRequest::parseFromBSON(nss, aggCmd, verbosity);
+        if (!request.isOK()) {
+            return request.getStatus();
         }
 
         ClusterAggregate::Namespaces nsStruct;
         nsStruct.requestedNss = nss;
         nsStruct.executionNss = std::move(nss);
 
-        return ClusterAggregate::runAggregate(opCtx, nsStruct, request.getValue(), cmdObj, out);
+        return ClusterAggregate::runAggregate(opCtx, nsStruct, request.getValue(), aggCmd, out);
     }
 } clusterPipelineCmd;