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_registry.cpp97
-rw-r--r--src/mongo/s/client/shard_registry.h8
-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
5 files changed, 91 insertions, 42 deletions
diff --git a/src/mongo/s/client/shard_registry.cpp b/src/mongo/s/client/shard_registry.cpp
index d722a9116ec..0b1691af82b 100644
--- a/src/mongo/s/client/shard_registry.cpp
+++ b/src/mongo/s/client/shard_registry.cpp
@@ -130,12 +130,8 @@ 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,
- fetchedFromConfigServers] = [&]()
- -> std::tuple<ShardRegistryData, Timestamp, Increment, ShardRegistryData::ShardMap, bool> {
+ auto [returnData, returnTopologyTime, returnForceReloadIncrement, removedShards] =
+ [&]() -> std::tuple<ShardRegistryData, Timestamp, Increment, ShardRegistryData::ShardMap> {
if (timeInStore.topologyTime > cachedData.getTime().topologyTime ||
timeInStore.forceReloadIncrement > cachedData.getTime().forceReloadIncrement) {
auto [reloadedData, maxTopologyTime] =
@@ -144,14 +140,12 @@ ShardRegistry::Cache::LookupResult ShardRegistry::_lookup(OperationContext* opCt
auto [mergedData, removedShards] =
ShardRegistryData::mergeExisting(*cachedData, reloadedData);
- return {
- mergedData, maxTopologyTime, timeInStore.forceReloadIncrement, removedShards, true};
+ return {mergedData, maxTopologyTime, timeInStore.forceReloadIncrement, removedShards};
} else {
return {*cachedData,
cachedData.getTime().topologyTime,
cachedData.getTime().forceReloadIncrement,
- {},
- false};
+ {}};
}
}();
@@ -186,11 +180,6 @@ 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,
@@ -218,9 +207,9 @@ void ShardRegistry::startupPeriodicReloader(OperationContext* opCtx) {
AsyncTry([this] {
LOGV2_DEBUG(22726, 1, "Reloading shardRegistry");
- return _reloadInternal();
+ return _reloadAsyncNoRetry();
})
- .until([](auto sw) {
+ .until([](auto&& sw) {
if (!sw.isOK()) {
LOGV2(22727,
"Error running periodic reload of shard registry",
@@ -232,7 +221,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",
@@ -295,6 +284,49 @@ 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()) {
@@ -380,8 +412,18 @@ std::unique_ptr<Shard> ShardRegistry::createConnection(const ConnectionString& c
return _shardFactory->createUniqueShard(ShardId("<unnamed>"), connStr);
}
-bool ShardRegistry::isUp() const {
- return _isUp.load();
+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;
}
void ShardRegistry::toBSON(BSONObjBuilder* result) const {
@@ -400,23 +442,26 @@ 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.
- _reloadInternal().get(opCtx);
+ return _reloadAsyncNoRetry();
} else {
- AsyncTry([=]() mutable { return _reloadInternal(); })
+ return AsyncTry([=]() mutable { return _reloadAsyncNoRetry(); })
.until([](auto sw) mutable {
return sw.getStatus() != ErrorCodes::ReadConcernMajorityNotAvailableYet;
})
.withBackoffBetweenIterations(kExponentialBackoff)
- .on(Grid::get(opCtx)->getExecutorPool()->getFixedExecutor(),
+ .on(Grid::get(getGlobalServiceContext())->getExecutorPool()->getFixedExecutor(),
CancellationToken::uncancelable())
- .semi()
- .get(opCtx);
+ .share();
}
}
-SharedSemiFuture<ShardRegistry::Cache::ValueHandle> ShardRegistry::_reloadInternal() {
+SharedSemiFuture<ShardRegistry::Cache::ValueHandle> ShardRegistry::_reloadAsyncNoRetry() {
// Make the next acquire do a lookup.
auto value = _forceReloadIncrement.addAndFetch(1);
LOGV2_DEBUG(4620253, 2, "Forcing ShardRegistry reload", "newForceReloadIncrement"_attr = value);
diff --git a/src/mongo/s/client/shard_registry.h b/src/mongo/s/client/shard_registry.h
index 2c329f11b4c..aceb45db9d2 100644
--- a/src/mongo/s/client/shard_registry.h
+++ b/src/mongo/s/client/shard_registry.h
@@ -239,6 +239,9 @@ 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.
@@ -272,7 +275,7 @@ public:
* The ShardRegistry is "up" once a successful lookup from the config servers has been
* completed.
*/
- bool isUp() const;
+ bool isUp();
void toBSON(BSONObjBuilder* result) const;
@@ -438,7 +441,8 @@ private:
void _initializeCacheIfNecessary() const;
- SharedSemiFuture<Cache::ValueHandle> _reloadInternal();
+ SharedSemiFuture<Cache::ValueHandle> _reloadAsync();
+ SharedSemiFuture<Cache::ValueHandle> _reloadAsyncNoRetry();
/**
* Factory to create shards. Never changed after startup so safe to access outside of _mutex.
diff --git a/src/mongo/s/client/sharding_connection_hook.cpp b/src/mongo/s/client/sharding_connection_hook.cpp
index 0a7374a240e..bc72c6ea064 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 isMasterResponse;
- if (!conn->runCommand("admin", BSON("ismaster" << 1), isMasterResponse)) {
- uassertStatusOK(getStatusFromCommandResult(isMasterResponse));
+ BSONObj helloResponse;
+ if (!conn->runCommand("admin", BSON("hello" << 1), helloResponse)) {
+ uassertStatusOK(getStatusFromCommandResult(helloResponse));
}
long long configServerModeNumber;
Status status =
- bsonExtractIntegerField(isMasterResponse, "configsvr", &configServerModeNumber);
+ bsonExtractIntegerField(helloResponse, "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 a6229316eaa..60368d2eb0f 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& isMasterReply) {
- return validateHostImpl(remoteHost, isMasterReply);
+ const executor::RemoteCommandResponse& helloReply) {
+ return validateHostImpl(remoteHost, helloReply);
}
Status ShardingNetworkConnectionHook::validateHostImpl(
- const HostAndPort& remoteHost, const executor::RemoteCommandResponse& isMasterReply) {
+ const HostAndPort& remoteHost, const executor::RemoteCommandResponse& helloReply) {
auto shard =
Grid::get(getGlobalServiceContext())->shardRegistry()->getShardForHostNoReload(remoteHost);
if (!shard) {
@@ -62,11 +62,11 @@ Status ShardingNetworkConnectionHook::validateHostImpl(
}
long long configServerModeNumber;
- auto status = bsonExtractIntegerField(isMasterReply.data, "configsvr", &configServerModeNumber);
+ auto status = bsonExtractIntegerField(helloReply.data, "configsvr", &configServerModeNumber);
switch (status.code()) {
case ErrorCodes::OK: {
- // The ismaster response indicates remoteHost is a config server.
+ // The hello 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 ismaster response indicates that remoteHost is not a config server, or that
+ // The hello 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 ismaster response was malformed.
+ // The hello 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 c77428e56d9..d3715dcec1b 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
- * isMaster response.
+ * "hello" response.
*/
Status validateHost(const HostAndPort& remoteHost,
const BSONObj& request,
- const executor::RemoteCommandResponse& isMasterReply) override;
+ const executor::RemoteCommandResponse& helloReply) override;
/**
* Implementation of validateHost can be called without a ShardingNetworkConnectionHook
* instance.
*/
static Status validateHostImpl(const HostAndPort& remoteHost,
- const executor::RemoteCommandResponse& isMasterReply);
+ const executor::RemoteCommandResponse& helloReply);
/**
* 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 12:10:52 +0000