SERVER-23648 Move executorForAddShard out of ShardRegistry and into addShard imlementation

stbrody · stbrody · commit a6caba6dd563 · 2016-04-18T14:04:52.000-04:00
diff --git a/src/mongo/db/commands/conn_pool_stats.cpp b/src/mongo/db/commands/conn_pool_stats.cpp
@@ -88,6 +88,7 @@ class PoolStats final : public Command {
         auto registry = grid.shardRegistry();
         if (registry) {
             registry->appendConnectionStats(&stats);
+            grid.catalogManager(txn)->appendConnectionStats(&stats);
         }
 
         // Output to a BSON object.
diff --git a/src/mongo/db/s/sharding_state_test.cpp b/src/mongo/db/s/sharding_state_test.cpp
@@ -73,16 +73,8 @@ void initGrid(OperationContext* txn, const ConnectionString& configConnString) {
     auto executorPool = stdx::make_unique<executor::TaskExecutorPool>();
     executorPool->addExecutors(std::move(executorsForPool), std::move(fixedExec));
 
-    // Set up executor used for a few special operations during addShard.
-    auto specialNet(stdx::make_unique<executor::NetworkInterfaceMock>());
-    // auto specialMockNet = specialNet.get();
-    auto specialExec = makeThreadPoolTestExecutor(std::move(specialNet));
-
-    auto shardRegistry(stdx::make_unique<ShardRegistry>(std::move(shardFactory),
-                                                        std::move(executorPool),
-                                                        mockNetwork,
-                                                        std::move(specialExec),
-                                                        configConnString));
+    auto shardRegistry(stdx::make_unique<ShardRegistry>(
+        std::move(shardFactory), std::move(executorPool), mockNetwork, configConnString));
     shardRegistry->startup();
 
     grid.init(
diff --git a/src/mongo/s/catalog/catalog_manager.h b/src/mongo/s/catalog/catalog_manager.h
@@ -63,6 +63,10 @@ template <typename T>
 class StatusWith;
 class TagsType;
 
+namespace executor {
+struct ConnectionPoolStats;
+}
+
 /**
  * Used to indicate to the caller of the removeShard method whether draining of chunks for
  * a particular shard has started, is ongoing, or has been completed.
@@ -439,6 +443,11 @@ class CatalogManager {
     virtual Status appendInfoForConfigServerDatabases(OperationContext* txn,
                                                       BSONArrayBuilder* builder) = 0;
 
+    /**
+     * Append information about the connection pools owned by the CatalogManager.
+     */
+    virtual void appendConnectionStats(executor::ConnectionPoolStats* stats) = 0;
+
 
     virtual StatusWith<DistLockManager::ScopedDistLock> distLock(
         OperationContext* txn,
diff --git a/src/mongo/s/catalog/catalog_manager_mock.cpp b/src/mongo/s/catalog/catalog_manager_mock.cpp
@@ -236,4 +236,6 @@ Status CatalogManagerMock::appendInfoForConfigServerDatabases(OperationContext*
     return Status::OK();
 }
 
+void CatalogManagerMock::appendConnectionStats(executor::ConnectionPoolStats* stats) {}
+
 }  // namespace mongo
diff --git a/src/mongo/s/catalog/catalog_manager_mock.h b/src/mongo/s/catalog/catalog_manager_mock.h
@@ -168,6 +168,8 @@ class CatalogManagerMock : public CatalogManager {
     Status appendInfoForConfigServerDatabases(OperationContext* txn,
                                               BSONArrayBuilder* builder) override;
 
+    void appendConnectionStats(executor::ConnectionPoolStats* stats) override;
+
 private:
     std::unique_ptr<DistLockManagerMock> _mockDistLockMgr;
 };
diff --git a/src/mongo/s/catalog/replset/catalog_manager_replica_set.cpp b/src/mongo/s/catalog/replset/catalog_manager_replica_set.cpp
@@ -124,27 +124,13 @@ void toBatchError(const Status& status, BatchedCommandResponse* response) {
     response->setOk(false);
 }
 
-/**
- * Validates that the specified connection string can serve as a shard server. In particular,
- * this function checks that the shard can be contacted, that it is not already member of
- * another sharded cluster and etc.
- *
- * @param shardRegistry Shard registry to use for opening connections to the shards.
- * @param connectionString Connection string to be attempted as a shard host.
- * @param shardProposedName Optional proposed name for the shard. Can be omitted in which case
- *      a unique name for the shard will be generated from the shard's connection string. If it
- *      is not omitted, the value cannot be the empty string.
- *
- * On success returns a partially initialized shard type object corresponding to the requested
- * shard. It will have the hostName field set and optionally the name, if the name could be
- * generated from either the proposed name or the connection string set name. The returned
- * shard's name should be checked and if empty, one should be generated using some uniform
- * algorithm.
- */
-StatusWith<ShardType> validateHostAsShard(OperationContext* txn,
-                                          ShardRegistry* shardRegistry,
-                                          const ConnectionString& connectionString,
-                                          const std::string* shardProposedName) {
+}  // namespace
+
+StatusWith<ShardType> CatalogManagerReplicaSet::_validateHostAsShard(
+    OperationContext* txn,
+    ShardRegistry* shardRegistry,
+    const ConnectionString& connectionString,
+    const std::string* shardProposedName) {
     if (connectionString.type() == ConnectionString::INVALID) {
         return {ErrorCodes::BadValue, "Invalid connection string"};
     }
@@ -153,14 +139,13 @@ StatusWith<ShardType> validateHostAsShard(OperationContext* txn,
         return {ErrorCodes::BadValue, "shard name cannot be empty"};
     }
 
+    // TODO: Don't create a detached Shard object, create a detached RemoteCommandTargeter instead.
     const std::shared_ptr<Shard> shardConn{shardRegistry->createConnection(connectionString)};
     invariant(shardConn);
-
-    const ReadPreferenceSetting readPref{ReadPreference::PrimaryOnly};
+    auto targeter = shardConn->getTargeter();
 
     // Is it mongos?
-    auto cmdStatus = shardRegistry->runIdempotentCommandForAddShard(
-        txn, shardConn, readPref, "admin", BSON("isdbgrid" << 1));
+    auto cmdStatus = _runCommandForAddShard(txn, targeter.get(), "admin", BSON("isdbgrid" << 1));
     if (!cmdStatus.isOK()) {
         return cmdStatus.getStatus();
     }
@@ -171,8 +156,7 @@ StatusWith<ShardType> validateHostAsShard(OperationContext* txn,
     }
 
     // Is it a replica set?
-    cmdStatus = shardRegistry->runIdempotentCommandForAddShard(
-        txn, shardConn, readPref, "admin", BSON("isMaster" << 1));
+    cmdStatus = _runCommandForAddShard(txn, targeter.get(), "admin", BSON("isMaster" << 1));
     if (!cmdStatus.isOK()) {
         return cmdStatus.getStatus();
     }
@@ -205,8 +189,7 @@ StatusWith<ShardType> validateHostAsShard(OperationContext* txn,
     }
 
     // Is it a mongos config server?
-    cmdStatus = shardRegistry->runIdempotentCommandForAddShard(
-        txn, shardConn, readPref, "admin", BSON("replSetGetStatus" << 1));
+    cmdStatus = _runCommandForAddShard(txn, targeter.get(), "admin", BSON("replSetGetStatus" << 1));
     if (!cmdStatus.isOK()) {
         return cmdStatus.getStatus();
     }
@@ -298,22 +281,15 @@ StatusWith<ShardType> validateHostAsShard(OperationContext* txn,
     return shard;
 }
 
-/**
- * Runs the listDatabases command on the specified host and returns the names of all databases
- * it returns excluding those named local and admin, since they serve administrative purpose.
- */
-StatusWith<std::vector<std::string>> getDBNamesListFromShard(
+StatusWith<std::vector<std::string>> CatalogManagerReplicaSet::_getDBNamesListFromShard(
     OperationContext* txn, ShardRegistry* shardRegistry, const ConnectionString& connectionString) {
+    // TODO: Don't create a detached Shard object, create a detached RemoteCommandTargeter instead.
     const std::shared_ptr<Shard> shardConn{
         shardRegistry->createConnection(connectionString).release()};
     invariant(shardConn);
 
-    auto cmdStatus = shardRegistry->runIdempotentCommandForAddShard(
-        txn,
-        shardConn,
-        ReadPreferenceSetting{ReadPreference::PrimaryOnly},
-        "admin",
-        BSON("listDatabases" << 1));
+    auto cmdStatus = _runCommandForAddShard(
+        txn, shardConn->getTargeter().get(), "admin", BSON("listDatabases" << 1));
     if (!cmdStatus.isOK()) {
         return cmdStatus.getStatus();
     }
@@ -338,15 +314,22 @@ StatusWith<std::vector<std::string>> getDBNamesListFromShard(
     return dbNames;
 }
 
-}  // namespace
-
-CatalogManagerReplicaSet::CatalogManagerReplicaSet(std::unique_ptr<DistLockManager> distLockManager)
-    : _distLockManager(std::move(distLockManager)) {}
+CatalogManagerReplicaSet::CatalogManagerReplicaSet(
+    std::unique_ptr<DistLockManager> distLockManager,
+    std::unique_ptr<executor::TaskExecutor> addShardExecutor)
+    : _distLockManager(std::move(distLockManager)),
+      _executorForAddShard(std::move(addShardExecutor)) {}
 
 CatalogManagerReplicaSet::~CatalogManagerReplicaSet() = default;
 
 Status CatalogManagerReplicaSet::startup(OperationContext* txn) {
+    stdx::lock_guard<stdx::mutex> lk(_mutex);
+    if (_started) {
+        return Status::OK();
+    }
+    _started = true;
     _distLockManager->startUp();
+    _executorForAddShard->startup();
     return Status::OK();
 }
 
@@ -359,6 +342,43 @@ void CatalogManagerReplicaSet::shutDown(OperationContext* txn) {
 
     invariant(_distLockManager);
     _distLockManager->shutDown(txn);
+    _executorForAddShard->shutdown();
+    _executorForAddShard->join();
+}
+
+StatusWith<BSONObj> CatalogManagerReplicaSet::_runCommandForAddShard(
+    OperationContext* txn,
+    RemoteCommandTargeter* targeter,
+    const std::string& dbName,
+    const BSONObj& cmdObj) {
+    auto host = targeter->findHost(ReadPreferenceSetting{ReadPreference::PrimaryOnly},
+                                   RemoteCommandTargeter::selectFindHostMaxWaitTime(txn));
+    if (!host.isOK()) {
+        return host.getStatus();
+    }
+
+    executor::RemoteCommandRequest request(
+        host.getValue(), dbName, cmdObj, rpc::makeEmptyMetadata(), Seconds(30));
+    StatusWith<executor::RemoteCommandResponse> responseStatus =
+        Status(ErrorCodes::InternalError, "Internal error running command");
+
+    auto callStatus = _executorForAddShard->scheduleRemoteCommand(
+        request,
+        [&responseStatus](const executor::TaskExecutor::RemoteCommandCallbackArgs& args) {
+            responseStatus = args.response;
+        });
+    if (!callStatus.isOK()) {
+        return callStatus.getStatus();
+    }
+
+    // Block until the command is carried out
+    _executorForAddShard->wait(callStatus.getValue());
+
+    if (!responseStatus.isOK()) {
+        return responseStatus.getStatus();
+    }
+
+    return responseStatus.getValue().data.getOwned();
 }
 
 StatusWith<string> CatalogManagerReplicaSet::addShard(OperationContext* txn,
@@ -367,7 +387,7 @@ StatusWith<string> CatalogManagerReplicaSet::addShard(OperationContext* txn,
                                                       const long long maxSize) {
     // Validate the specified connection string may serve as shard at all
     auto shardStatus =
-        validateHostAsShard(txn, grid.shardRegistry(), shardConnectionString, shardProposedName);
+        _validateHostAsShard(txn, grid.shardRegistry(), shardConnectionString, shardProposedName);
     if (!shardStatus.isOK()) {
         // TODO: This is a workaround for the case were we could have some bad shard being
         // requested to be added and we put that bad connection string on the global replica set
@@ -379,7 +399,7 @@ StatusWith<string> CatalogManagerReplicaSet::addShard(OperationContext* txn,
 
     ShardType& shardType = shardStatus.getValue();
 
-    auto dbNamesStatus = getDBNamesListFromShard(txn, grid.shardRegistry(), shardConnectionString);
+    auto dbNamesStatus = _getDBNamesListFromShard(txn, grid.shardRegistry(), shardConnectionString);
     if (!dbNamesStatus.isOK()) {
         return dbNamesStatus.getStatus();
     }
@@ -1969,4 +1989,8 @@ Status CatalogManagerReplicaSet::appendInfoForConfigServerDatabases(OperationCon
     return Status::OK();
 }
 
+void CatalogManagerReplicaSet::appendConnectionStats(executor::ConnectionPoolStats* stats) {
+    _executorForAddShard->appendConnectionStats(stats);
+}
+
 }  // namespace mongo
diff --git a/src/mongo/s/catalog/replset/catalog_manager_replica_set.h b/src/mongo/s/catalog/replset/catalog_manager_replica_set.h
@@ -45,7 +45,8 @@ class VersionType;
  */
 class CatalogManagerReplicaSet final : public CatalogManager {
 public:
-    explicit CatalogManagerReplicaSet(std::unique_ptr<DistLockManager> distLockManager);
+    CatalogManagerReplicaSet(std::unique_ptr<DistLockManager> distLockManager,
+                             std::unique_ptr<executor::TaskExecutor> addShardExecutor);
     virtual ~CatalogManagerReplicaSet();
 
     /**
@@ -179,6 +180,8 @@ class CatalogManagerReplicaSet final : public CatalogManager {
     Status appendInfoForConfigServerDatabases(OperationContext* txn,
                                               BSONArrayBuilder* builder) override;
 
+    void appendConnectionStats(executor::ConnectionPoolStats* stats) override;
+
     /**
      * Runs a read command against the config server with majority read concern.
      */
@@ -212,6 +215,37 @@ class CatalogManagerReplicaSet final : public CatalogManager {
      */
     StatusWith<std::string> _generateNewShardName(OperationContext* txn);
 
+    /**
+     * Validates that the specified connection string can serve as a shard server. In particular,
+     * this function checks that the shard can be contacted, that it is not already member of
+     * another sharded cluster and etc.
+     *
+     * @param shardRegistry Shard registry to use for getting a targeter to the shard-to-be.
+     * @param connectionString Connection string to be attempted as a shard host.
+     * @param shardProposedName Optional proposed name for the shard. Can be omitted in which case
+     *      a unique name for the shard will be generated from the shard's connection string. If it
+     *      is not omitted, the value cannot be the empty string.
+     *
+     * On success returns a partially initialized ShardType object corresponding to the requested
+     * shard. It will have the hostName field set and optionally the name, if the name could be
+     * generated from either the proposed name or the connection string set name. The returned
+     * shard's name should be checked and if empty, one should be generated using some uniform
+     * algorithm.
+     */
+    StatusWith<ShardType> _validateHostAsShard(OperationContext* txn,
+                                               ShardRegistry* shardRegistry,
+                                               const ConnectionString& connectionString,
+                                               const std::string* shardProposedName);
+
+    /**
+     * Runs the listDatabases command on the specified host and returns the names of all databases
+     * it returns excluding those named local and admin, since they serve administrative purpose.
+     */
+    StatusWith<std::vector<std::string>> _getDBNamesListFromShard(
+        OperationContext* txn,
+        ShardRegistry* shardRegistry,
+        const ConnectionString& connectionString);
+
     /**
      * Creates the specified collection name in the config database.
      */
@@ -236,6 +270,15 @@ class CatalogManagerReplicaSet final : public CatalogManager {
                                                    const NamespaceString& ns,
                                                    BSONObj query);
 
+    /**
+     * Runs a command against a "shard" that is not yet in the cluster and thus not present in the
+     * ShardRegistry.
+     */
+    StatusWith<BSONObj> _runCommandForAddShard(OperationContext* txn,
+                                               RemoteCommandTargeter* targeter,
+                                               const std::string& dbName,
+                                               const BSONObj& cmdObj);
+
     StatusWith<repl::OpTimeWith<std::vector<BSONObj>>> _exhaustiveFindOnConfig(
         OperationContext* txn,
         const ReadPreferenceSetting& readPref,
@@ -289,12 +332,20 @@ class CatalogManagerReplicaSet final : public CatalogManager {
 
     stdx::mutex _mutex;
 
-    // Distribted lock manager singleton.
+    // Distributed lock manager singleton.
     std::unique_ptr<DistLockManager> _distLockManager;  // (R)
 
+    // Executor specifically used for sending commands to servers that are in the process of being
+    // added as shards.  Does not have any connection hook set on it, thus it can be used to talk
+    // to servers that are not yet in the ShardRegistry.
+    std::unique_ptr<executor::TaskExecutor> _executorForAddShard;  // (R)
+
     // True if shutDown() has been called. False, otherwise.
     bool _inShutdown = false;  // (M)
 
+    // True if startup() has been called.
+    bool _started = false;  // (M)
+
     // Last known highest opTime from the config server.
     repl::OpTime _configOpTime;  // (M)
 
diff --git a/src/mongo/s/catalog/replset/catalog_manager_replica_set_add_shard_test.cpp b/src/mongo/s/catalog/replset/catalog_manager_replica_set_add_shard_test.cpp
@@ -485,12 +485,11 @@ TEST_F(AddShardTest, UnreachableHost) {
         ASSERT_EQUALS("host unreachable", status.getStatus().reason());
     });
 
-    for (int i = 0; i < 3; i++) {  // ShardRegistry will retry 3 times
-        onCommandForAddShard([](const RemoteCommandRequest& request) {
-            ASSERT_EQ(request.target, HostAndPort("StandaloneHost:12345"));
-            return StatusWith<BSONObj>{ErrorCodes::HostUnreachable, "host unreachable"};
-        });
-    }
+    onCommandForAddShard([](const RemoteCommandRequest& request) {
+        ASSERT_EQ(request.target, HostAndPort("StandaloneHost:12345"));
+        return StatusWith<BSONObj>{ErrorCodes::HostUnreachable, "host unreachable"};
+    });
+
 
     future.timed_get(kFutureTimeout);
 }
diff --git a/src/mongo/s/catalog/replset/dist_lock_catalog_impl_test.cpp b/src/mongo/s/catalog/replset/dist_lock_catalog_impl_test.cpp
@@ -122,15 +122,9 @@ class DistLockCatalogFixture : public mongo::unittest::Test {
         auto executorPool = stdx::make_unique<executor::TaskExecutorPool>();
         executorPool->addExecutors(std::move(executorsForPool), std::move(fixedExecutor));
 
-        auto addShardExecutor = executor::makeThreadPoolTestExecutor(
-            stdx::make_unique<executor::NetworkInterfaceMock>());
-
         ConnectionString configCS(HostAndPort("dummy:1234"));
-        _shardRegistry = stdx::make_unique<ShardRegistry>(stdx::make_unique<ShardFactoryMock>(),
-                                                          std::move(executorPool),
-                                                          network,
-                                                          std::move(addShardExecutor),
-                                                          configCS);
+        _shardRegistry = stdx::make_unique<ShardRegistry>(
+            stdx::make_unique<ShardFactoryMock>(), std::move(executorPool), network, configCS);
         _shardRegistry->startup();
 
         _distLockCatalog = stdx::make_unique<DistLockCatalogImpl>(_shardRegistry.get());
diff --git a/src/mongo/s/client/shard_registry.cpp b/src/mongo/s/client/shard_registry.cpp
diff --git a/src/mongo/s/client/shard_registry.h b/src/mongo/s/client/shard_registry.h
diff --git a/src/mongo/s/sharding_initialization.cpp b/src/mongo/s/sharding_initialization.cpp
diff --git a/src/mongo/s/sharding_test_fixture.cpp b/src/mongo/s/sharding_test_fixture.cpp

Original file line number	Diff line number	Diff line change
`@@ -88,6 +88,7 @@ class PoolStats final : public Command {`
`88`	`88`	`auto registry = grid.shardRegistry();`
`89`	`89`	`if (registry) {`
`90`	`90`	`registry->appendConnectionStats(&stats);`
	`91`	`+ grid.catalogManager(txn)->appendConnectionStats(&stats);`
`91`	`92`	`}`
`92`	`93`
`93`	`94`	`// Output to a BSON object.`
Original file line number	Diff line number	Diff line change
`@@ -236,4 +236,6 @@ Status CatalogManagerMock::appendInfoForConfigServerDatabases(OperationContext*`
`236`	`236`	`return Status::OK();`
`237`	`237`	`}`
`238`	`238`
	`239`	`+void CatalogManagerMock::appendConnectionStats(executor::ConnectionPoolStats* stats) {}`
	`240`	`+`
`239`	`241`	`} // namespace mongo`