SERVER-32291 Implement collection creation on config server

4 files changed, 203 insertions, 6 deletions

diff --git a/src/mongo/s/catalog/sharding_catalog_create_collection_test.cpp b/src/mongo/s/catalog/sharding_catalog_create_collection_test.cpp
new file mode 100644
index 00000000000..f3e3bc5aae5
--- /dev/null
+++ b/src/mongo/s/catalog/sharding_catalog_create_collection_test.cpp
@@ -0,0 +1,186 @@
+/**
+ *    Copyright (C) 2018 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    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
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    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 GNU Affero General 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_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kSharding
+
+#include "mongo/platform/basic.h"
+
+#include <set>
+#include <string>
+#include <vector>
+
+#include "mongo/client/read_preference.h"
+#include "mongo/client/remote_command_targeter_factory_mock.h"
+#include "mongo/client/remote_command_targeter_mock.h"
+#include "mongo/db/client.h"
+#include "mongo/db/commands.h"
+#include "mongo/executor/network_interface_mock.h"
+#include "mongo/executor/task_executor.h"
+#include "mongo/s/catalog/sharding_catalog_manager.h"
+#include "mongo/s/catalog/type_database.h"
+#include "mongo/s/client/shard_registry.h"
+#include "mongo/s/config_server_test_fixture.h"
+#include "mongo/s/grid.h"
+#include "mongo/s/shard_key_pattern.h"
+#include "mongo/stdx/future.h"
+#include "mongo/util/log.h"
+#include "mongo/util/scopeguard.h"
+
+namespace mongo {
+namespace {
+
+using executor::NetworkInterfaceMock;
+using executor::RemoteCommandRequest;
+using executor::RemoteCommandResponse;
+using executor::TaskExecutor;
+using std::set;
+using std::string;
+using std::vector;
+using unittest::assertGet;
+
+class CreateCollectionTest : public ConfigServerTestFixture {
+public:
+    void expectCreate(const HostAndPort& receivingHost,
+                      const NamespaceString& expectedNs,
+                      Status response) {
+        onCommand([&](const RemoteCommandRequest& request) {
+            ASSERT_EQUALS(receivingHost, request.target);
+            string cmdName = request.cmdObj.firstElement().fieldName();
+
+            ASSERT_EQUALS("create", cmdName);
+
+            const NamespaceString nss(request.dbname, request.cmdObj.firstElement().String());
+            ASSERT_EQUALS(expectedNs.toString(), nss.toString());
+
+            BSONObjBuilder responseBuilder;
+            CommandHelpers::appendCommandStatus(responseBuilder, response);
+            return responseBuilder.obj();
+        });
+    }
+
+    void expectListCollection(const HostAndPort& receivingHost,
+                              const string& expectedDb,
+                              StatusWith<BSONObj> collectionOptionsReponse,
+                              const UUID& uuid) {
+        onCommand([&](const RemoteCommandRequest& request) {
+            ASSERT_EQUALS(receivingHost, request.target);
+            string cmdName = request.cmdObj.firstElement().fieldName();
+
+            ASSERT_EQUALS("listCollections", cmdName);
+
+            ASSERT_EQUALS(expectedDb, request.dbname);
+
+            BSONObjBuilder responseBuilder;
+
+            if (!collectionOptionsReponse.isOK()) {
+                CommandHelpers::appendCommandStatus(responseBuilder,
+                                                    collectionOptionsReponse.getStatus());
+            } else {
+                BSONObjBuilder listCollResponse(responseBuilder.subobjStart("cursor"));
+                BSONArrayBuilder collArrayBuilder(listCollResponse.subarrayStart("firstBatch"));
+
+
+                BSONObjBuilder collBuilder;
+                collBuilder.append("options", collectionOptionsReponse.getValue());
+                collBuilder.append("info", BSON("uuid" << uuid));
+                collArrayBuilder.append(collBuilder.obj());
+
+                collArrayBuilder.done();
+                listCollResponse.done();
+            }
+
+            return responseBuilder.obj();
+        });
+    }
+
+protected:
+    void setUp() override {
+        ConfigServerTestFixture::setUp();
+
+        extraShard.setName("extra");
+        extraShard.setHost("a:10");
+
+        testPrimaryShard.setName("primary");
+        testPrimaryShard.setHost("b:20");
+
+        uassertStatusOK(setupShards({extraShard, testPrimaryShard}));
+
+        // Prime the shard registry with information about the existing shards
+        shardRegistry()->reload(operationContext());
+
+        // Set up all the target mocks return values.
+        RemoteCommandTargeterMock::get(
+            uassertStatusOK(shardRegistry()->getShard(operationContext(), extraShard.getName()))
+                ->getTargeter())
+            ->setFindHostReturnValue(HostAndPort(extraShard.getHost()));
+        RemoteCommandTargeterMock::get(
+            uassertStatusOK(
+                shardRegistry()->getShard(operationContext(), testPrimaryShard.getName()))
+                ->getTargeter())
+            ->setFindHostReturnValue(HostAndPort(testPrimaryShard.getHost()));
+    }
+
+    const ShardType& getPrimaryShard() const {
+        return testPrimaryShard;
+    }
+
+private:
+    ShardType extraShard;
+    ShardType testPrimaryShard;
+};
+
+TEST_F(CreateCollectionTest, BaseCase) {
+    NamespaceString testNS("test", "foo");
+    const auto& primaryShard = getPrimaryShard();
+
+    setupDatabase(testNS.db().toString(), {primaryShard.getName()}, false);
+
+    CollectionOptions requestOptions;
+    requestOptions.capped = true;
+    requestOptions.cappedSize = 256;
+
+    auto future = launchAsync([this, &testNS, &requestOptions] {
+        ON_BLOCK_EXIT([&] { Client::destroy(); });
+
+        Client::initThreadIfNotAlready("BaseCaseTest");
+        auto opCtx = cc().makeOperationContext();
+        ShardingCatalogManager::get(opCtx.get())
+            ->createCollection(opCtx.get(), testNS, requestOptions);
+    });
+
+    HostAndPort primaryHost(primaryShard.getHost());
+    expectCreate(primaryHost, testNS, Status::OK());
+
+    auto uuid = UUID::gen();
+    auto options = fromjson("{ capped: true, size: 256 }");
+    expectListCollection(primaryHost, testNS.db().toString(), options, uuid);
+    future.timed_get(kFutureTimeout);
+}
+
+}  // namespace
+}  // namespace mongo
diff --git a/src/mongo/s/catalog/type_collection.h b/src/mongo/s/catalog/type_collection.h
index db7c7490a08..612a18b3f62 100644
--- a/src/mongo/s/catalog/type_collection.h
+++ b/src/mongo/s/catalog/type_collection.h
@@ -158,6 +158,10 @@ public:
         return _allowBalance.get_value_or(true);
     }
 
+    void setIsAssignedShardKey(bool isAssignedShardKey) {
+        _isAssignedShardKey = isAssignedShardKey;
+    }
+
     bool isAssignedShardKey() const {
         return _isAssignedShardKey.get_value_or(true);
     }
diff --git a/src/mongo/s/client/shard.cpp b/src/mongo/s/client/shard.cpp
index 721949af326..e2ff6751148 100644
--- a/src/mongo/s/client/shard.cpp
+++ b/src/mongo/s/client/shard.cpp
@@ -47,11 +47,15 @@ namespace {
 
 const int kOnErrorNumRetries = 3;
 
-Status _getEffectiveStatus(const StatusWith<Shard::CommandResponse>& swResponse) {
+}  // namespace
+
+Status Shard::CommandResponse::getEffectiveStatus(
+    const StatusWith<Shard::CommandResponse>& swResponse) {
     // Check if the request even reached the shard.
     if (!swResponse.isOK()) {
         return swResponse.getStatus();
     }
+
     auto& response = swResponse.getValue();
 
     // If the request reached the shard, check if the command failed.
@@ -67,11 +71,9 @@ Status _getEffectiveStatus(const StatusWith<Shard::CommandResponse>& swResponse)
     return Status::OK();
 }
 
-}  // namespace
-
 Status Shard::CommandResponse::processBatchWriteResponse(
     StatusWith<Shard::CommandResponse> swResponse, BatchedCommandResponse* batchResponse) {
-    auto status = _getEffectiveStatus(swResponse);
+    auto status = getEffectiveStatus(swResponse);
     if (status.isOK()) {
         string errmsg;
         if (!batchResponse->parseBSON(swResponse.getValue().response, &errmsg)) {
@@ -126,7 +128,7 @@ StatusWith<Shard::CommandResponse> Shard::runCommand(OperationContext* opCtx,
         }
 
         auto swResponse = _runCommand(opCtx, readPref, dbName, maxTimeMSOverride, cmdObj);
-        auto status = _getEffectiveStatus(swResponse);
+        auto status = CommandResponse::getEffectiveStatus(swResponse);
         if (isRetriableError(status.code(), retryPolicy)) {
             LOG(2) << "Command " << redact(cmdObj)
                    << " failed with retriable error and will be retried"
@@ -163,7 +165,7 @@ StatusWith<Shard::CommandResponse> Shard::runCommandWithFixedRetryAttempts(
         }
 
         auto swResponse = _runCommand(opCtx, readPref, dbName, maxTimeMSOverride, cmdObj);
-        auto status = _getEffectiveStatus(swResponse);
+        auto status = CommandResponse::getEffectiveStatus(swResponse);
         if (retry < kOnErrorNumRetries && isRetriableError(status.code(), retryPolicy)) {
             LOG(2) << "Command " << redact(cmdObj)
                    << " failed with retriable error and will be retried"
diff --git a/src/mongo/s/client/shard.h b/src/mongo/s/client/shard.h
index ca1c0a39ffd..19144129e42 100644
--- a/src/mongo/s/client/shard.h
+++ b/src/mongo/s/client/shard.h
@@ -71,6 +71,11 @@ public:
         static Status processBatchWriteResponse(StatusWith<CommandResponse> response,
                                                 BatchedCommandResponse* batchResponse);
 
+        /**
+         * Returns an error status if either commandStatus or writeConcernStatus has an error.
+         */
+        static Status getEffectiveStatus(const StatusWith<CommandResponse>& swResponse);
+
         boost::optional<HostAndPort> hostAndPort;
         BSONObj response;
         BSONObj metadata;