From 1c390a0c50104a04cbd8ecbefb99eaf22e1bc914 Mon Sep 17 00:00:00 2001 From: Paolo Polato Date: Wed, 17 May 2023 08:05:19 +0000 Subject: SERVER-76908 Fix generation and format of the "createDatabase" op entries --- src/mongo/db/commands/notify_sharding_event.idl | 1 - .../db/repl/change_stream_oplog_notification.cpp | 25 +++++++++++++++------- ...harding_catalog_manager_database_operations.cpp | 18 +++++++++------- .../sharding_catalog_manager_shard_operations.cpp | 4 ++-- 4 files changed, 29 insertions(+), 19 deletions(-) (limited to 'src') diff --git a/src/mongo/db/commands/notify_sharding_event.idl b/src/mongo/db/commands/notify_sharding_event.idl index 19720db4eb2..f8459c58c29 100644 --- a/src/mongo/db/commands/notify_sharding_event.idl +++ b/src/mongo/db/commands/notify_sharding_event.idl @@ -75,7 +75,6 @@ structs: phase: description: "The sub-phase of the event being notified." type: CommitPhase - optional: true commands: _shardsvrNotifyShardingEvent: diff --git a/src/mongo/db/repl/change_stream_oplog_notification.cpp b/src/mongo/db/repl/change_stream_oplog_notification.cpp index 1c8fb6a78c2..0f8e1a2e0b2 100644 --- a/src/mongo/db/repl/change_stream_oplog_notification.cpp +++ b/src/mongo/db/repl/change_stream_oplog_notification.cpp @@ -102,20 +102,29 @@ void notifyChangeStreamsOnShardCollection(OperationContext* opCtx, void notifyChangeStreamsOnDatabaseAdded(OperationContext* opCtx, const DatabasesAdded& databasesAddedNotification) { - const auto& notifiedPhase = databasesAddedNotification.getPhase(); + const std::string operationName = [&] { + switch (databasesAddedNotification.getPhase()) { + case CommitPhaseEnum::kSuccessful: + return "createDatabase"; + case CommitPhaseEnum::kAborted: + return "createDatabaseAbort"; + case CommitPhaseEnum::kPrepare: + return "createDatabasePrepare"; + default: + MONGO_UNREACHABLE; + } + }(); + for (const auto& dbName : databasesAddedNotification.getNames()) { repl::MutableOplogEntry oplogEntry; oplogEntry.setOpType(repl::OpTypeEnum::kNoop); oplogEntry.setNss(NamespaceString(dbName)); oplogEntry.setTid(dbName.tenantId()); - oplogEntry.setObject(BSON("msg" << BSON("createDatabase" << dbName.db()))); + oplogEntry.setObject(BSON("msg" << BSON(operationName << dbName.db()))); BSONObjBuilder o2Builder; - o2Builder.append("createDatabase", dbName.db()); - if (notifiedPhase) { - o2Builder.append("phase", *notifiedPhase); - if (*notifiedPhase == CommitPhaseEnum::kPrepare) { - o2Builder.append("primaryShard", *databasesAddedNotification.getPrimaryShard()); - } + o2Builder.append(operationName, dbName.db()); + if (databasesAddedNotification.getPhase() == CommitPhaseEnum::kPrepare) { + o2Builder.append("primaryShard", *databasesAddedNotification.getPrimaryShard()); } o2Builder.append("isImported", databasesAddedNotification.getAreImported()); diff --git a/src/mongo/db/s/config/sharding_catalog_manager_database_operations.cpp b/src/mongo/db/s/config/sharding_catalog_manager_database_operations.cpp index 2f77383c745..19b2c07def0 100644 --- a/src/mongo/db/s/config/sharding_catalog_manager_database_operations.cpp +++ b/src/mongo/db/s/config/sharding_catalog_manager_database_operations.cpp @@ -233,13 +233,14 @@ DatabaseType ShardingCatalogManager::createDatabase( // - a "commitSuccessful" notification after completing the write into config.databases // will allow change streams to stop collecting events on the namespace created from // shards != resolvedPrimaryShard. + const auto allShards = Grid::get(opCtx)->shardRegistry()->getAllShardIds(opCtx); { DatabasesAdded prepareCommitEvent( - {DatabaseNameUtil::deserialize(boost::none, dbName)}, false /*areImported*/); - prepareCommitEvent.setPhase(CommitPhaseEnum::kPrepare); + {DatabaseNameUtil::deserialize(boost::none, dbName)}, + false /*areImported*/, + CommitPhaseEnum::kPrepare); prepareCommitEvent.setPrimaryShard(resolvedPrimaryShard->getId()); - uassertStatusOK(_notifyClusterOnNewDatabases( - opCtx, prepareCommitEvent, {resolvedPrimaryShard->getId()})); + uassertStatusOK(_notifyClusterOnNewDatabases(opCtx, prepareCommitEvent, allShards)); } const auto transactionChain = [db](const txn_api::TransactionClient& txnClient, @@ -279,10 +280,11 @@ DatabaseType ShardingCatalogManager::createDatabase( hangBeforeNotifyingCreateDatabaseCommitted.pauseWhileSet(); DatabasesAdded commitCompletedEvent( - {DatabaseNameUtil::deserialize(boost::none, dbName)}, false /*areImported*/); - commitCompletedEvent.setPhase(CommitPhaseEnum::kSuccessful); - const auto notificationOutcome = _notifyClusterOnNewDatabases( - opCtx, commitCompletedEvent, {resolvedPrimaryShard->getId()}); + {DatabaseNameUtil::deserialize(boost::none, dbName)}, + false /*areImported*/, + CommitPhaseEnum::kSuccessful); + const auto notificationOutcome = + _notifyClusterOnNewDatabases(opCtx, commitCompletedEvent, allShards); if (!notificationOutcome.isOK()) { LOGV2_WARNING(7175500, "Unable to send out notification of successful createDatabase", diff --git a/src/mongo/db/s/config/sharding_catalog_manager_shard_operations.cpp b/src/mongo/db/s/config/sharding_catalog_manager_shard_operations.cpp index d964c31d4c6..d1a2781012d 100644 --- a/src/mongo/db/s/config/sharding_catalog_manager_shard_operations.cpp +++ b/src/mongo/db/s/config/sharding_catalog_manager_shard_operations.cpp @@ -1436,8 +1436,8 @@ void ShardingCatalogManager::_addShardInTransaction( databasesInNewShard.end(), std::back_inserter(importedDbNames), [](const std::string& s) { return DatabaseNameUtil::deserialize(boost::none, s); }); - DatabasesAdded notification(std::move(importedDbNames), true /*addImported*/); - notification.setPhase(CommitPhaseEnum::kPrepare); + DatabasesAdded notification( + std::move(importedDbNames), true /*addImported*/, CommitPhaseEnum::kPrepare); notification.setPrimaryShard(ShardId(newShard.getName())); uassertStatusOK(_notifyClusterOnNewDatabases(opCtx, notification, existingShardIds)); -- cgit v1.2.1