summaryrefslogtreecommitdiff
path: root/src/mongo/s/client
diff options
context:
space:
mode:
Diffstat (limited to 'src/mongo/s/client')
-rw-r--r--src/mongo/s/client/shard.cpp61
-rw-r--r--src/mongo/s/client/shard.h25
-rw-r--r--src/mongo/s/client/shard_registry.cpp99
-rw-r--r--src/mongo/s/client/shard_registry.h8
-rw-r--r--src/mongo/s/client/shard_remote.cpp31
-rw-r--r--src/mongo/s/client/shard_remote.h6
-rw-r--r--src/mongo/s/client/sharding_connection_hook.cpp8
-rw-r--r--src/mongo/s/client/sharding_network_connection_hook.cpp14
-rw-r--r--src/mongo/s/client/sharding_network_connection_hook.h6
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.
generated by cgit v1.2.3 (git 2.39.1) at 2026-09-11 14:09:13 +0000