summaryrefslogtreecommitdiff
path: root/src/mongo/db/s/collmod_coordinator.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'src/mongo/db/s/collmod_coordinator.cpp')
-rw-r--r--src/mongo/db/s/collmod_coordinator.cpp239
1 files changed, 58 insertions, 181 deletions
diff --git a/src/mongo/db/s/collmod_coordinator.cpp b/src/mongo/db/s/collmod_coordinator.cpp
index 2dbf5e176e3..9e75175964f 100644
--- a/src/mongo/db/s/collmod_coordinator.cpp
+++ b/src/mongo/db/s/collmod_coordinator.cpp
@@ -31,7 +31,6 @@
#include "mongo/db/s/collmod_coordinator.h"
-#include "mongo/db/catalog/coll_mod.h"
#include "mongo/db/catalog/collection_catalog.h"
#include "mongo/db/catalog/collection_uuid_mismatch.h"
#include "mongo/db/catalog/database_holder.h"
@@ -40,7 +39,6 @@
#include "mongo/db/ops/insert.h"
#include "mongo/db/s/participant_block_gen.h"
#include "mongo/db/s/sharded_collmod_gen.h"
-#include "mongo/db/s/sharding_ddl_coordinator_gen.h"
#include "mongo/db/s/sharding_ddl_util.h"
#include "mongo/db/s/sharding_state.h"
#include "mongo/db/timeseries/catalog_helper.h"
@@ -48,6 +46,7 @@
#include "mongo/db/timeseries/timeseries_options.h"
#include "mongo/idl/idl_parser.h"
#include "mongo/logv2/log.h"
+#include "mongo/s/async_requests_sender.h"
#include "mongo/s/cluster_commands_helpers.h"
#include "mongo/s/grid.h"
#include "mongo/util/fail_point.h"
@@ -135,13 +134,7 @@ void CollModCoordinator::_enterPhase(Phase newPhase) {
if (_doc.getPhase() == Phase::kUnset) {
newDoc = _insertStateDocument(std::move(newDoc));
} else {
- ServiceContext::UniqueOperationContext uniqueOpCtx;
- auto opCtx = cc().getOperationContext();
- if (!opCtx) {
- uniqueOpCtx = cc().makeOperationContext();
- opCtx = uniqueOpCtx.get();
- }
- newDoc = _updateStateDocument(opCtx, std::move(newDoc));
+ newDoc = _updateStateDocument(cc().makeOperationContext().get(), std::move(newDoc));
}
{
@@ -180,85 +173,19 @@ void CollModCoordinator::_saveShardingInfoOnCoordinatorIfNecessary(OperationCont
6522700, "Sharding information must be gathered after collection information", _collInfo);
if (!_shardingInfo && _collInfo->isSharded) {
ShardingInfo info;
- info.isPrimaryOwningChunks = false;
const auto chunkManager =
uassertStatusOK(Grid::get(opCtx)->catalogCache()->getCollectionRoutingInfoWithRefresh(
opCtx, _collInfo->nsForTargeting));
- // Coordinator is guaranteed to be running on primary shard
- info.primaryShard = ShardingState::get(opCtx)->shardId();
-
+ info.primaryShard = chunkManager.dbPrimary();
std::set<ShardId> shardIdsSet;
chunkManager.getAllShardIds(&shardIdsSet);
- std::vector<ShardId> participantsNotOwningChunks;
-
- std::vector<ShardId> shardIdsVec;
- shardIdsVec.reserve(shardIdsSet.size());
- for (const auto& shard : shardIdsSet) {
- if (shard != info.primaryShard) {
- shardIdsVec.push_back(shard);
- } else {
- info.isPrimaryOwningChunks = true;
- }
- }
-
- auto allShards = Grid::get(opCtx)->shardRegistry()->getAllShardIds(opCtx);
- for (const auto& shard : allShards) {
- if (std::find(shardIdsVec.begin(), shardIdsVec.end(), shard) == shardIdsVec.end() &&
- shard != info.primaryShard) {
- participantsNotOwningChunks.push_back(shard);
- }
- }
-
- info.participantsOwningChunks = std::move(shardIdsVec);
- info.participantsNotOwningChunks = std::move(participantsNotOwningChunks);
+ std::vector<ShardId> shardIdsVec{shardIdsSet.begin(), shardIdsSet.end()};
+ info.shardsOwningChunks = std::move(shardIdsVec);
_shardingInfo = std::move(info);
}
}
-std::vector<AsyncRequestsSender::Response> CollModCoordinator::_sendCollModToPrimaryShard(
- OperationContext* opCtx,
- ShardsvrCollModParticipant& request,
- const std::shared_ptr<executor::ScopedTaskExecutor>& executor) {
- // A view definition will only be present on the primary shard. So we pass an addition
- // 'performViewChange' flag only to the primary shard.
- request.setPerformViewChange(true);
-
- return sharding_ddl_util::sendAuthenticatedCommandToShards(
- opCtx,
- nss().db(),
- CommandHelpers::appendMajorityWriteConcern(request.toBSON({})),
- {_shardingInfo->primaryShard},
- **executor,
- _shardingInfo->isPrimaryOwningChunks);
-}
-
-std::vector<AsyncRequestsSender::Response> CollModCoordinator::_sendCollModToParticipantShards(
- OperationContext* opCtx,
- ShardsvrCollModParticipant& request,
- const std::shared_ptr<executor::ScopedTaskExecutor>& executor) {
- request.setPerformViewChange(false);
-
- // The collMod command targets all shards, regardless of whether they have chunks. The shards
- // that have no chunks for the collection will not throw nor will be included in the responses.
-
- sharding_ddl_util::sendAuthenticatedCommandToShards(
- opCtx,
- nss().db(),
- CommandHelpers::appendMajorityWriteConcern(request.toBSON({})),
- _shardingInfo->participantsNotOwningChunks,
- **executor,
- false /* throwOnError */);
-
- return sharding_ddl_util::sendAuthenticatedCommandToShards(
- opCtx,
- nss().db(),
- CommandHelpers::appendMajorityWriteConcern(request.toBSON({})),
- _shardingInfo->participantsOwningChunks,
- **executor,
- true /* throwOnError */);
-}
-
ExecutorFuture<void> CollModCoordinator::_runImpl(
std::shared_ptr<executor::ScopedTaskExecutor> executor,
const CancellationToken& token) noexcept {
@@ -305,37 +232,19 @@ ExecutorFuture<void> CollModCoordinator::_runImpl(
_saveCollectionInfoOnCoordinatorIfNecessary(opCtx);
if (_collInfo->isSharded) {
- const auto migrationsAlreadyBlockedForBucketNss =
- hasTimeSeriesGranularityUpdate(_request) &&
- _doc.getMigrationsAlreadyBlockedForBucketNss();
-
- if (!migrationsAlreadyBlockedForBucketNss) {
- _doc.setCollUUID(sharding_ddl_util::getCollectionUUID(
- opCtx, _collInfo->nsForTargeting, true /* allowViews */));
- sharding_ddl_util::stopMigrations(
- opCtx, _collInfo->nsForTargeting, _doc.getCollUUID());
- }
+ _doc.setCollUUID(
+ sharding_ddl_util::getCollectionUUID(opCtx, nss(), true /* allowViews */));
+ sharding_ddl_util::stopMigrations(opCtx, nss(), _doc.getCollUUID());
}
_saveShardingInfoOnCoordinatorIfNecessary(opCtx);
if (_collInfo->isSharded && hasTimeSeriesGranularityUpdate(_request)) {
- {
- // Persist the migrationAlreadyBlocked flag on the coordinator document
- auto newDoc = _doc;
- newDoc.setMigrationsAlreadyBlockedForBucketNss(true);
- _updateStateDocument(opCtx, std::move(newDoc));
- }
-
ShardsvrParticipantBlock blockCRUDOperationsRequest(_collInfo->nsForTargeting);
const auto cmdObj = CommandHelpers::appendMajorityWriteConcern(
blockCRUDOperationsRequest.toBSON({}));
- std::vector<ShardId> shards = _shardingInfo->participantsOwningChunks;
- if (_shardingInfo->isPrimaryOwningChunks) {
- shards.push_back(_shardingInfo->primaryShard);
- }
sharding_ddl_util::sendAuthenticatedCommandToShards(
- opCtx, nss().db(), cmdObj, shards, **executor, true /* throwOnError */);
+ opCtx, nss().db(), cmdObj, _shardingInfo->shardsOwningChunks, **executor);
}
}))
.then(_executePhase(
@@ -380,92 +289,51 @@ ExecutorFuture<void> CollModCoordinator::_runImpl(
_saveShardingInfoOnCoordinatorIfNecessary(opCtx);
if (_collInfo->isSharded) {
- try {
- if (!_firstExecution) {
- bool allowMigrations = sharding_ddl_util::checkAllowMigrations(
- opCtx, _collInfo->nsForTargeting);
- if (_result.is_initialized() && allowMigrations) {
- // The command finished and we have the response. Return it.
- return;
- } else if (allowMigrations) {
- // Previous run on a different node completed, but we lost the
- // result in the stepdown. Restart from stage in which we disallow
- // migrations.
- _enterPhase(Phase::kBlockShards);
- uasserted(ErrorCodes::Interrupted,
- "Retriable error to move to previous stage");
- }
- }
-
- ShardsvrCollModParticipant request(nss(), _request);
- bool needsUnblock = _collInfo->timeSeriesOptions &&
- hasTimeSeriesGranularityUpdate(_request);
- request.setNeedsUnblock(needsUnblock);
-
- // If trying to convert an index to unique, executes a dryRun first to find
- // any duplicates without actually changing the indexes to avoid
- // inconsistent index specs on different shards. Example:
- // Shard0: {_id: 0, a: 1}
- // Shard1: {_id: 1, a: 2}, {_id: 2, a: 2}
- // When trying to convert index {a: 1} to unique, the dry run will return
- // the duplicate errors to the user without converting the indexes.
- if (isCollModIndexUniqueConversion(_request)) {
- // The 'dryRun' option only works with 'unique' index option. We need to
- // strip out other incompatible options.
- auto dryRunRequest = ShardsvrCollModParticipant{
- nss(), makeCollModDryRunRequest(_request)};
- std::vector<ShardId> shards = _shardingInfo->participantsOwningChunks;
- if (_shardingInfo->isPrimaryOwningChunks) {
- shards.push_back(_shardingInfo->primaryShard);
- }
+ ShardsvrCollModParticipant request(nss(), _request);
+ bool needsUnblock =
+ _collInfo->timeSeriesOptions && hasTimeSeriesGranularityUpdate(_request);
+ request.setNeedsUnblock(needsUnblock);
+
+ std::vector<AsyncRequestsSender::Response> responses;
+ auto shardsOwningChunks = _shardingInfo->shardsOwningChunks;
+ auto primaryShardOwningChunk = std::find(shardsOwningChunks.begin(),
+ shardsOwningChunks.end(),
+ _shardingInfo->primaryShard);
+ // A view definition will only be present on the primary shard. So we pass an
+ // addition 'performViewChange' flag only to the primary shard.
+ if (primaryShardOwningChunk != shardsOwningChunks.end()) {
+ request.setPerformViewChange(true);
+ const auto& primaryResponse =
sharding_ddl_util::sendAuthenticatedCommandToShards(
opCtx,
nss().db(),
- CommandHelpers::appendMajorityWriteConcern(
- dryRunRequest.toBSON({})),
- shards,
+ CommandHelpers::appendMajorityWriteConcern(request.toBSON({})),
+ {_shardingInfo->primaryShard},
**executor);
- }
-
- std::vector<AsyncRequestsSender::Response> responses;
-
- // In the case of the participants, we are broadcasting the collMod to all
- // the shards. On one hand, if the shard contains chunks for the
- // collections, we parse all the responses. On the other hand, if the shard
- // does not contain chunks, we make a best effort to not process the
- // returned responses or throw any errors.
-
- auto primaryResponse = _sendCollModToPrimaryShard(opCtx, request, executor);
- if (_shardingInfo->isPrimaryOwningChunks) {
- responses.insert(responses.end(),
- std::make_move_iterator(primaryResponse.begin()),
- std::make_move_iterator(primaryResponse.end()));
- }
-
- auto participantsResponses =
- _sendCollModToParticipantShards(opCtx, request, executor);
- responses.insert(responses.end(),
- std::make_move_iterator(participantsResponses.begin()),
- std::make_move_iterator(participantsResponses.end()));
-
-
- BSONObjBuilder builder;
- std::string errmsg;
- auto ok =
- appendRawResponses(opCtx, &errmsg, &builder, responses).responseOK;
- if (!errmsg.empty()) {
- CommandHelpers::appendSimpleCommandStatus(builder, ok, errmsg);
- }
- _result = builder.obj();
- sharding_ddl_util::resumeMigrations(
- opCtx, _collInfo->nsForTargeting, _doc.getCollUUID());
- } catch (DBException& ex) {
- if (!_isRetriableErrorForDDLCoordinator(ex.toStatus())) {
- sharding_ddl_util::resumeMigrations(
- opCtx, _collInfo->nsForTargeting, _doc.getCollUUID());
- }
- throw;
+ responses.insert(
+ responses.end(), primaryResponse.begin(), primaryResponse.end());
+ shardsOwningChunks.erase(primaryShardOwningChunk);
}
+
+ request.setPerformViewChange(false);
+ const auto& secondaryResponses =
+ sharding_ddl_util::sendAuthenticatedCommandToShards(
+ opCtx,
+ nss().db(),
+ CommandHelpers::appendMajorityWriteConcern(request.toBSON({})),
+ shardsOwningChunks,
+ **executor);
+ responses.insert(
+ responses.end(), secondaryResponses.begin(), secondaryResponses.end());
+
+ BSONObjBuilder builder;
+ std::string errmsg;
+ auto ok = appendRawResponses(opCtx, &errmsg, &builder, responses).responseOK;
+ if (!errmsg.empty()) {
+ CommandHelpers::appendSimpleCommandStatus(builder, ok, errmsg);
+ }
+ _result = builder.obj();
+ sharding_ddl_util::resumeMigrations(opCtx, nss(), _doc.getCollUUID());
} else {
CollMod cmd(nss());
cmd.setCollModRequest(_request);
@@ -493,6 +361,15 @@ ExecutorFuture<void> CollModCoordinator::_runImpl(
"Error running collMod",
"namespace"_attr = nss(),
"error"_attr = redact(status));
+ // If we have the collection UUID set, this error happened in a sharded collection,
+ // we should restore the migrations.
+ if (_doc.getCollUUID()) {
+ auto opCtxHolder = cc().makeOperationContext();
+ auto* opCtx = opCtxHolder.get();
+ getForwardableOpMetadata().setOn(opCtx);
+
+ sharding_ddl_util::resumeMigrations(opCtx, nss(), _doc.getCollUUID());
+ }
}
return status;
});
generated by cgit v1.2.3 (git 2.39.1) at 2026-09-12 03:19:25 +0000