SERVER-46955 Create MongosTopologyCoordinator

7 files changed, 377 insertions, 43 deletions

diff --git a/src/mongo/s/SConscript b/src/mongo/s/SConscript
index c2071ca64c5..6456e681791 100644
--- a/src/mongo/s/SConscript
+++ b/src/mongo/s/SConscript
@@ -387,6 +387,31 @@ env.Library(
 )
 
 env.Library(
+    target='mongos_is_master_response',
+    source=[
+        'mongos_is_master_response.cpp',
+    ],
+    LIBDEPS=[
+        '$BUILD_DIR/mongo/base',
+        '$BUILD_DIR/mongo/bson/util/bson_extract',
+        '$BUILD_DIR/mongo/rpc/metadata',
+    ]
+)
+
+env.Library(
+    target='mongos_topology_coordinator',
+    source=[
+        'mongos_topology_coordinator.cpp',
+    ],
+    LIBDEPS=[
+        '$BUILD_DIR/mongo/transport/transport_layer_common',
+        '$BUILD_DIR/mongo/util/fail_point',
+        '$BUILD_DIR/mongo/rpc/metadata',
+        'mongos_is_master_response',
+    ],
+)
+
+env.Library(
     target='mongos_server_parameters',
     source=[
         'mongos_server_parameters.cpp',
diff --git a/src/mongo/s/commands/SConscript b/src/mongo/s/commands/SConscript
index b48f3d881e0..0a3a215c07c 100644
--- a/src/mongo/s/commands/SConscript
+++ b/src/mongo/s/commands/SConscript
@@ -133,6 +133,7 @@ env.Library(
         '$BUILD_DIR/mongo/idl/server_parameter',
         '$BUILD_DIR/mongo/rpc/client_metadata',
         '$BUILD_DIR/mongo/s/cluster_last_error_info',
+        '$BUILD_DIR/mongo/s/mongos_topology_coordinator',
         '$BUILD_DIR/mongo/s/query/cluster_aggregate',
         '$BUILD_DIR/mongo/s/query/cluster_client_cursor',
         '$BUILD_DIR/mongo/s/sharding_api',
diff --git a/src/mongo/s/commands/cluster_is_master_cmd.cpp b/src/mongo/s/commands/cluster_is_master_cmd.cpp
index 16c0cc2b3b5..bed72f2f652 100644
--- a/src/mongo/s/commands/cluster_is_master_cmd.cpp
+++ b/src/mongo/s/commands/cluster_is_master_cmd.cpp
@@ -42,7 +42,7 @@
 #include "mongo/rpc/metadata/client_metadata.h"
 #include "mongo/rpc/metadata/client_metadata_ismaster.h"
 #include "mongo/rpc/topology_version_gen.h"
-#include "mongo/transport/ismaster_metrics.h"
+#include "mongo/s/mongos_topology_coordinator.h"
 #include "mongo/transport/message_compressor_manager.h"
 #include "mongo/util/map_util.h"
 #include "mongo/util/net/socket_utils.h"
@@ -52,16 +52,6 @@ namespace mongo {
 
 // Hangs in the beginning of each isMaster command when set.
 MONGO_FAIL_POINT_DEFINE(waitInIsMaster);
-// Awaitable isMaster requests with the proper topologyVersions are expected to sleep for
-// maxAwaitTimeMS on mongos. This failpoint will hang right before doing this sleep when set.
-MONGO_FAIL_POINT_DEFINE(hangWhileWaitingForIsMasterResponse);
-
-TopologyVersion mongosTopologyVersion;
-
-MONGO_INITIALIZER(GenerateMongosTopologyVersion)(InitializerContext*) {
-    mongosTopologyVersion = TopologyVersion(OID::gen(), 0);
-    return Status::OK();
-}
 
 namespace {
 
@@ -133,6 +123,7 @@ public:
         auto topologyVersionElement = cmdObj["topologyVersion"];
         auto maxAwaitTimeMSField = cmdObj["maxAwaitTimeMS"];
         boost::optional<TopologyVersion> clientTopologyVersion;
+        boost::optional<long long> maxAwaitTimeMS;
         if (topologyVersionElement && maxAwaitTimeMSField) {
             clientTopologyVersion = TopologyVersion::parse(IDLParserErrorContext("TopologyVersion"),
                                                            topologyVersionElement.Obj());
@@ -140,32 +131,16 @@ public:
                     "topologyVersion must have a non-negative counter",
                     clientTopologyVersion->getCounter() >= 0);
 
-            long long maxAwaitTimeMS;
-            uassertStatusOK(bsonExtractIntegerField(cmdObj, "maxAwaitTimeMS", &maxAwaitTimeMS));
-            uassert(51759, "maxAwaitTimeMS must be a non-negative integer", maxAwaitTimeMS >= 0);
+            long long parsedMaxAwaitTimeMS;
+            uassertStatusOK(
+                bsonExtractIntegerField(cmdObj, "maxAwaitTimeMS", &parsedMaxAwaitTimeMS));
 
-            LOGV2_DEBUG(23871, 3, "Using maxAwaitTimeMS for awaitable isMaster protocol.");
+            uassert(
+                51759, "maxAwaitTimeMS must be a non-negative integer", parsedMaxAwaitTimeMS >= 0);
 
-            if (clientTopologyVersion->getProcessId() == mongosTopologyVersion.getProcessId()) {
-                uassert(51761,
-                        str::stream()
-                            << "Received a topology version with counter: "
-                            << clientTopologyVersion->getCounter()
-                            << " which is greater than the mongos topology version counter: "
-                            << mongosTopologyVersion.getCounter(),
-                        clientTopologyVersion->getCounter() == mongosTopologyVersion.getCounter());
-
-                // The topologyVersion never changes on a running mongos process, so just sleep for
-                // maxAwaitTimeMS.
-                IsMasterMetrics::get(opCtx)->incrementNumAwaitingTopologyChanges();
-                ON_BLOCK_EXIT(
-                    [&] { IsMasterMetrics::get(opCtx)->decrementNumAwaitingTopologyChanges(); });
-                if (MONGO_unlikely(hangWhileWaitingForIsMasterResponse.shouldFail())) {
-                    LOGV2(31463, "hangWhileWaitingForIsMasterResponse failpoint enabled.");
-                    hangWhileWaitingForIsMasterResponse.pauseWhileSet(opCtx);
-                }
-                opCtx->sleepFor(Milliseconds(maxAwaitTimeMS));
-            }
+            maxAwaitTimeMS = parsedMaxAwaitTimeMS;
+
+            LOGV2_DEBUG(23871, 3, "Using maxAwaitTimeMS for awaitable isMaster protocol.");
         } else {
             uassert(51760,
                     (topologyVersionElement
@@ -175,8 +150,15 @@ public:
         }
 
         auto result = replyBuilder->getBodyBuilder();
-        result.appendBool("ismaster", true);
-        result.append("msg", "isdbgrid");
+        const auto* mongosTopCoord = MongosTopologyCoordinator::get(opCtx);
+
+        auto mongosIsMasterResponse =
+            mongosTopCoord->awaitIsMasterResponse(opCtx, clientTopologyVersion, maxAwaitTimeMS);
+
+        mongosIsMasterResponse->appendToBuilder(&result);
+        // The isMaster response always includes a topologyVersion.
+        auto currentMongosTopologyVersion = mongosIsMasterResponse->getTopologyVersion();
+
         result.appendNumber("maxBsonObjectSize", BSONObjMaxUserSize);
         result.appendNumber("maxMessageSizeBytes", MaxMessageSizeBytes);
         result.appendNumber("maxWriteBatchSize", write_ops::kMaxWriteBatchSize);
@@ -201,9 +183,6 @@ public:
         auto& saslMechanismRegistry = SASLServerMechanismRegistry::get(opCtx->getServiceContext());
         saslMechanismRegistry.advertiseMechanismNamesForUser(opCtx, cmdObj, &result);
 
-        BSONObjBuilder topologyVersionBuilder(result.subobjStart("topologyVersion"));
-        mongosTopologyVersion.serialize(&topologyVersionBuilder);
-
         if (opCtx->isExhaust()) {
             LOGV2_DEBUG(23872, 3, "Using exhaust for isMaster protocol");
 
@@ -215,8 +194,9 @@ public:
             InExhaustIsMaster::get(opCtx->getClient()->session().get())
                 ->setInExhaustIsMaster(true /* inExhaustIsMaster */);
 
-            if (clientTopologyVersion->getProcessId() == mongosTopologyVersion.getProcessId() &&
-                clientTopologyVersion->getCounter() == mongosTopologyVersion.getCounter()) {
+            if (clientTopologyVersion->getProcessId() ==
+                    currentMongosTopologyVersion.getProcessId() &&
+                clientTopologyVersion->getCounter() == currentMongosTopologyVersion.getCounter()) {
                 // Indicate that an exhaust message should be generated and the previous BSONObj
                 // command parameters should be reused as the next BSONObj command parameters.
                 replyBuilder->setNextInvocation(boost::none);
@@ -226,7 +206,7 @@ public:
                     if (elt.fieldNameStringData() == "topologyVersion"_sd) {
                         BSONObjBuilder topologyVersionBuilder(
                             nextInvocationBuilder.subobjStart("topologyVersion"));
-                        mongosTopologyVersion.serialize(&topologyVersionBuilder);
+                        currentMongosTopologyVersion.serialize(&topologyVersionBuilder);
                     } else {
                         nextInvocationBuilder.append(elt);
                     }
diff --git a/src/mongo/s/mongos_is_master_response.cpp b/src/mongo/s/mongos_is_master_response.cpp
new file mode 100644
index 00000000000..9f314beba62
--- /dev/null
+++ b/src/mongo/s/mongos_is_master_response.cpp
@@ -0,0 +1,48 @@
+/**
+ *    Copyright (C) 2020-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/s/mongos_is_master_response.h"
+#include "mongo/bson/bsonobjbuilder.h"
+#include "mongo/util/str.h"
+
+namespace mongo {
+
+MongosIsMasterResponse::MongosIsMasterResponse(TopologyVersion topologyVersion) {
+    _topologyVersion = topologyVersion;
+}
+
+void MongosIsMasterResponse::appendToBuilder(BSONObjBuilder* builder) const {
+    builder->append(kIsMasterFieldName, "true");
+    builder->append(kMsgFieldName, "isdbgrid");
+
+    BSONObjBuilder topologyVersionBuilder(builder->subobjStart(kTopologyVersionFieldName));
+    _topologyVersion.serialize(&topologyVersionBuilder);
+}
+
+}  // namespace mongo
diff --git a/src/mongo/s/mongos_is_master_response.h b/src/mongo/s/mongos_is_master_response.h
new file mode 100644
index 00000000000..ec70eaa5d4e
--- /dev/null
+++ b/src/mongo/s/mongos_is_master_response.h
@@ -0,0 +1,70 @@
+/**
+ *    Copyright (C) 2020-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#pragma once
+
+#include "mongo/rpc/topology_version_gen.h"
+#include <boost/optional.hpp>
+#include <string>
+
+namespace mongo {
+
+class BSONObj;
+class BSONObjBuilder;
+
+/**
+ * Response structure for the ismaster command.
+ *
+ * Only handles responses from mongos.
+ */
+class MongosIsMasterResponse {
+public:
+    static constexpr StringData kTopologyVersionFieldName = "topologyVersion"_sd;
+    static constexpr StringData kIsMasterFieldName = "ismaster"_sd;
+    static constexpr StringData kMsgFieldName = "msg"_sd;
+
+    /**
+     * Explicit constructor that sets the _topologyVersion field.
+     */
+    MongosIsMasterResponse(TopologyVersion topologyVersion);
+
+    /**
+     * Appends MongosIsMasterResponse fields to "builder".
+     */
+    void appendToBuilder(BSONObjBuilder* builder) const;
+
+    TopologyVersion getTopologyVersion() const {
+        return _topologyVersion;
+    }
+
+private:
+    TopologyVersion _topologyVersion;
+};
+
+}  // namespace mongo
diff --git a/src/mongo/s/mongos_topology_coordinator.cpp b/src/mongo/s/mongos_topology_coordinator.cpp
new file mode 100644
index 00000000000..78385fe5ea8
--- /dev/null
+++ b/src/mongo/s/mongos_topology_coordinator.cpp
@@ -0,0 +1,127 @@
+/**
+ *    Copyright (C) 2020-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kCommand
+
+#include "mongo/logv2/log.h"
+
+#include "mongo/db/client.h"
+#include "mongo/db/service_context.h"
+#include "mongo/s/mongos_topology_coordinator.h"
+#include "mongo/util/fail_point.h"
+
+namespace mongo {
+
+namespace {
+
+const auto getMongosTopologyCoordinator =
+    ServiceContext::declareDecoration<MongosTopologyCoordinator>();
+
+/**
+ * Generate an identifier unique to this instance.
+ */
+OID instanceId;
+
+MONGO_INITIALIZER(GenerateMongosInstanceId)(InitializerContext*) {
+    instanceId = OID::gen();
+    return Status::OK();
+}
+
+// Awaitable isMaster requests with the proper topologyVersions are expected to wait for
+// maxAwaitTimeMS on mongos. When set, this failpoint will hang right before waiting on a
+// topology change.
+MONGO_FAIL_POINT_DEFINE(hangWhileWaitingForIsMasterResponse);
+
+}  // namespace
+
+// Make MongosTopologyCoordinator a decoration on the ServiceContext.
+MongosTopologyCoordinator* MongosTopologyCoordinator::get(OperationContext* opCtx) {
+    return &getMongosTopologyCoordinator(opCtx->getClient()->getServiceContext());
+}
+
+MongosTopologyCoordinator::MongosTopologyCoordinator() : _topologyVersion(instanceId, 0) {}
+
+std::shared_ptr<MongosIsMasterResponse> MongosTopologyCoordinator::_makeIsMasterResponse(
+    WithLock lock) const {
+    auto response = std::make_shared<MongosIsMasterResponse>(_topologyVersion);
+    return response;
+}
+
+std::shared_ptr<const MongosIsMasterResponse> MongosTopologyCoordinator::awaitIsMasterResponse(
+    OperationContext* opCtx,
+    boost::optional<TopologyVersion> clientTopologyVersion,
+    boost::optional<long long> maxAwaitTimeMS) const {
+    stdx::unique_lock lk(_mutex);
+
+    // Respond immediately if:
+    // (1) There is no clientTopologyVersion, which indicates that the client is not using
+    //     awaitable ismaster.
+    // (2) The process IDs are different.
+    // (3) The clientTopologyVersion counter is less than mongos' counter.
+    if (!clientTopologyVersion ||
+        clientTopologyVersion->getProcessId() != _topologyVersion.getProcessId() ||
+        clientTopologyVersion->getCounter() < _topologyVersion.getCounter()) {
+        return _makeIsMasterResponse(lk);
+    }
+    uassert(51761,
+            str::stream() << "Received a topology version with counter: "
+                          << clientTopologyVersion->getCounter()
+                          << " which is greater than the mongos topology version counter: "
+                          << _topologyVersion.getCounter(),
+            clientTopologyVersion->getCounter() == _topologyVersion.getCounter());
+
+    lk.unlock();
+
+    // At this point, we have verified that clientTopologyVersion is not none. It this is true,
+    // maxAwaitTimeMS must also be not none.
+    invariant(maxAwaitTimeMS);
+
+    IsMasterMetrics::get(opCtx)->incrementNumAwaitingTopologyChanges();
+
+    ON_BLOCK_EXIT([&] { IsMasterMetrics::get(opCtx)->decrementNumAwaitingTopologyChanges(); });
+
+    if (MONGO_unlikely(hangWhileWaitingForIsMasterResponse.shouldFail())) {
+        LOGV2(4695501, "hangWhileWaitingForIsMasterResponse failpoint enabled");
+        hangWhileWaitingForIsMasterResponse.pauseWhileSet(opCtx);
+    }
+
+    // Sleep for maxTimeMS.
+    LOGV2_DEBUG(4695502,
+                1,
+                "Waiting for an isMaster response for maxAwaitTimeMS",
+                "maxAwaitTimeMS"_attr = maxAwaitTimeMS.get(),
+                "currentMongosTopologyVersionCounter"_attr = _topologyVersion.getCounter());
+
+    opCtx->sleepFor(Milliseconds(*maxAwaitTimeMS));
+
+    lk.lock();
+    return _makeIsMasterResponse(lk);
+}
+
+}  // namespace mongo
diff --git a/src/mongo/s/mongos_topology_coordinator.h b/src/mongo/s/mongos_topology_coordinator.h
new file mode 100644
index 00000000000..4c19655d50b
--- /dev/null
+++ b/src/mongo/s/mongos_topology_coordinator.h
@@ -0,0 +1,83 @@
+/**
+ *    Copyright (C) 2020-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#pragma once
+
+#include "mongo/db/operation_context.h"
+#include "mongo/platform/mutex.h"
+#include "mongo/rpc/topology_version_gen.h"
+#include "mongo/s/mongos_is_master_response.h"
+#include "mongo/transport/ismaster_metrics.h"
+#include "mongo/util/concurrency/with_lock.h"
+#include <memory>
+
+namespace mongo {
+
+class MongosTopologyCoordinator {
+public:
+    static MongosTopologyCoordinator* get(OperationContext* opCtx);
+
+    /**
+     * Constructs a Mongos Topology Coordinator object.
+     **/
+    MongosTopologyCoordinator();
+
+    /**
+     * Constructs and returns a MongosIsMasterResponse. Will block until the given deadline waiting
+     * for a significant topology change if the 'counter' field of 'clientTopologyVersion' is equal
+     * to the current TopologyVersion 'counter' maintained by this class. Returns immediately if
+     * 'clientTopologyVersion' < TopologyVersion of this class or if the processId
+     * differs.
+     *
+     * Note that Quiesce Mode is the only valid topology change on mongos.
+     */
+    std::shared_ptr<const MongosIsMasterResponse> awaitIsMasterResponse(
+        OperationContext* opCtx,
+        boost::optional<TopologyVersion> clientTopologyVersion,
+        boost::optional<long long> maxAwaitTimeMS) const;
+
+private:
+    /**
+     * Helper for constructing a MongosIsMasterResponse.
+     **/
+    std::shared_ptr<MongosIsMasterResponse> _makeIsMasterResponse(WithLock) const;
+    //
+    // All member variables are labeled with one of the following codes indicating the
+    // synchronization rules for accessing them.
+    //
+    // (M)  Reads and writes guarded by _mutex
+
+    // Protects member data of this MongosTopologyCoordinator.
+    mutable Mutex _mutex = MONGO_MAKE_LATCH("MongosTopologyCoordinator::_mutex");
+
+    // Keeps track of the current mongos TopologyVersion.
+    TopologyVersion _topologyVersion;  // (M)
+};
+
+}  // namespace mongo