SERVER-33027 add computeAtClusterTime helper

7 files changed, 166 insertions, 2 deletions

diff --git a/src/mongo/s/commands/SConscript b/src/mongo/s/commands/SConscript
index b6dbcc0bbd3..33b5f0813f4 100644
--- a/src/mongo/s/commands/SConscript
+++ b/src/mongo/s/commands/SConscript
@@ -112,9 +112,9 @@ env.Library(
 )
 
 env.CppUnitTest(
-    target='cluster_commands_helpers_test',
+    target='append_at_cluster_time_test',
     source=[
-        'cluster_commands_helpers_test.cpp',
+        'append_at_cluster_time_test.cpp',
     ],
     LIBDEPS=[
         'shared_cluster_commands',
@@ -124,3 +124,18 @@ env.CppUnitTest(
         '$BUILD_DIR/mongo/unittest/unittest',
     ]
 )
+
+env.CppUnitTest(
+    target='compute_at_cluster_time_test',
+    source=[
+        'compute_at_cluster_time_test.cpp',
+    ],
+    LIBDEPS=[
+        'shared_cluster_commands',
+        '$BUILD_DIR/mongo/s/sharding_router_test_fixture',
+        '$BUILD_DIR/mongo/db/auth/authorization_manager_mock_init',
+        '$BUILD_DIR/mongo/db/service_context_noop_init',
+        '$BUILD_DIR/mongo/db/logical_time',
+        '$BUILD_DIR/mongo/unittest/unittest',
+    ]
+)
diff --git a/src/mongo/s/commands/cluster_commands_helpers_test.cpp b/src/mongo/s/commands/append_at_cluster_time_test.cpp
index c5ffa6c7882..c5ffa6c7882 100644
--- a/src/mongo/s/commands/cluster_commands_helpers_test.cpp
+++ b/src/mongo/s/commands/append_at_cluster_time_test.cpp
diff --git a/src/mongo/s/commands/cluster_commands_helpers.cpp b/src/mongo/s/commands/cluster_commands_helpers.cpp
index a8cbf7224ad..c5a5652ace3 100644
--- a/src/mongo/s/commands/cluster_commands_helpers.cpp
+++ b/src/mongo/s/commands/cluster_commands_helpers.cpp
@@ -498,4 +498,18 @@ BSONObj appendAtClusterTime(BSONObj cmdObj, LogicalTime atClusterTime) {
     return cmdAtClusterTimeBob.obj();
 }
 
+LogicalTime computeAtClusterTime(OperationContext* opCtx, std::set<ShardId> shardIds) {
+    auto shardRegistry = Grid::get(opCtx)->shardRegistry();
+    invariant(shardRegistry);
+    LogicalTime highestTime;
+    for (const auto& shardId : shardIds) {
+        auto lastCommittedOpTime =
+            shardRegistry->getShardNoReload(shardId)->getLastCommittedOpTime();
+        if (lastCommittedOpTime > highestTime) {
+            highestTime = lastCommittedOpTime;
+        }
+    }
+    return highestTime;
+}
+
 }  // namespace mongo
diff --git a/src/mongo/s/commands/cluster_commands_helpers.h b/src/mongo/s/commands/cluster_commands_helpers.h
index 4e66a6f964e..1bcb82a0ff8 100644
--- a/src/mongo/s/commands/cluster_commands_helpers.h
+++ b/src/mongo/s/commands/cluster_commands_helpers.h
@@ -167,4 +167,10 @@ bool appendEmptyResultSet(BSONObjBuilder& result, Status status, const std::stri
  */
 StatusWith<CachedDatabaseInfo> createShardDatabase(OperationContext* opCtx, StringData dbName);
 
+/**
+ *  Computes the cluster snapshot time for provided shards. Returns uninitialized LogicalTime if
+ *  the set is empty or every shard's lastCommittedOpTime is not initialized.
+ */
+LogicalTime computeAtClusterTime(OperationContext* opCtx, std::set<ShardId> shardIds);
+
 }  // namespace mongo
diff --git a/src/mongo/s/commands/compute_at_cluster_time_test.cpp b/src/mongo/s/commands/compute_at_cluster_time_test.cpp
new file mode 100644
index 00000000000..66f36da1bc3
--- /dev/null
+++ b/src/mongo/s/commands/compute_at_cluster_time_test.cpp
@@ -0,0 +1,103 @@
+/**
+ * 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.
+ */
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/client/remote_command_targeter_mock.h"
+#include "mongo/db/logical_time.h"
+#include "mongo/s/client/shard_remote.h"
+#include "mongo/s/commands/cluster_commands_helpers.h"
+#include "mongo/s/shard_id.h"
+#include "mongo/s/sharding_router_test_fixture.h"
+#include "mongo/unittest/unittest.h"
+
+namespace mongo {
+namespace {
+
+const HostAndPort kTestConfigShardHost = HostAndPort("FakeConfigHost", 12345);
+
+const ShardId shardOneId("shardOne");
+const HostAndPort shardOne("shardOne:1234");
+
+const ShardId shardTwoId("shardTwo");
+const HostAndPort shardTwo("shardTwo:1234");
+
+class AtClusterTimeTest : public ShardingTestFixture {
+protected:
+    void setUp() {
+        ShardingTestFixture::setUp();
+        configTargeter()->setFindHostReturnValue(kTestConfigShardHost);
+        std::vector<std::tuple<ShardId, HostAndPort>> shardInfos;
+        shardInfos.push_back(std::make_tuple(shardOneId, shardOne));
+        shardInfos.push_back(std::make_tuple(shardTwoId, shardTwo));
+
+        ShardingTestFixture::addRemoteShards(shardInfos);
+    }
+};
+
+TEST_F(AtClusterTimeTest, ComputeValidValid) {
+    auto shardOne = shardRegistry()->getShardNoReload(shardOneId);
+    LogicalTime timeOne(Timestamp(10, 2));
+    shardOne->updateLastCommittedOpTime(timeOne);
+    ASSERT_EQ(timeOne, shardOne->getLastCommittedOpTime());
+
+    auto shardTwo = shardRegistry()->getShardNoReload(shardTwoId);
+    LogicalTime timeTwo(Timestamp(15, 1));
+    shardTwo->updateLastCommittedOpTime(timeTwo);
+    ASSERT_EQ(timeTwo, shardTwo->getLastCommittedOpTime());
+
+    auto maxTime = computeAtClusterTime(operationContext(), {shardOneId, shardTwoId});
+    ASSERT_EQ(maxTime, timeTwo);
+}
+
+TEST_F(AtClusterTimeTest, ComputeValidInvalid) {
+    auto shardOne = shardRegistry()->getShardNoReload(shardOneId);
+    ASSERT_EQ(LogicalTime(), shardOne->getLastCommittedOpTime());
+
+    auto shardTwo = shardRegistry()->getShardNoReload(shardTwoId);
+    LogicalTime timeTwo(Timestamp(15, 1));
+    shardTwo->updateLastCommittedOpTime(timeTwo);
+    ASSERT_EQ(timeTwo, shardTwo->getLastCommittedOpTime());
+
+    auto maxTime = computeAtClusterTime(operationContext(), {shardOneId, shardTwoId});
+    ASSERT_EQ(maxTime, timeTwo);
+}
+
+TEST_F(AtClusterTimeTest, ComputeInvalidInvalid) {
+    auto shardOne = shardRegistry()->getShardNoReload(shardOneId);
+    ASSERT_EQ(LogicalTime(), shardOne->getLastCommittedOpTime());
+
+    auto shardTwo = shardRegistry()->getShardNoReload(shardTwoId);
+    ASSERT_EQ(LogicalTime(), shardTwo->getLastCommittedOpTime());
+
+    auto maxTime = computeAtClusterTime(operationContext(), {shardOneId, shardTwoId});
+    ASSERT_EQ(maxTime, LogicalTime());
+}
+
+}  // namespace
+}  // namespace mongo
diff --git a/src/mongo/s/sharding_router_test_fixture.cpp b/src/mongo/s/sharding_router_test_fixture.cpp
index cee379506c8..03d5fb6d01b 100644
--- a/src/mongo/s/sharding_router_test_fixture.cpp
+++ b/src/mongo/s/sharding_router_test_fixture.cpp
@@ -282,6 +282,27 @@ void ShardingTestFixture::onCommandForPoolExecutor(NetworkTestEnv::OnCommandFunc
     _networkTestEnvForPool->onCommand(func);
 }
 
+void ShardingTestFixture::addRemoteShards(
+    const std::vector<std::tuple<ShardId, HostAndPort>>& shardInfos) {
+    std::vector<ShardType> shards;
+
+    for (auto shard : shardInfos) {
+        ShardType shardType;
+        shardType.setName(std::get<0>(shard).toString());
+        shardType.setHost(std::get<1>(shard).toString());
+        shards.push_back(shardType);
+
+        std::unique_ptr<RemoteCommandTargeterMock> targeter(
+            stdx::make_unique<RemoteCommandTargeterMock>());
+        targeter->setConnectionStringReturnValue(ConnectionString(std::get<1>(shard)));
+        targeter->setFindHostReturnValue(std::get<1>(shard));
+
+        targeterFactory()->addTargeterToReturn(ConnectionString(std::get<1>(shard)),
+                                               std::move(targeter));
+    }
+    setupShards(shards);
+}
+
 void ShardingTestFixture::setupShards(const std::vector<ShardType>& shards) {
     auto future = launchAsync([this] { shardRegistry()->reload(operationContext()); });
 
diff --git a/src/mongo/s/sharding_router_test_fixture.h b/src/mongo/s/sharding_router_test_fixture.h
index 51ee25f0707..574f484faac 100644
--- a/src/mongo/s/sharding_router_test_fixture.h
+++ b/src/mongo/s/sharding_router_test_fixture.h
@@ -90,6 +90,11 @@ public:
     void setupShards(const std::vector<ShardType>& shards);
 
     /**
+     * Adds ShardRemote shards to the shard registry.
+     */
+    void addRemoteShards(const std::vector<std::tuple<ShardId, HostAndPort>>& shards);
+
+    /**
      * Wait for the shards listing command to be run and returns the specified set of shards.
      */
     void expectGetShards(const std::vector<ShardType>& shards);