diff options
-rw-r--r-- | src/mongo/db/s/create_collection_coordinator.cpp | 5 | ||||
-rw-r--r-- | src/mongo/db/s/create_collection_coordinator.h | 3 | ||||
-rw-r--r-- | src/mongo/db/s/drop_collection_coordinator.cpp | 5 | ||||
-rw-r--r-- | src/mongo/db/s/drop_collection_coordinator.h | 2 | ||||
-rw-r--r-- | src/mongo/db/s/drop_database_coordinator.cpp | 5 | ||||
-rw-r--r-- | src/mongo/db/s/drop_database_coordinator.h | 2 | ||||
-rw-r--r-- | src/mongo/db/s/rename_collection_coordinator.cpp | 5 | ||||
-rw-r--r-- | src/mongo/db/s/rename_collection_coordinator.h | 3 | ||||
-rw-r--r-- | src/mongo/db/s/sharding_ddl_coordinator.cpp | 95 | ||||
-rw-r--r-- | src/mongo/db/s/sharding_ddl_coordinator.h | 7 | ||||
-rw-r--r-- | src/mongo/db/s/sharding_ddl_coordinator_service.cpp | 30 | ||||
-rw-r--r-- | src/mongo/db/s/sharding_ddl_coordinator_service.h | 3 |
12 files changed, 105 insertions, 60 deletions
diff --git a/src/mongo/db/s/create_collection_coordinator.cpp b/src/mongo/db/s/create_collection_coordinator.cpp index a03877475c4..351ac3df87c 100644 --- a/src/mongo/db/s/create_collection_coordinator.cpp +++ b/src/mongo/db/s/create_collection_coordinator.cpp @@ -354,8 +354,9 @@ void broadcastDropCollection(OperationContext* opCtx, } // namespace -CreateCollectionCoordinator::CreateCollectionCoordinator(const BSONObj& initialState) - : ShardingDDLCoordinator(initialState), +CreateCollectionCoordinator::CreateCollectionCoordinator(ShardingDDLCoordinatorService* service, + const BSONObj& initialState) + : ShardingDDLCoordinator(service, initialState), _doc(CreateCollectionCoordinatorDocument::parse( IDLParserErrorContext("CreateCollectionCoordinatorDocument"), initialState)), _critSecReason(BSON("command" diff --git a/src/mongo/db/s/create_collection_coordinator.h b/src/mongo/db/s/create_collection_coordinator.h index 4d5ea68ce0f..9171e2a6b80 100644 --- a/src/mongo/db/s/create_collection_coordinator.h +++ b/src/mongo/db/s/create_collection_coordinator.h @@ -44,7 +44,8 @@ public: using CoordDoc = CreateCollectionCoordinatorDocument; using Phase = CreateCollectionCoordinatorPhaseEnum; - CreateCollectionCoordinator(const BSONObj& initialState); + CreateCollectionCoordinator(ShardingDDLCoordinatorService* service, + const BSONObj& initialState); ~CreateCollectionCoordinator() = default; diff --git a/src/mongo/db/s/drop_collection_coordinator.cpp b/src/mongo/db/s/drop_collection_coordinator.cpp index eb583e81682..04e33213405 100644 --- a/src/mongo/db/s/drop_collection_coordinator.cpp +++ b/src/mongo/db/s/drop_collection_coordinator.cpp @@ -44,8 +44,9 @@ namespace mongo { -DropCollectionCoordinator::DropCollectionCoordinator(const BSONObj& initialState) - : ShardingDDLCoordinator(initialState), +DropCollectionCoordinator::DropCollectionCoordinator(ShardingDDLCoordinatorService* service, + const BSONObj& initialState) + : ShardingDDLCoordinator(service, initialState), _doc(DropCollectionCoordinatorDocument::parse( IDLParserErrorContext("DropCollectionCoordinatorDocument"), initialState)) {} diff --git a/src/mongo/db/s/drop_collection_coordinator.h b/src/mongo/db/s/drop_collection_coordinator.h index 33cf72693b5..01c7ce8ccb9 100644 --- a/src/mongo/db/s/drop_collection_coordinator.h +++ b/src/mongo/db/s/drop_collection_coordinator.h @@ -39,7 +39,7 @@ public: using StateDoc = DropCollectionCoordinatorDocument; using Phase = DropCollectionCoordinatorPhaseEnum; - DropCollectionCoordinator(const BSONObj& initialState); + DropCollectionCoordinator(ShardingDDLCoordinatorService* service, const BSONObj& initialState); ~DropCollectionCoordinator() = default; void checkIfOptionsConflict(const BSONObj& doc) const override {} diff --git a/src/mongo/db/s/drop_database_coordinator.cpp b/src/mongo/db/s/drop_database_coordinator.cpp index f86478f62ba..df4bf544049 100644 --- a/src/mongo/db/s/drop_database_coordinator.cpp +++ b/src/mongo/db/s/drop_database_coordinator.cpp @@ -92,8 +92,9 @@ void removeDatabaseMetadataFromConfig(OperationContext* opCtx, StringData dbName } // namespace -DropDatabaseCoordinator::DropDatabaseCoordinator(const BSONObj& initialState) - : ShardingDDLCoordinator(initialState), +DropDatabaseCoordinator::DropDatabaseCoordinator(ShardingDDLCoordinatorService* service, + const BSONObj& initialState) + : ShardingDDLCoordinator(service, initialState), _doc(DropDatabaseCoordinatorDocument::parse( IDLParserErrorContext("DropDatabaseCoordinatorDocument"), initialState)), _dbName(nss().db()) {} diff --git a/src/mongo/db/s/drop_database_coordinator.h b/src/mongo/db/s/drop_database_coordinator.h index 54651b3ecf3..085e0b6d8cd 100644 --- a/src/mongo/db/s/drop_database_coordinator.h +++ b/src/mongo/db/s/drop_database_coordinator.h @@ -39,7 +39,7 @@ public: using StateDoc = DropDatabaseCoordinatorDocument; using Phase = DropDatabaseCoordinatorPhaseEnum; - DropDatabaseCoordinator(const BSONObj& initialState); + DropDatabaseCoordinator(ShardingDDLCoordinatorService* service, const BSONObj& initialState); ~DropDatabaseCoordinator() = default; void checkIfOptionsConflict(const BSONObj& doc) const override {} diff --git a/src/mongo/db/s/rename_collection_coordinator.cpp b/src/mongo/db/s/rename_collection_coordinator.cpp index c80d696e8a0..d5dbc59f41e 100644 --- a/src/mongo/db/s/rename_collection_coordinator.cpp +++ b/src/mongo/db/s/rename_collection_coordinator.cpp @@ -66,8 +66,9 @@ boost::optional<CollectionType> getShardedCollection(OperationContext* opCtx, } // namespace -RenameCollectionCoordinator::RenameCollectionCoordinator(const BSONObj& initialState) - : ShardingDDLCoordinator(initialState), +RenameCollectionCoordinator::RenameCollectionCoordinator(ShardingDDLCoordinatorService* service, + const BSONObj& initialState) + : ShardingDDLCoordinator(service, initialState), _doc(RenameCollectionCoordinatorDocument::parse( IDLParserErrorContext("RenameCollectionCoordinatorDocument"), initialState)) {} diff --git a/src/mongo/db/s/rename_collection_coordinator.h b/src/mongo/db/s/rename_collection_coordinator.h index e753b946693..1621c323604 100644 --- a/src/mongo/db/s/rename_collection_coordinator.h +++ b/src/mongo/db/s/rename_collection_coordinator.h @@ -40,7 +40,8 @@ public: using StateDoc = RenameCollectionCoordinatorDocument; using Phase = RenameCollectionCoordinatorPhaseEnum; - RenameCollectionCoordinator(const BSONObj& initialState); + RenameCollectionCoordinator(ShardingDDLCoordinatorService* service, + const BSONObj& initialState); void checkIfOptionsConflict(const BSONObj& doc) const override; diff --git a/src/mongo/db/s/sharding_ddl_coordinator.cpp b/src/mongo/db/s/sharding_ddl_coordinator.cpp index 2b112ee4b4e..5e05dae0a09 100644 --- a/src/mongo/db/s/sharding_ddl_coordinator.cpp +++ b/src/mongo/db/s/sharding_ddl_coordinator.cpp @@ -33,12 +33,15 @@ #include "mongo/db/s/sharding_ddl_coordinator.h" -#include "mongo/db/persistent_task_store.h" +#include "mongo/db/dbdirectclient.h" +#include "mongo/db/repl/repl_client_info.h" #include "mongo/db/s/database_sharding_state.h" #include "mongo/db/s/operation_sharding_state.h" #include "mongo/db/s/sharding_ddl_coordinator_gen.h" +#include "mongo/db/write_concern.h" #include "mongo/logv2/log.h" #include "mongo/s/grid.h" +#include "mongo/s/write_ops/batched_command_response.h" namespace mongo { @@ -47,8 +50,10 @@ ShardingDDLCoordinatorMetadata extractShardingDDLCoordinatorMetadata(const BSONO IDLParserErrorContext("ShardingDDLCoordinatorMetadata"), coorDoc); } -ShardingDDLCoordinator::ShardingDDLCoordinator(const BSONObj& coorDoc) - : _coorMetadata(extractShardingDDLCoordinatorMetadata(coorDoc)), +ShardingDDLCoordinator::ShardingDDLCoordinator(ShardingDDLCoordinatorService* service, + const BSONObj& coorDoc) + : _service(service), + _coorMetadata(extractShardingDDLCoordinatorMetadata(coorDoc)), _recoveredFromDisk(_coorMetadata.getRecoveredFromDisk()) {} ShardingDDLCoordinator::~ShardingDDLCoordinator() { @@ -56,17 +61,39 @@ ShardingDDLCoordinator::~ShardingDDLCoordinator() { invariant(_completionPromise.getFuture().isReady()); } -void ShardingDDLCoordinator::_removeDocument(OperationContext* opCtx) { - PersistentTaskStore<ShardingDDLCoordinatorMetadata> store( - NamespaceString::kShardingDDLCoordinatorsNamespace); - LOGV2_DEBUG(5565601, - 2, - "Removing sharding DDL coordinator document", - "coordinatorId"_attr = _coorMetadata.getId()); - store.remove( - opCtx, - BSON(ShardingDDLCoordinatorMetadata::kIdFieldName << _coorMetadata.getId().toBSON()), - WriteConcerns::kMajorityWriteConcern); +bool ShardingDDLCoordinator::_removeDocument(OperationContext* opCtx) { + DBDirectClient dbClient(opCtx); + auto commandResponse = dbClient.runCommand([&] { + write_ops::DeleteCommandRequest deleteOp( + NamespaceString::kShardingDDLCoordinatorsNamespace); + + deleteOp.setDeletes({[&] { + write_ops::DeleteOpEntry entry; + entry.setQ(BSON(ShardingDDLCoordinatorMetadata::kIdFieldName + << _coorMetadata.getId().toBSON())); + entry.setMulti(true); + return entry; + }()}); + + return deleteOp.serialize({}); + }()); + + const auto commandReply = commandResponse->getCommandReply(); + uassertStatusOK(getStatusFromWriteCommandReply(commandReply)); + + BatchedCommandResponse batchedResponse; + std::string unusedErrmsg; + batchedResponse.parseBSON(commandReply, &unusedErrmsg); + + WriteConcernResult ignoreResult; + const WriteConcernOptions majorityWriteConcern{ + WriteConcernOptions::kMajority, + WriteConcernOptions::SyncMode::UNSET, + WriteConcernOptions::kWriteConcernTimeoutSharding}; + auto latestOpTime = repl::ReplClientInfo::forClient(opCtx->getClient()).getLastOp(); + uassertStatusOK(waitForWriteConcern(opCtx, latestOpTime, majorityWriteConcern, &ignoreResult)); + + return batchedResponse.getN() > 0; } void ShardingDDLCoordinator::interrupt(Status status) { @@ -134,7 +161,12 @@ SemiFuture<void> ShardingDDLCoordinator::run(std::shared_ptr<executor::ScopedTas errorMsg, "coordinatorId"_attr = _coorMetadata.getId(), "reason"_attr = redact(status)); - interrupt(status.withContext(errorMsg)); + + stdx::lock_guard<Latch> lg(_mutex); + if (!_constructionCompletionPromise.getFuture().isReady()) { + _constructionCompletionPromise.setError(status); + } + return status; }) .then([this, executor, token, anchor = shared_from_this()] { @@ -144,28 +176,33 @@ SemiFuture<void> ShardingDDLCoordinator::run(std::shared_ptr<executor::ScopedTas auto opCtxHolder = cc().makeOperationContext(); auto* opCtx = opCtxHolder.get(); - const auto completionStatus = [&] { - if (!status.isOK() && - (status.isA<ErrorCategory::NotPrimaryError>() || - status.isA<ErrorCategory::ShutdownError>())) { - // Do not remove the coordinator document - // if we had a stepdown related error. - return status; - } + auto completionStatus = status; + + // Release the coordinator only if we are not stepping down + if (!status.isA<ErrorCategory::NotPrimaryError>() && + !status.isA<ErrorCategory::ShutdownError>()) { try { - _removeDocument(opCtx); - return status; + LOGV2(5565601, + "Releasing sharding DDL coordinator", + "coordinatorId"_attr = _coorMetadata.getId()); + + const auto docWasRemoved = _removeDocument(opCtx); + + if (!docWasRemoved) { + _service->releaseInstance(BSON(ShardingDDLCoordinatorMetadata::kIdFieldName + << _coorMetadata.getId().toBSON()), + status); + } } catch (DBException& ex) { - static constexpr auto& errMsg = - "Failed to remove sharding DDL coordinator document"; + static constexpr auto errMsg = "Failed to release sharding DDL coordinator"; LOGV2_WARNING(5565605, errMsg, "coordinatorId"_attr = _coorMetadata.getId(), "error"_attr = redact(ex)); - return ex.toStatus(errMsg); + completionStatus = ex.toStatus(errMsg); } - }(); + } while (!_scopedLocks.empty()) { _scopedLocks.top().assignNewOpCtx(opCtx); diff --git a/src/mongo/db/s/sharding_ddl_coordinator.h b/src/mongo/db/s/sharding_ddl_coordinator.h index f8586309ba3..e28746ff561 100644 --- a/src/mongo/db/s/sharding_ddl_coordinator.h +++ b/src/mongo/db/s/sharding_ddl_coordinator.h @@ -31,10 +31,10 @@ #include "mongo/db/namespace_string.h" #include "mongo/db/operation_context.h" -#include "mongo/db/repl/primary_only_service.h" #include "mongo/db/s/dist_lock_manager.h" #include "mongo/db/s/forwardable_operation_metadata.h" #include "mongo/db/s/sharding_ddl_coordinator_gen.h" +#include "mongo/db/s/sharding_ddl_coordinator_service.h" #include "mongo/executor/task_executor.h" #include "mongo/util/future.h" @@ -45,7 +45,7 @@ ShardingDDLCoordinatorMetadata extractShardingDDLCoordinatorMetadata(const BSONO class ShardingDDLCoordinator : public repl::PrimaryOnlyService::TypedInstance<ShardingDDLCoordinator> { public: - explicit ShardingDDLCoordinator(const BSONObj& coorDoc); + explicit ShardingDDLCoordinator(ShardingDDLCoordinatorService* service, const BSONObj& coorDoc); ~ShardingDDLCoordinator(); @@ -95,6 +95,7 @@ protected: return {}; }; + ShardingDDLCoordinatorService* _service; ShardingDDLCoordinatorMetadata _coorMetadata; bool _recoveredFromDisk; @@ -111,7 +112,7 @@ private: void interrupt(Status status) override final; - void _removeDocument(OperationContext* opCtx); + bool _removeDocument(OperationContext* opCtx); Mutex _mutex = MONGO_MAKE_LATCH("ShardingDDLCoordinator::_mutex"); SharedPromise<void> _constructionCompletionPromise; diff --git a/src/mongo/db/s/sharding_ddl_coordinator_service.cpp b/src/mongo/db/s/sharding_ddl_coordinator_service.cpp index 70155b6dfc0..ae27f9ff9eb 100644 --- a/src/mongo/db/s/sharding_ddl_coordinator_service.cpp +++ b/src/mongo/db/s/sharding_ddl_coordinator_service.cpp @@ -48,29 +48,24 @@ #include "mongo/db/s/rename_collection_coordinator.h" namespace mongo { +namespace { -ShardingDDLCoordinatorService* ShardingDDLCoordinatorService::getService(OperationContext* opCtx) { - auto registry = repl::PrimaryOnlyServiceRegistry::get(opCtx->getServiceContext()); - auto service = registry->lookupServiceByName(kServiceName); - return checked_cast<ShardingDDLCoordinatorService*>(std::move(service)); -} - -std::shared_ptr<ShardingDDLCoordinator> ShardingDDLCoordinatorService::_constructCoordinator( - BSONObj initialState) const { +std::shared_ptr<ShardingDDLCoordinator> constructShardingDDLCoordinatorInstance( + ShardingDDLCoordinatorService* service, BSONObj initialState) { const auto op = extractShardingDDLCoordinatorMetadata(initialState); LOGV2( 5390510, "Constructing new sharding DDL coordinator", "coordinatorDoc"_attr = op.toBSON()); switch (op.getId().getOperationType()) { case DDLCoordinatorTypeEnum::kDropDatabase: - return std::make_shared<DropDatabaseCoordinator>(std::move(initialState)); + return std::make_shared<DropDatabaseCoordinator>(service, std::move(initialState)); break; case DDLCoordinatorTypeEnum::kDropCollection: - return std::make_shared<DropCollectionCoordinator>(std::move(initialState)); + return std::make_shared<DropCollectionCoordinator>(service, std::move(initialState)); break; case DDLCoordinatorTypeEnum::kRenameCollection: - return std::make_shared<RenameCollectionCoordinator>(std::move(initialState)); + return std::make_shared<RenameCollectionCoordinator>(service, std::move(initialState)); case DDLCoordinatorTypeEnum::kCreateCollection: - return std::make_shared<CreateCollectionCoordinator>(std::move(initialState)); + return std::make_shared<CreateCollectionCoordinator>(service, std::move(initialState)); break; default: uasserted(ErrorCodes::BadValue, @@ -80,9 +75,18 @@ std::shared_ptr<ShardingDDLCoordinator> ShardingDDLCoordinatorService::_construc } } + +} // namespace + +ShardingDDLCoordinatorService* ShardingDDLCoordinatorService::getService(OperationContext* opCtx) { + auto registry = repl::PrimaryOnlyServiceRegistry::get(opCtx->getServiceContext()); + auto service = registry->lookupServiceByName(kServiceName); + return checked_cast<ShardingDDLCoordinatorService*>(std::move(service)); +} + std::shared_ptr<ShardingDDLCoordinatorService::Instance> ShardingDDLCoordinatorService::constructInstance(BSONObj initialState) { - auto coord = _constructCoordinator(std::move(initialState)); + auto coord = constructShardingDDLCoordinatorInstance(this, std::move(initialState)); { stdx::lock_guard lg(_completionMutex); diff --git a/src/mongo/db/s/sharding_ddl_coordinator_service.h b/src/mongo/db/s/sharding_ddl_coordinator_service.h index 559852ab832..b52e7061794 100644 --- a/src/mongo/db/s/sharding_ddl_coordinator_service.h +++ b/src/mongo/db/s/sharding_ddl_coordinator_service.h @@ -31,7 +31,6 @@ #include "mongo/db/namespace_string.h" #include "mongo/db/repl/primary_only_service.h" -#include "mongo/db/s/sharding_ddl_coordinator.h" namespace mongo { @@ -66,8 +65,6 @@ public: void waitForAllCoordinatorsToComplete(OperationContext* opCtx) const; private: - std::shared_ptr<ShardingDDLCoordinator> _constructCoordinator(BSONObj initialState) const; - ExecutorFuture<void> _rebuildService(std::shared_ptr<executor::ScopedTaskExecutor> executor, const CancellationToken& token) override; |