summaryrefslogtreecommitdiff
path: root/src/mongo/db/s/move_primary_source_manager.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'src/mongo/db/s/move_primary_source_manager.cpp')
-rw-r--r--src/mongo/db/s/move_primary_source_manager.cpp181
1 files changed, 58 insertions, 123 deletions
diff --git a/src/mongo/db/s/move_primary_source_manager.cpp b/src/mongo/db/s/move_primary_source_manager.cpp
index 356ce6c8f88..dea310c8430 100644
--- a/src/mongo/db/s/move_primary_source_manager.cpp
+++ b/src/mongo/db/s/move_primary_source_manager.cpp
@@ -34,7 +34,6 @@
#include "mongo/db/s/move_primary_source_manager.h"
#include "mongo/client/connpool.h"
-#include "mongo/db/catalog/drop_collection.h"
#include "mongo/db/catalog_raii.h"
#include "mongo/db/commands.h"
#include "mongo/db/dbdirectclient.h"
@@ -47,7 +46,6 @@
#include "mongo/rpc/get_status_from_command_result.h"
#include "mongo/s/catalog_cache.h"
#include "mongo/s/grid.h"
-#include "mongo/s/request_types/move_primary_gen.h"
#include "mongo/util/exit.h"
#include "mongo/util/scopeguard.h"
@@ -217,8 +215,6 @@ Status MovePrimarySourceManager::commitOnConfig(OperationContext* opCtx) {
invariant(_state == kCriticalSection);
ScopeGuard scopedGuard([&] { cleanupOnError(opCtx); });
- boost::optional<DatabaseVersion> expectedDbVersion;
-
{
AutoGetDb autoDb(opCtx, getNss().toString(), MODE_X);
@@ -234,13 +230,13 @@ Status MovePrimarySourceManager::commitOnConfig(OperationContext* opCtx) {
// Read operations must begin to wait on the critical section just before we send the
// commit operation to the config server
dss->enterCriticalSectionCommitPhase(opCtx, dssLock, _critSecReason);
-
- expectedDbVersion = dss->getDbVersion(opCtx, dssLock);
}
+ auto configShard = Grid::get(opCtx)->shardRegistry()->getConfigShard();
+
auto commitStatus = [&]() {
try {
- return _commitOnConfig(opCtx, *expectedDbVersion);
+ return _commitOnConfig(opCtx);
} catch (const DBException& ex) {
return ex.toStatus();
}
@@ -248,7 +244,8 @@ Status MovePrimarySourceManager::commitOnConfig(OperationContext* opCtx) {
if (!commitStatus.isOK()) {
// Need to get the latest optime in case the refresh request goes to a secondary --
- // otherwise the read won't wait for the write that commit on config server may have done.
+ // otherwise the read won't wait for the write that _commitOnConfig may have
+ // done
LOGV2(22044,
"Error occurred while committing the movePrimary. Performing a majority write "
"against the config server to obtain its latest optime: {error}",
@@ -332,121 +329,62 @@ Status MovePrimarySourceManager::commitOnConfig(OperationContext* opCtx) {
return Status::OK();
}
-Status MovePrimarySourceManager::_commitOnConfig(OperationContext* opCtx,
- const DatabaseVersion& expectedDbVersion) {
- LOGV2_DEBUG(6854100,
- 3,
- "Committing movePrimary",
- "db"_attr = _dbname,
- "fromShard"_attr = _fromShard,
- "toShard"_attr = _toShard,
- "expectedDbVersion"_attr = expectedDbVersion);
-
- const auto commitStatus = [&] {
- ConfigsvrCommitMovePrimary commitRequest(_dbname.toString(), expectedDbVersion, _toShard);
- commitRequest.setDbName(NamespaceString::kAdminDb);
-
- const auto commitResponse =
- Grid::get(opCtx)->shardRegistry()->getConfigShard()->runCommandWithFixedRetryAttempts(
- opCtx,
- ReadPreferenceSetting(ReadPreference::PrimaryOnly),
- NamespaceString::kAdminDb.toString(),
- CommandHelpers::appendMajorityWriteConcern(commitRequest.toBSON({})),
- Shard::RetryPolicy::kIdempotent);
-
- const auto status = Shard::CommandResponse::getEffectiveStatus(commitResponse);
- if (status != ErrorCodes::CommandNotFound) {
- return status;
- }
+Status MovePrimarySourceManager::_commitOnConfig(OperationContext* opCtx) {
+ auto const configShard = Grid::get(opCtx)->shardRegistry()->getConfigShard();
+
+ auto findResponse = uassertStatusOK(
+ configShard->exhaustiveFindOnConfig(opCtx,
+ ReadPreferenceSetting{ReadPreference::PrimaryOnly},
+ repl::ReadConcernLevel::kMajorityReadConcern,
+ NamespaceString::kConfigDatabasesNamespace,
+ BSON(DatabaseType::kNameFieldName << _dbname),
+ BSON(DatabaseType::kNameFieldName << -1),
+ 1));
+
+ const auto databasesVector = std::move(findResponse.docs);
+ uassert(ErrorCodes::IncompatibleShardingMetadata,
+ str::stream() << "Tried to find max database version for database '" << _dbname
+ << "', but found no databases",
+ !databasesVector.empty());
+
+ const auto dbType =
+ DatabaseType::parse(IDLParserErrorContext("DatabaseType"), databasesVector.front());
+
+ if (dbType.getPrimary() == _toShard) {
+ return Status::OK();
+ }
- LOGV2(6854101,
- "_configsvrCommitMovePrimary command not found on config server, so try to update "
- "the metadata document directly",
- "db"_attr = _dbname);
+ auto newDbType = dbType;
+ newDbType.setPrimary(_toShard);
- // The fallback logic is not synchronized with the removeShard command and simultaneous
- // invocations of movePrimary and removeShard can lead to data loss.
- return _fallbackCommitOnConfig(opCtx, expectedDbVersion);
- }();
+ auto const currentDatabaseVersion = dbType.getVersion();
- if (!commitStatus.isOK()) {
- LOGV2(6854102,
+ newDbType.setVersion(currentDatabaseVersion.makeUpdated());
+
+ auto const updateQuery =
+ BSON(DatabaseType::kNameFieldName << _dbname << DatabaseType::kVersionFieldName
+ << currentDatabaseVersion.toBSON());
+
+ auto updateStatus = Grid::get(opCtx)->catalogClient()->updateConfigDocument(
+ opCtx,
+ NamespaceString::kConfigDatabasesNamespace,
+ updateQuery,
+ newDbType.toBSON(),
+ false,
+ ShardingCatalogClient::kMajorityWriteConcern);
+
+ if (!updateStatus.isOK()) {
+ LOGV2(5448803,
+ "Error committing movePrimary for {db}: {error}",
"Error committing movePrimary",
"db"_attr = _dbname,
- "error"_attr = redact(commitStatus));
- return commitStatus;
+ "error"_attr = redact(updateStatus.getStatus()));
+ return updateStatus.getStatus();
}
- const auto updatedDbType = [&]() {
- const auto findResponse = uassertStatusOK(
- Grid::get(opCtx)->shardRegistry()->getConfigShard()->exhaustiveFindOnConfig(
- opCtx,
- ReadPreferenceSetting{ReadPreference::PrimaryOnly},
- repl::ReadConcernLevel::kMajorityReadConcern,
- NamespaceString::kConfigDatabasesNamespace,
- BSON(DatabaseType::kNameFieldName << _dbname),
- BSON(DatabaseType::kNameFieldName << -1),
- 1));
-
- const auto databases = std::move(findResponse.docs);
- uassert(ErrorCodes::IncompatibleShardingMetadata,
- "Tried to find version for database {}, but found no databases"_format(_dbname),
- !databases.empty());
-
- return DatabaseType::parse(IDLParserErrorContext("DatabaseType"), databases.front());
- }();
- tassert(6851100,
- "Error committing movePrimary: database version went backwards",
- updatedDbType.getVersion() > expectedDbVersion);
- uassert(6851101,
- "Error committing movePrimary: update of config.databases failed",
- updatedDbType.getPrimary() != _fromShard);
-
- LOGV2_DEBUG(6854103,
- 3,
- "Commited movePrimary",
- "db"_attr = _dbname,
- "fromShard"_attr = _fromShard,
- "toShard"_attr = _toShard,
- "updatedDbVersion"_attr = updatedDbType.getVersion());
-
return Status::OK();
}
-Status MovePrimarySourceManager::_fallbackCommitOnConfig(OperationContext* opCtx,
- const DatabaseVersion& expectedDbVersion) {
- const auto query = [&] {
- BSONObjBuilder bsonBuilder;
- bsonBuilder.append(DatabaseType::kNameFieldName, _dbname);
- // Include the version in the update filter to be resilient to potential network retries and
- // delayed messages.
- for (const auto [fieldName, fieldValue] : expectedDbVersion.toBSON()) {
- const auto dottedFieldName = DatabaseType::kVersionFieldName + "." + fieldName;
- bsonBuilder.appendAs(fieldValue, dottedFieldName);
- }
- return bsonBuilder.obj();
- }();
-
- const auto update = [&] {
- const auto newDbVersion = expectedDbVersion.makeUpdated();
-
- BSONObjBuilder bsonBuilder;
- bsonBuilder.append(DatabaseType::kPrimaryFieldName, _toShard);
- bsonBuilder.append(DatabaseType::kVersionFieldName, newDbVersion.toBSON());
- return BSON("$set" << bsonBuilder.obj());
- }();
-
- return Grid::get(opCtx)
- ->catalogClient()
- ->updateConfigDocument(opCtx,
- NamespaceString::kConfigDatabasesNamespace,
- query,
- update,
- false,
- ShardingCatalogClient::kMajorityWriteConcern)
- .getStatus();
-}
-
Status MovePrimarySourceManager::cleanStaleData(OperationContext* opCtx) {
invariant(!opCtx->lockState()->isLocked());
invariant(_state == kNeedCleanStaleData);
@@ -457,21 +395,17 @@ Status MovePrimarySourceManager::cleanStaleData(OperationContext* opCtx) {
}
// Only drop the cloned (unsharded) collections.
+ DBDirectClient client(opCtx);
for (auto& coll : _clonedColls) {
- DropReply unusedDropReply;
- try {
- uassertStatusOK(
- dropCollection(opCtx,
- coll,
- &unusedDropReply,
- DropCollectionSystemCollectionMode::kAllowSystemCollectionDrops,
- false /* fromMigrate */));
- } catch (const DBException& e) {
+ BSONObj dropCollResult;
+ client.runCommand(_dbname.toString(), BSON("drop" << coll.coll()), dropCollResult);
+ Status dropStatus = getStatusFromCommandResult(dropCollResult);
+ if (!dropStatus.isOK()) {
LOGV2(22045,
"Failed to drop cloned collection {namespace} in movePrimary: {error}",
"Failed to drop cloned collection in movePrimary",
"namespace"_attr = coll,
- "error"_attr = redact(e));
+ "error"_attr = redact(dropStatus));
}
}
@@ -479,6 +413,7 @@ Status MovePrimarySourceManager::cleanStaleData(OperationContext* opCtx) {
return Status::OK();
}
+
void MovePrimarySourceManager::cleanupOnError(OperationContext* opCtx) {
if (_state == kDone) {
return;
generated by cgit v1.2.3 (git 2.39.1) at 2026-09-11 14:04:19 +0000