summaryrefslogtreecommitdiff
path: root/src/mongo/s/catalog/sharding_catalog_client_impl.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'src/mongo/s/catalog/sharding_catalog_client_impl.cpp')
-rw-r--r--src/mongo/s/catalog/sharding_catalog_client_impl.cpp227
1 files changed, 134 insertions, 93 deletions
diff --git a/src/mongo/s/catalog/sharding_catalog_client_impl.cpp b/src/mongo/s/catalog/sharding_catalog_client_impl.cpp
index 38949060bf1..bf98a08c7aa 100644
--- a/src/mongo/s/catalog/sharding_catalog_client_impl.cpp
+++ b/src/mongo/s/catalog/sharding_catalog_client_impl.cpp
@@ -102,6 +102,37 @@ void toBatchError(const Status& status, BatchedCommandResponse* response) {
response->setStatus(status);
}
+void sendRetryableWriteBatchRequestToConfig(OperationContext* opCtx,
+ const NamespaceString& nss,
+ std::vector<BSONObj>& docs,
+ TxnNumber txnNumber,
+ const WriteConcernOptions& writeConcern) {
+ auto configShard = Grid::get(opCtx)->shardRegistry()->getConfigShard();
+
+ BatchedCommandRequest request([&] {
+ write_ops::InsertCommandRequest insertOp(nss);
+ insertOp.setDocuments(docs);
+ return insertOp;
+ }());
+ request.setWriteConcern(writeConcern.toBSON());
+
+ BSONObj cmdObj = request.toBSON();
+ BSONObjBuilder bob(cmdObj);
+ bob.append(OperationSessionInfo::kTxnNumberFieldName, txnNumber);
+
+ BatchedCommandResponse batchResponse;
+ auto response = configShard->runCommand(opCtx,
+ ReadPreferenceSetting{ReadPreference::PrimaryOnly},
+ nss.db().toString(),
+ bob.obj(),
+ Shard::kDefaultConfigCommandTimeout,
+ Shard::RetryPolicy::kIdempotent);
+
+ auto writeStatus = Shard::CommandResponse::processBatchWriteResponse(response, &batchResponse);
+
+ uassertStatusOK(batchResponse.toStatus());
+ uassertStatusOK(writeStatus);
+}
AggregateCommandRequest makeCollectionAndChunksAggregation(OperationContext* opCtx,
const NamespaceString& nss,
@@ -271,45 +302,6 @@ AggregateCommandRequest makeCollectionAndChunksAggregation(OperationContext* opC
return AggregateCommandRequest(CollectionType::ConfigNS, std::move(serializedPipeline));
}
-/**
- * Returns keys for the given purpose and have an expiresAt value greater than newerThanThis on the
- * given shard.
- */
-template <typename KeyDocumentType>
-StatusWith<std::vector<KeyDocumentType>> _getNewKeys(OperationContext* opCtx,
- std::shared_ptr<Shard> shard,
- const NamespaceString& nss,
- StringData purpose,
- const LogicalTime& newerThanThis,
- repl::ReadConcernLevel readConcernLevel) {
- BSONObjBuilder queryBuilder;
- queryBuilder.append("purpose", purpose);
- queryBuilder.append("expiresAt", BSON("$gt" << newerThanThis.asTimestamp()));
-
- auto findStatus = shard->exhaustiveFindOnConfig(opCtx,
- kConfigReadSelector,
- readConcernLevel,
- nss,
- queryBuilder.obj(),
- BSON("expiresAt" << 1),
- boost::none);
- if (!findStatus.isOK()) {
- return findStatus.getStatus();
- }
- const auto& objs = findStatus.getValue().docs;
-
- std::vector<KeyDocumentType> keyDocs;
- keyDocs.reserve(objs.size());
- for (auto&& obj : objs) {
- try {
- keyDocs.push_back(KeyDocumentType::parse(IDLParserErrorContext("keyDoc"), obj));
- } catch (...) {
- return exceptionToStatus();
- }
- }
- return keyDocs;
-}
-
} // namespace
ShardingCatalogClientImpl::ShardingCatalogClientImpl() = default;
@@ -465,10 +457,7 @@ CollectionType ShardingCatalogClientImpl::getCollection(OperationContext* opCtx,
}
std::vector<CollectionType> ShardingCatalogClientImpl::getCollections(
- OperationContext* opCtx,
- StringData dbName,
- repl::ReadConcernLevel readConcernLevel,
- const BSONObj& sort) {
+ OperationContext* opCtx, StringData dbName, repl::ReadConcernLevel readConcernLevel) {
BSONObjBuilder b;
if (!dbName.empty())
b.appendRegex(CollectionType::kNssFieldName,
@@ -480,7 +469,7 @@ std::vector<CollectionType> ShardingCatalogClientImpl::getCollections(
readConcernLevel,
CollectionType::ConfigNS,
b.obj(),
- sort,
+ BSONObj(),
boost::none))
.value;
std::vector<CollectionType> collections;
@@ -493,7 +482,7 @@ std::vector<CollectionType> ShardingCatalogClientImpl::getCollections(
std::vector<NamespaceString> ShardingCatalogClientImpl::getAllShardedCollectionsForDb(
OperationContext* opCtx, StringData dbName, repl::ReadConcernLevel readConcern) {
- auto collectionsOnConfig = getCollections(opCtx, dbName, readConcern, BSONObj());
+ auto collectionsOnConfig = getCollections(opCtx, dbName, readConcern);
std::vector<NamespaceString> collectionsToReturn;
collectionsToReturn.reserve(collectionsOnConfig.size());
@@ -773,35 +762,38 @@ StatusWith<std::vector<TagsType>> ShardingCatalogClientImpl::getTagsForCollectio
}
StatusWith<repl::OpTimeWith<std::vector<ShardType>>> ShardingCatalogClientImpl::getAllShards(
- OperationContext* opCtx, repl::ReadConcernLevel readConcern, bool excludeDraining) {
- const auto& findRes = uassertStatusOK(
- _exhaustiveFindOnConfig(opCtx,
- kConfigReadSelector,
- readConcern,
- ShardType::ConfigNS,
- excludeDraining ? BSON(ShardType::draining.ne(true)) : BSONObj(),
- BSONObj() /* No sorting */,
- boost::none /* No limit */));
+ OperationContext* opCtx, repl::ReadConcernLevel readConcern) {
+ auto findStatus = _exhaustiveFindOnConfig(opCtx,
+ kConfigReadSelector,
+ readConcern,
+ ShardType::ConfigNS,
+ BSONObj(), // no query filter
+ BSONObj(), // no sort
+ boost::none); // no limit
+ if (!findStatus.isOK()) {
+ return findStatus.getStatus();
+ }
std::vector<ShardType> shards;
- shards.reserve(findRes.value.size());
- for (const BSONObj& doc : findRes.value) {
+ shards.reserve(findStatus.getValue().value.size());
+ for (const BSONObj& doc : findStatus.getValue().value) {
auto shardRes = ShardType::fromBSON(doc);
if (!shardRes.isOK()) {
return shardRes.getStatus().withContext(stream()
<< "Failed to parse shard document " << doc);
}
- ShardType& shard = shardRes.getValue();
- if (const Status validateStatus = shard.validate(); !validateStatus.isOK()) {
- return validateStatus.withContext(str::stream()
+ Status validateStatus = shardRes.getValue().validate();
+ if (!validateStatus.isOK()) {
+ return validateStatus.withContext(stream()
<< "Failed to validate shard document " << doc);
}
- shards.push_back(std::move(shard));
+ shards.push_back(shardRes.getValue());
}
- return repl::OpTimeWith<std::vector<ShardType>>{std::move(shards), findRes.opTime};
+ return repl::OpTimeWith<std::vector<ShardType>>{std::move(shards),
+ findStatus.getValue().opTime};
}
Status ShardingCatalogClientImpl::runUserManagementWriteCommand(OperationContext* opCtx,
@@ -1011,14 +1003,12 @@ Status ShardingCatalogClientImpl::insertConfigDocument(OperationContext* opCtx,
insertOp.setDocuments({doc});
return insertOp;
}());
+ request.setWriteConcern(writeConcern.toBSON());
auto configShard = Grid::get(opCtx)->shardRegistry()->getConfigShard();
for (int retry = 1; retry <= kMaxWriteRetry; retry++) {
- auto response = configShard->runBatchWriteCommand(opCtx,
- Shard::kDefaultConfigCommandTimeout,
- request,
- writeConcern,
- Shard::RetryPolicy::kNoRetry);
+ auto response = configShard->runBatchWriteCommand(
+ opCtx, Shard::kDefaultConfigCommandTimeout, request, Shard::RetryPolicy::kNoRetry);
Status status = response.toStatus();
@@ -1073,6 +1063,49 @@ Status ShardingCatalogClientImpl::insertConfigDocument(OperationContext* opCtx,
MONGO_UNREACHABLE;
}
+void ShardingCatalogClientImpl::insertConfigDocumentsAsRetryableWrite(
+ OperationContext* opCtx,
+ const NamespaceString& nss,
+ std::vector<BSONObj> docs,
+ const WriteConcernOptions& writeConcern) {
+ invariant(nss.db() == NamespaceString::kAdminDb || nss.db() == NamespaceString::kConfigDb);
+
+ AlternativeSessionRegion asr(opCtx);
+ TxnNumber currentTxnNumber = 0;
+
+ std::vector<BSONObj> workingBatch;
+ size_t workingBatchItemSize = 0;
+ int workingBatchDocSize = 0;
+
+ while (!docs.empty()) {
+ BSONObj toAdd = docs.back();
+ docs.pop_back();
+
+ const int docSizePlusOverhead =
+ toAdd.objsize() + write_ops::kRetryableAndTxnBatchWriteBSONSizeOverhead;
+ // Check if pushing this object will exceed the batch size limit or the max object size
+ if ((workingBatchItemSize + 1 > write_ops::kMaxWriteBatchSize) ||
+ (workingBatchDocSize + docSizePlusOverhead > BSONObjMaxUserSize)) {
+ sendRetryableWriteBatchRequestToConfig(
+ asr.opCtx(), nss, workingBatch, currentTxnNumber, writeConcern);
+ ++currentTxnNumber;
+
+ workingBatch.clear();
+ workingBatchItemSize = 0;
+ workingBatchDocSize = 0;
+ }
+
+ workingBatch.push_back(toAdd);
+ ++workingBatchItemSize;
+ workingBatchDocSize += docSizePlusOverhead;
+ }
+
+ if (!workingBatch.empty()) {
+ sendRetryableWriteBatchRequestToConfig(
+ asr.opCtx(), nss, workingBatch, currentTxnNumber, writeConcern);
+ }
+}
+
StatusWith<bool> ShardingCatalogClientImpl::updateConfigDocument(
OperationContext* opCtx,
const NamespaceString& nss,
@@ -1117,10 +1150,11 @@ StatusWith<bool> ShardingCatalogClientImpl::_updateConfigDocument(
}()});
return updateOp;
}());
+ request.setWriteConcern(writeConcern.toBSON());
auto configShard = Grid::get(opCtx)->shardRegistry()->getConfigShard();
auto response = configShard->runBatchWriteCommand(
- opCtx, maxTimeMs, request, writeConcern, Shard::RetryPolicy::kIdempotent);
+ opCtx, maxTimeMs, request, Shard::RetryPolicy::kIdempotent);
Status status = response.toStatus();
if (!status.isOK()) {
@@ -1152,13 +1186,11 @@ Status ShardingCatalogClientImpl::removeConfigDocuments(OperationContext* opCtx,
}()});
return deleteOp;
}());
+ request.setWriteConcern(writeConcern.toBSON());
auto configShard = Grid::get(opCtx)->shardRegistry()->getConfigShard();
- auto response = configShard->runBatchWriteCommand(opCtx,
- Shard::kDefaultConfigCommandTimeout,
- request,
- writeConcern,
- Shard::RetryPolicy::kIdempotent);
+ auto response = configShard->runBatchWriteCommand(
+ opCtx, Shard::kDefaultConfigCommandTimeout, request, Shard::RetryPolicy::kIdempotent);
return response.toStatus();
}
@@ -1181,32 +1213,41 @@ StatusWith<repl::OpTimeWith<vector<BSONObj>>> ShardingCatalogClientImpl::_exhaus
response.getValue().opTime);
}
-StatusWith<std::vector<KeysCollectionDocument>> ShardingCatalogClientImpl::getNewInternalKeys(
+StatusWith<std::vector<KeysCollectionDocument>> ShardingCatalogClientImpl::getNewKeys(
OperationContext* opCtx,
StringData purpose,
const LogicalTime& newerThanThis,
repl::ReadConcernLevel readConcernLevel) {
- auto configShard = Grid::get(opCtx)->shardRegistry()->getConfigShard();
- return _getNewKeys<KeysCollectionDocument>(opCtx,
- configShard,
- NamespaceString::kKeysCollectionNamespace,
- purpose,
- newerThanThis,
- readConcernLevel);
-}
+ auto config = Grid::get(opCtx)->shardRegistry()->getConfigShard();
-StatusWith<std::vector<ExternalKeysCollectionDocument>>
-ShardingCatalogClientImpl::getAllExternalKeys(OperationContext* opCtx,
- StringData purpose,
- repl::ReadConcernLevel readConcernLevel) {
- auto configShard = Grid::get(opCtx)->shardRegistry()->getConfigShard();
- return _getNewKeys<ExternalKeysCollectionDocument>(
- opCtx,
- configShard,
- NamespaceString::kExternalKeysCollectionNamespace,
- purpose,
- LogicalTime(),
- readConcernLevel);
+ BSONObjBuilder queryBuilder;
+ queryBuilder.append("purpose", purpose);
+ queryBuilder.append("expiresAt", BSON("$gt" << newerThanThis.asTimestamp()));
+
+ auto findStatus = config->exhaustiveFindOnConfig(opCtx,
+ kConfigReadSelector,
+ readConcernLevel,
+ NamespaceString::kKeysCollectionNamespace,
+ queryBuilder.obj(),
+ BSON("expiresAt" << 1),
+ boost::none);
+
+ if (!findStatus.isOK()) {
+ return findStatus.getStatus();
+ }
+
+ const auto& keyDocs = findStatus.getValue().docs;
+ std::vector<KeysCollectionDocument> keys;
+ keys.reserve(keyDocs.size());
+ for (auto&& keyDoc : keyDocs) {
+ try {
+ keys.push_back(KeysCollectionDocument::parse(IDLParserErrorContext("keyDoc"), keyDoc));
+ } catch (...) {
+ return exceptionToStatus();
+ }
+ }
+
+ return keys;
}
} // namespace mongo
generated by cgit v1.2.3 (git 2.39.1) at 2026-09-11 13:38:00 +0000