Update upstream source from tag 'upstream/6.0.0'master

Update to upstream version '6.0.0'
with Debian dir 5604a80ec1c96ca76f25f40d78e6ef855abec322

9 files changed, 84 insertions, 174 deletions

diff --git a/src/mongo/s/client/shard.cpp b/src/mongo/s/client/shard.cpp
index 1408b227b1e..ac360694af4 100644
--- a/src/mongo/s/client/shard.cpp
+++ b/src/mongo/s/client/shard.cpp
@@ -41,6 +41,7 @@ namespace mongo {
 namespace {
 
 const int kOnErrorNumRetries = 3;
+
 }  // namespace
 
 Status Shard::CommandResponse::getEffectiveStatus(
@@ -193,6 +194,36 @@ StatusWith<Shard::QueryResponse> Shard::runExhaustiveCursorCommand(
     MONGO_UNREACHABLE;
 }
 
+BatchedCommandResponse Shard::runBatchWriteCommand(OperationContext* opCtx,
+                                                   const Milliseconds maxTimeMS,
+                                                   const BatchedCommandRequest& batchRequest,
+                                                   RetryPolicy retryPolicy) {
+    const StringData dbname = batchRequest.getNS().db();
+    const BSONObj cmdObj = batchRequest.toBSON();
+
+    for (int retry = 1; retry <= kOnErrorNumRetries; ++retry) {
+        // Note: write commands can only be issued against a primary.
+        auto swResponse = _runCommand(
+            opCtx, ReadPreferenceSetting{ReadPreference::PrimaryOnly}, dbname, maxTimeMS, cmdObj);
+
+        BatchedCommandResponse batchResponse;
+        auto writeStatus = CommandResponse::processBatchWriteResponse(swResponse, &batchResponse);
+        if (retry < kOnErrorNumRetries && isRetriableError(writeStatus.code(), retryPolicy)) {
+            LOGV2_DEBUG(22721,
+                        2,
+                        "Batch write command to shard {shardId} failed with retryable error "
+                        "and will be retried. Caused by {error}",
+                        "Batch write command failed with retryable error and will be retried",
+                        "shardId"_attr = getId(),
+                        "error"_attr = redact(writeStatus));
+            continue;
+        }
+
+        return batchResponse;
+    }
+    MONGO_UNREACHABLE;
+}
+
 StatusWith<Shard::QueryResponse> Shard::exhaustiveFindOnConfig(
     OperationContext* opCtx,
     const ReadPreferenceSetting& readPref,
@@ -219,34 +250,4 @@ StatusWith<Shard::QueryResponse> Shard::exhaustiveFindOnConfig(
     MONGO_UNREACHABLE;
 }
 
-BatchedCommandResponse Shard::_submitBatchWriteCommand(OperationContext* opCtx,
-                                                       const BSONObj& serialisedBatchRequest,
-                                                       StringData dbName,
-                                                       Milliseconds maxTimeMS,
-                                                       RetryPolicy retryPolicy) {
-    for (int retry = 1; retry <= kOnErrorNumRetries; ++retry) {
-        // Note: write commands can only be issued against a primary.
-        auto swResponse = _runCommand(opCtx,
-                                      ReadPreferenceSetting{ReadPreference::PrimaryOnly},
-                                      dbName,
-                                      maxTimeMS,
-                                      serialisedBatchRequest);
-
-        BatchedCommandResponse batchResponse;
-        auto writeStatus = CommandResponse::processBatchWriteResponse(swResponse, &batchResponse);
-        if (retry < kOnErrorNumRetries && isRetriableError(writeStatus.code(), retryPolicy)) {
-            LOGV2_DEBUG(22721,
-                        2,
-                        "Batch write command failed with retryable error and will be retried",
-                        "shardId"_attr = getId(),
-                        "error"_attr = redact(writeStatus));
-            continue;
-        }
-
-        return batchResponse;
-    }
-    MONGO_UNREACHABLE;
-}
-
-
 }  // namespace mongo
diff --git a/src/mongo/s/client/shard.h b/src/mongo/s/client/shard.h
index e545f0b2f6b..f690341fcb2 100644
--- a/src/mongo/s/client/shard.h
+++ b/src/mongo/s/client/shard.h
@@ -226,11 +226,10 @@ public:
      * commands return errors in a different format than regular commands do, so checking for
      * retriable errors must be done differently.
      */
-    virtual BatchedCommandResponse runBatchWriteCommand(OperationContext* opCtx,
-                                                        Milliseconds maxTimeMS,
-                                                        const BatchedCommandRequest& batchRequest,
-                                                        const WriteConcernOptions& writeConcern,
-                                                        RetryPolicy retryPolicy) = 0;
+    BatchedCommandResponse runBatchWriteCommand(OperationContext* opCtx,
+                                                Milliseconds maxTimeMS,
+                                                const BatchedCommandRequest& batchRequest,
+                                                RetryPolicy retryPolicy);
 
     /**
      * Warning: This method exhausts the cursor and pulls all data into memory.
@@ -293,22 +292,10 @@ public:
 protected:
     Shard(const ShardId& id);
 
-    /**
-     * Submits the batch request applying the specified retry policy and timeout and using the
-     * machinery provided by each implementation.
-     * Callers of this function must ensure to have configured the write concern settings
-     * accordingly to their specific semantics.
-     */
-    BatchedCommandResponse _submitBatchWriteCommand(OperationContext* opCtx,
-                                                    const BSONObj& serialisedBatchRequest,
-                                                    StringData dbName,
-                                                    Milliseconds maxTimeMS,
-                                                    RetryPolicy retryPolicy);
-
 private:
     /**
-     * Runs the specified command against the shard backed by this object with a timeout set to
-     * the minimum of maxTimeMSOverride or the timeout of the OperationContext.
+     * Runs the specified command against the shard backed by this object with a timeout set to the
+     * minimum of maxTimeMSOverride or the timeout of the OperationContext.
      *
      * The return value exposes RemoteShard's host for calls to updateReplSetMonitor.
      *
diff --git a/src/mongo/s/client/shard_registry.cpp b/src/mongo/s/client/shard_registry.cpp
index dfb3932f3c0..d722a9116ec 100644
--- a/src/mongo/s/client/shard_registry.cpp
+++ b/src/mongo/s/client/shard_registry.cpp
@@ -130,8 +130,12 @@ ShardRegistry::Cache::LookupResult ShardRegistry::_lookup(OperationContext* opCt
     // Check if we need to refresh from the configsvrs.  If so, then do that and get the results,
     // otherwise (this is a lookup only to incorporate updated connection strings from the RSM),
     // then get the equivalent values from the previously cached data.
-    auto [returnData, returnTopologyTime, returnForceReloadIncrement, removedShards] =
-        [&]() -> std::tuple<ShardRegistryData, Timestamp, Increment, ShardRegistryData::ShardMap> {
+    auto [returnData,
+          returnTopologyTime,
+          returnForceReloadIncrement,
+          removedShards,
+          fetchedFromConfigServers] = [&]()
+        -> std::tuple<ShardRegistryData, Timestamp, Increment, ShardRegistryData::ShardMap, bool> {
         if (timeInStore.topologyTime > cachedData.getTime().topologyTime ||
             timeInStore.forceReloadIncrement > cachedData.getTime().forceReloadIncrement) {
             auto [reloadedData, maxTopologyTime] =
@@ -140,12 +144,14 @@ ShardRegistry::Cache::LookupResult ShardRegistry::_lookup(OperationContext* opCt
             auto [mergedData, removedShards] =
                 ShardRegistryData::mergeExisting(*cachedData, reloadedData);
 
-            return {mergedData, maxTopologyTime, timeInStore.forceReloadIncrement, removedShards};
+            return {
+                mergedData, maxTopologyTime, timeInStore.forceReloadIncrement, removedShards, true};
         } else {
             return {*cachedData,
                     cachedData.getTime().topologyTime,
                     cachedData.getTime().forceReloadIncrement,
-                    {}};
+                    {},
+                    false};
         }
     }();
 
@@ -180,6 +186,11 @@ ShardRegistry::Cache::LookupResult ShardRegistry::_lookup(OperationContext* opCt
         }
     }
 
+    // The registry is "up" once there has been a successful lookup from the config servers.
+    if (fetchedFromConfigServers) {
+        _isUp.store(true);
+    }
+
     Time returnTime{returnTopologyTime, rsmIncrementForConnStrings, returnForceReloadIncrement};
     LOGV2_DEBUG(4620251,
                 2,
@@ -207,9 +218,9 @@ void ShardRegistry::startupPeriodicReloader(OperationContext* opCtx) {
 
     AsyncTry([this] {
         LOGV2_DEBUG(22726, 1, "Reloading shardRegistry");
-        return _reloadAsyncNoRetry();
+        return _reloadInternal();
     })
-        .until([](auto&& sw) {
+        .until([](auto sw) {
             if (!sw.isOK()) {
                 LOGV2(22727,
                       "Error running periodic reload of shard registry",
@@ -221,7 +232,7 @@ void ShardRegistry::startupPeriodicReloader(OperationContext* opCtx) {
         })
         .withDelayBetweenIterations(kRefreshPeriod)  // This call is optional.
         .on(_executor, CancellationToken::uncancelable())
-        .getAsync([](auto&& sw) {
+        .getAsync([](auto sw) {
             LOGV2_DEBUG(22725,
                         1,
                         "Exiting periodic shard registry reloader",
@@ -284,49 +295,6 @@ StatusWith<std::shared_ptr<Shard>> ShardRegistry::getShard(OperationContext* opC
     return {ErrorCodes::ShardNotFound, str::stream() << "Shard " << shardId << " not found"};
 }
 
-SemiFuture<std::shared_ptr<Shard>> ShardRegistry::getShard(ExecutorPtr executor,
-                                                           const ShardId& shardId) noexcept {
-
-    // Fetch the shard registry data associated to the latest known topology time
-    return _getDataAsync()
-        .thenRunOn(executor)
-        .then([this, executor, shardId](auto&& cachedData) {
-            // First check if this is a non config shard lookup
-            if (auto shard = cachedData->findShard(shardId)) {
-                return SemiFuture<std::shared_ptr<Shard>>::makeReady(std::move(shard));
-            }
-
-            // then check if this is a config shard (this call is blocking in any case)
-            {
-                stdx::lock_guard<Latch> lk(_mutex);
-                if (auto shard = _configShardData.findShard(shardId)) {
-                    return SemiFuture<std::shared_ptr<Shard>>::makeReady(std::move(shard));
-                }
-            }
-
-            // If the shard was not found, force reload the shard regitry data and try again.
-            //
-            // This is to cover the following scenario:
-            // 1. Primary of the replicaset fetch the list of shards and store it on disk
-            // 2. Primary crash before the latest VectorClock topology time is majority written to
-            //    disk
-            // 3. A new primary with a stale ShardRegistry is elected and read the set of shards
-            //    from disk and calls ShardRegistry::getShard
-
-            return _reloadAsync()
-                .thenRunOn(executor)
-                .then([this, executor, shardId](auto&& cachedData) -> std::shared_ptr<Shard> {
-                    auto shard = cachedData->findShard(shardId);
-                    uassert(ErrorCodes::ShardNotFound,
-                            str::stream() << "Shard " << shardId << " not found",
-                            shard);
-                    return shard;
-                })
-                .semi();
-        })
-        .semi();
-}
-
 std::vector<ShardId> ShardRegistry::getAllShardIds(OperationContext* opCtx) {
     auto shardIds = _getData(opCtx)->getAllShardIds();
     if (shardIds.empty()) {
@@ -412,18 +380,8 @@ std::unique_ptr<Shard> ShardRegistry::createConnection(const ConnectionString& c
     return _shardFactory->createUniqueShard(ShardId("<unnamed>"), connStr);
 }
 
-bool ShardRegistry::isUp() {
-    if (_isUp.load())
-        return true;
-
-    // Before the first lookup is completed, the latest cached value is either empty or it is
-    // associated to the default constructed time
-    const auto latestCached = _cache->peekLatestCached(_kSingleton);
-    if (latestCached && latestCached.getTime() != Time()) {
-        _isUp.store(true);
-        return true;
-    }
-    return false;
+bool ShardRegistry::isUp() const {
+    return _isUp.load();
 }
 
 void ShardRegistry::toBSON(BSONObjBuilder* result) const {
@@ -442,26 +400,23 @@ void ShardRegistry::toBSON(BSONObjBuilder* result) const {
 }
 
 void ShardRegistry::reload(OperationContext* opCtx) {
-    _reloadAsync().get(opCtx);
-}
-
-SharedSemiFuture<ShardRegistry::Cache::ValueHandle> ShardRegistry::_reloadAsync() {
     if (MONGO_unlikely(TestingProctor::instance().isEnabled())) {
         // Some unit tests don't support running the reload's AsyncTry on the fixed executor.
-        return _reloadAsyncNoRetry();
+        _reloadInternal().get(opCtx);
     } else {
-        return AsyncTry([=]() mutable { return _reloadAsyncNoRetry(); })
+        AsyncTry([=]() mutable { return _reloadInternal(); })
             .until([](auto sw) mutable {
                 return sw.getStatus() != ErrorCodes::ReadConcernMajorityNotAvailableYet;
             })
             .withBackoffBetweenIterations(kExponentialBackoff)
-            .on(Grid::get(getGlobalServiceContext())->getExecutorPool()->getFixedExecutor(),
+            .on(Grid::get(opCtx)->getExecutorPool()->getFixedExecutor(),
                 CancellationToken::uncancelable())
-            .share();
+            .semi()
+            .get(opCtx);
     }
 }
 
-SharedSemiFuture<ShardRegistry::Cache::ValueHandle> ShardRegistry::_reloadAsyncNoRetry() {
+SharedSemiFuture<ShardRegistry::Cache::ValueHandle> ShardRegistry::_reloadInternal() {
     // Make the next acquire do a lookup.
     auto value = _forceReloadIncrement.addAndFetch(1);
     LOGV2_DEBUG(4620253, 2, "Forcing ShardRegistry reload", "newForceReloadIncrement"_attr = value);
@@ -614,7 +569,7 @@ std::pair<ShardRegistryData, Timestamp> ShardRegistryData::createFromCatalogClie
     OperationContext* opCtx, ShardFactory* shardFactory) {
     auto const catalogClient = Grid::get(opCtx)->catalogClient();
 
-    auto readConcern = repl::ReadConcernLevel::kSnapshotReadConcern;
+    auto readConcern = repl::ReadConcernLevel::kMajorityReadConcern;
 
     // ShardRemote requires a majority read. We can only allow a non-majority read if we are a
     // config server.
diff --git a/src/mongo/s/client/shard_registry.h b/src/mongo/s/client/shard_registry.h
index aceb45db9d2..2c329f11b4c 100644
--- a/src/mongo/s/client/shard_registry.h
+++ b/src/mongo/s/client/shard_registry.h
@@ -239,9 +239,6 @@ public:
      */
     StatusWith<std::shared_ptr<Shard>> getShard(OperationContext* opCtx, const ShardId& shardId);
 
-    SemiFuture<std::shared_ptr<Shard>> getShard(ExecutorPtr executor,
-                                                const ShardId& shardId) noexcept;
-
     /**
      * Returns a vector containing all known shard IDs.
      * The order of the elements is not guaranteed.
@@ -275,7 +272,7 @@ public:
      * The ShardRegistry is "up" once a successful lookup from the config servers has been
      * completed.
      */
-    bool isUp();
+    bool isUp() const;
 
     void toBSON(BSONObjBuilder* result) const;
 
@@ -441,8 +438,7 @@ private:
 
     void _initializeCacheIfNecessary() const;
 
-    SharedSemiFuture<Cache::ValueHandle> _reloadAsync();
-    SharedSemiFuture<Cache::ValueHandle> _reloadAsyncNoRetry();
+    SharedSemiFuture<Cache::ValueHandle> _reloadInternal();
 
     /**
      * Factory to create shards.  Never changed after startup so safe to access outside of _mutex.
diff --git a/src/mongo/s/client/shard_remote.cpp b/src/mongo/s/client/shard_remote.cpp
index 156dad7171a..b678c276510 100644
--- a/src/mongo/s/client/shard_remote.cpp
+++ b/src/mongo/s/client/shard_remote.cpp
@@ -378,16 +378,10 @@ StatusWith<Shard::QueryResponse> ShardRemote::_exhaustiveFindOnConfig(
     }();
 
     BSONObj readConcernObj = [&] {
-        auto readConcern = [&] {
-            if (readConcernLevel == repl::ReadConcernLevel::kMajorityReadConcern) {
-                repl::OpTime configOpTime{configTime.asTimestamp(),
-                                          mongo::repl::OpTime::kUninitializedTerm};
-                return repl::ReadConcernArgs{configOpTime, readConcernLevel};
-            } else {
-                invariant(readConcernLevel == repl::ReadConcernLevel::kSnapshotReadConcern);
-                return repl::ReadConcernArgs{configTime, readConcernLevel};
-            }
-        }();
+        invariant(readConcernLevel == repl::ReadConcernLevel::kMajorityReadConcern);
+        repl::OpTime configOpTime{configTime.asTimestamp(),
+                                  mongo::repl::OpTime::kUninitializedTerm};
+        repl::ReadConcernArgs readConcern{configOpTime, readConcernLevel};
         BSONObjBuilder bob;
         readConcern.appendInfo(&bob);
         return bob.done().getObjectField(repl::ReadConcernArgs::kReadConcernFieldName).getOwned();
@@ -540,23 +534,6 @@ Status ShardRemote::runAggregation(
 }
 
 
-BatchedCommandResponse ShardRemote::runBatchWriteCommand(OperationContext* opCtx,
-                                                         const Milliseconds maxTimeMS,
-                                                         const BatchedCommandRequest& batchRequest,
-                                                         const WriteConcernOptions& writeConcern,
-                                                         RetryPolicy retryPolicy) {
-    const auto dbName = batchRequest.getNS().db();
-    const BSONObj cmdObj = [&] {
-        BSONObjBuilder cmdObjBuilder;
-        batchRequest.serialize(&cmdObjBuilder);
-        cmdObjBuilder.append(WriteConcernOptions::kWriteConcernField, writeConcern.toBSON());
-        return cmdObjBuilder.obj();
-    }();
-
-    return _submitBatchWriteCommand(opCtx, cmdObj, dbName, maxTimeMS, retryPolicy);
-}
-
-
 StatusWith<ShardRemote::AsyncCmdHandle> ShardRemote::_scheduleCommand(
     OperationContext* opCtx,
     const ReadPreferenceSetting& readPref,
diff --git a/src/mongo/s/client/shard_remote.h b/src/mongo/s/client/shard_remote.h
index 6aec3f87da5..6c99a8a5247 100644
--- a/src/mongo/s/client/shard_remote.h
+++ b/src/mongo/s/client/shard_remote.h
@@ -91,12 +91,6 @@ public:
         std::function<bool(const std::vector<BSONObj>& batch,
                            const boost::optional<BSONObj>& postBatchResumeToken)> callback);
 
-    BatchedCommandResponse runBatchWriteCommand(OperationContext* opCtx,
-                                                Milliseconds maxTimeMS,
-                                                const BatchedCommandRequest& batchRequest,
-                                                const WriteConcernOptions& writeConcern,
-                                                RetryPolicy retryPolicy) final;
-
 private:
     struct AsyncCmdHandle {
         HostAndPort hostTargetted;
diff --git a/src/mongo/s/client/sharding_connection_hook.cpp b/src/mongo/s/client/sharding_connection_hook.cpp
index bc72c6ea064..0a7374a240e 100644
--- a/src/mongo/s/client/sharding_connection_hook.cpp
+++ b/src/mongo/s/client/sharding_connection_hook.cpp
@@ -73,14 +73,14 @@ void ShardingConnectionHook::onCreate(DBClientBase* conn) {
 
 
     if (conn->type() == ConnectionString::ConnectionType::kStandalone) {
-        BSONObj helloResponse;
-        if (!conn->runCommand("admin", BSON("hello" << 1), helloResponse)) {
-            uassertStatusOK(getStatusFromCommandResult(helloResponse));
+        BSONObj isMasterResponse;
+        if (!conn->runCommand("admin", BSON("ismaster" << 1), isMasterResponse)) {
+            uassertStatusOK(getStatusFromCommandResult(isMasterResponse));
         }
 
         long long configServerModeNumber;
         Status status =
-            bsonExtractIntegerField(helloResponse, "configsvr", &configServerModeNumber);
+            bsonExtractIntegerField(isMasterResponse, "configsvr", &configServerModeNumber);
 
         if (status == ErrorCodes::NoSuchKey) {
             // This isn't a config server we're talking to.
diff --git a/src/mongo/s/client/sharding_network_connection_hook.cpp b/src/mongo/s/client/sharding_network_connection_hook.cpp
index 60368d2eb0f..a6229316eaa 100644
--- a/src/mongo/s/client/sharding_network_connection_hook.cpp
+++ b/src/mongo/s/client/sharding_network_connection_hook.cpp
@@ -48,12 +48,12 @@ namespace mongo {
 Status ShardingNetworkConnectionHook::validateHost(
     const HostAndPort& remoteHost,
     const BSONObj&,
-    const executor::RemoteCommandResponse& helloReply) {
-    return validateHostImpl(remoteHost, helloReply);
+    const executor::RemoteCommandResponse& isMasterReply) {
+    return validateHostImpl(remoteHost, isMasterReply);
 }
 
 Status ShardingNetworkConnectionHook::validateHostImpl(
-    const HostAndPort& remoteHost, const executor::RemoteCommandResponse& helloReply) {
+    const HostAndPort& remoteHost, const executor::RemoteCommandResponse& isMasterReply) {
     auto shard =
         Grid::get(getGlobalServiceContext())->shardRegistry()->getShardForHostNoReload(remoteHost);
     if (!shard) {
@@ -62,11 +62,11 @@ Status ShardingNetworkConnectionHook::validateHostImpl(
     }
 
     long long configServerModeNumber;
-    auto status = bsonExtractIntegerField(helloReply.data, "configsvr", &configServerModeNumber);
+    auto status = bsonExtractIntegerField(isMasterReply.data, "configsvr", &configServerModeNumber);
 
     switch (status.code()) {
         case ErrorCodes::OK: {
-            // The hello response indicates remoteHost is a config server.
+            // The ismaster response indicates remoteHost is a config server.
             if (!shard->isConfig()) {
                 return {ErrorCodes::InvalidOptions,
                         str::stream() << "Surprised to discover that " << remoteHost.toString()
@@ -75,7 +75,7 @@ Status ShardingNetworkConnectionHook::validateHostImpl(
             return Status::OK();
         }
         case ErrorCodes::NoSuchKey: {
-            // The hello response indicates that remoteHost is not a config server, or that
+            // The ismaster response indicates that remoteHost is not a config server, or that
             // the config server is running a version prior to the 3.1 development series.
             if (!shard->isConfig()) {
                 return Status::OK();
@@ -86,7 +86,7 @@ Status ShardingNetworkConnectionHook::validateHostImpl(
                                   << " does not believe it is a config server"};
         }
         default:
-            // The hello response was malformed.
+            // The ismaster response was malformed.
             return status;
     }
 }
diff --git a/src/mongo/s/client/sharding_network_connection_hook.h b/src/mongo/s/client/sharding_network_connection_hook.h
index d3715dcec1b..c77428e56d9 100644
--- a/src/mongo/s/client/sharding_network_connection_hook.h
+++ b/src/mongo/s/client/sharding_network_connection_hook.h
@@ -45,18 +45,18 @@ public:
 
     /**
      * Checks that the given host is valid to be used in this sharded cluster, based on its
-     * "hello" response.
+     * isMaster response.
      */
     Status validateHost(const HostAndPort& remoteHost,
                         const BSONObj& request,
-                        const executor::RemoteCommandResponse& helloReply) override;
+                        const executor::RemoteCommandResponse& isMasterReply) override;
 
     /**
      * Implementation of validateHost can be called without a ShardingNetworkConnectionHook
      * instance.
      */
     static Status validateHostImpl(const HostAndPort& remoteHost,
-                                   const executor::RemoteCommandResponse& helloReply);
+                                   const executor::RemoteCommandResponse& isMasterReply);
 
     /**
      * Makes a SetShardVersion request for initializing sharding information on the new connection.