From 066663949749e88d48b6f01ced8b28bb13150895 Mon Sep 17 00:00:00 2001 From: Ilya Repin Date: Sat, 31 Jan 2026 10:34:13 +0000 Subject: [PATCH 1/2] add leader service --- configs/static_config.yaml | 9 +- src/app/dto/leader/coordination.hpp | 17 ++ src/app/services/leader/leader_service.cpp | 24 +++ src/app/services/leader/leader_service.hpp | 30 ++++ .../leader/coordination/coordination.cpp | 43 ++++++ .../leader/coordination/coordination.hpp | 35 +++++ src/core/coordination/coordination_state.cpp | 4 +- src/core/coordination/coordination_state.hpp | 2 +- .../coordination/coordination_state_ut.cpp | 37 ++--- .../partition_balancing/balancing_impl.cpp | 43 ++++-- .../partition_balancing/balancing_impl.hpp | 10 +- .../load_factor_predictor.hpp | 9 +- .../partition_balancer.cpp | 4 +- .../partition_balancer.hpp | 4 +- .../tests/accumulate_migrating_weight_ut.cpp | 1 - .../tests/assign_orphaned_partitions_ut.cpp | 79 +++++++--- .../tests/build_coordination_context_ut.cpp | 35 ++--- .../tests/build_prediction_params_ut.cpp | 10 +- .../tests/collect_active_hubs_ut.cpp | 72 +++++---- .../tests/execute_rebalancing_phase_ut.cpp | 146 ++++++++++++------ .../tests/load_factor_predictor_mock.hpp | 4 +- .../tests/rebalance_partitions_ut.cpp | 62 ++++---- .../tests/separate_partitions_ut.cpp | 24 ++- .../tests/test_helpers_ut.cpp | 8 +- .../tests/test_helpers_ut.hpp | 4 +- src/hello.cpp | 13 +- src/hello.hpp | 5 +- src/infra/components/components.cpp | 15 +- src/infra/components/components.hpp | 3 + .../coordination_dist_lock_component.cpp | 52 ------- .../components/hub/hub_gateway_component.cpp | 30 ++++ .../components/hub/hub_gateway_component.hpp | 31 ++++ .../leader/leader_dist_lock_component.cpp | 72 +++++++++ .../leader_dist_lock_component.hpp} | 12 +- .../leader/leader_service_component.cpp | 41 +++++ .../leader/leader_service_component.hpp | 31 ++++ .../load_factor_predictor_component.cpp | 44 ++++++ .../load_factor_predictor_component.hpp | 33 ++++ .../kesus_coordination_gateway.cpp | 1 + .../heuristic_predictor.cpp | 54 +++++++ .../heuristic_predictor.hpp | 22 +++ .../heuristic_predictor_ut.cpp | 131 ++++++++++++++++ src/infra/serializer/serializer_ut.cpp | 8 + src/main.cpp | 1 + 44 files changed, 1013 insertions(+), 302 deletions(-) create mode 100644 src/app/dto/leader/coordination.hpp create mode 100644 src/app/services/leader/leader_service.cpp create mode 100644 src/app/services/leader/leader_service.hpp create mode 100644 src/app/use_cases/leader/coordination/coordination.cpp create mode 100644 src/app/use_cases/leader/coordination/coordination.hpp delete mode 100644 src/infra/components/coordination/coordination_dist_lock_component.cpp create mode 100644 src/infra/components/hub/hub_gateway_component.cpp create mode 100644 src/infra/components/hub/hub_gateway_component.hpp create mode 100644 src/infra/components/leader/leader_dist_lock_component.cpp rename src/infra/components/{coordination/coordination_dist_lock_component.hpp => leader/leader_dist_lock_component.hpp} (63%) create mode 100644 src/infra/components/leader/leader_service_component.cpp create mode 100644 src/infra/components/leader/leader_service_component.hpp create mode 100644 src/infra/components/partition_balancing/load_factor_predictor_component.cpp create mode 100644 src/infra/components/partition_balancing/load_factor_predictor_component.hpp create mode 100644 src/infra/load_factor_predictor/heuristic_predictor.cpp create mode 100644 src/infra/load_factor_predictor/heuristic_predictor.hpp create mode 100644 src/infra/load_factor_predictor/heuristic_predictor_ut.cpp diff --git a/configs/static_config.yaml b/configs/static_config.yaml index 8dcf650..3a0ab13 100644 --- a/configs/static_config.yaml +++ b/configs/static_config.yaml @@ -73,12 +73,19 @@ components_manager: discovery-semaphore: discovery-lock initial-setup: true - coordination-dist-lock: + hub-gateway: + + load-factor-predictor: + type: heuristic + + leader-dist-lock: database-settings: dbname: chat_db coordination-node: chat-coordination semaphore-name: partition-map-lock + leader-service: + handler-ping: path: /ping method: GET diff --git a/src/app/dto/leader/coordination.hpp b/src/app/dto/leader/coordination.hpp new file mode 100644 index 0000000..0502f54 --- /dev/null +++ b/src/app/dto/leader/coordination.hpp @@ -0,0 +1,17 @@ +#pragma once + +#include +#include + +namespace NCoordinator::NApp::NDto { + +//////////////////////////////////////////////////////////////////////////////// + +struct TCoordinationRequest { + NCore::NDomain::TStateBuildingSettings StateBuildingSettings; + NCore::TBalancingSettings BalancingSettings; +}; + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NCoordinator::NApp::NDto diff --git a/src/app/services/leader/leader_service.cpp b/src/app/services/leader/leader_service.cpp new file mode 100644 index 0000000..b8ad600 --- /dev/null +++ b/src/app/services/leader/leader_service.cpp @@ -0,0 +1,24 @@ +#include "leader_service.hpp" + +#include + +namespace NCoordinator::NApp::NService { + +//////////////////////////////////////////////////////////////////////////////// + +TLeaderService::TLeaderService( + NCore::NDomain::ICoordinationGateway& coordinationGateway_, + NCore::NDomain::ICoordinationRepository& coordinationRepository_, + NCore::NDomain::IHubGateway& hubGateway, + NCore::ILoadFactorPredictor& loadFactorPredictor) + : CoordinationUseCase_(coordinationGateway_, coordinationRepository_, hubGateway, loadFactorPredictor) +{ } + +void TLeaderService::Coordinate(const NDto::TCoordinationRequest& request) const +{ + CoordinationUseCase_.Execute(request); +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NCoordinator::NApp::NService diff --git a/src/app/services/leader/leader_service.hpp b/src/app/services/leader/leader_service.hpp new file mode 100644 index 0000000..c21c55e --- /dev/null +++ b/src/app/services/leader/leader_service.hpp @@ -0,0 +1,30 @@ +#pragma once + +#include +#include +#include +#include +#include + +namespace NCoordinator::NApp::NService { + +//////////////////////////////////////////////////////////////////////////////// + +class TLeaderService final +{ +public: + TLeaderService( + NCore::NDomain::ICoordinationGateway& coordinationGateway_, + NCore::NDomain::ICoordinationRepository& coordinationRepository_, + NCore::NDomain::IHubGateway& hubGateway, + NCore::ILoadFactorPredictor& loadFactorPredictor); + + void Coordinate(const NDto::TCoordinationRequest& request) const; + +private: + NUseCase::TCoordinationUseCase CoordinationUseCase_; +}; + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NCoordinator::NApp::NService diff --git a/src/app/use_cases/leader/coordination/coordination.cpp b/src/app/use_cases/leader/coordination/coordination.cpp new file mode 100644 index 0000000..a1cbca7 --- /dev/null +++ b/src/app/use_cases/leader/coordination/coordination.cpp @@ -0,0 +1,43 @@ +#include "coordination.hpp" + +#include + +namespace NCoordinator::NApp::NUseCase { + +//////////////////////////////////////////////////////////////////////////////// + +TCoordinationUseCase::TCoordinationUseCase( + NCore::NDomain::ICoordinationGateway& coordinationGateway_, + NCore::NDomain::ICoordinationRepository& coordinationRepository_, + NCore::NDomain::IHubGateway& hubGateway, + NCore::ILoadFactorPredictor& loadFactorPredictor) + : CoordinationGateway_(coordinationGateway_) + , CoordinationRepository_(coordinationRepository_) + , HubGateway_(hubGateway) + , Balancer_(loadFactorPredictor) +{ } + +void TCoordinationUseCase::Execute(const NDto::TCoordinationRequest& request) const +{ + auto hubDiscovery = CoordinationGateway_.GetHubDiscovery(); + auto hubReports = HubGateway_.GetHubReports(hubDiscovery); + + auto partitionMap = CoordinationGateway_.GetPartitionMap().value(); + + auto coordinationContext = CoordinationRepository_.GetCoordinationContext(); + + auto coordinationState = NCore::NDomain::TCoordinationState( + partitionMap, + hubReports, + coordinationContext, + request.StateBuildingSettings); + + auto balancingResult = Balancer_.BalancePartitions(coordinationState, request.BalancingSettings); + + CoordinationGateway_.BroadcastPartitionMap(balancingResult.PartitionMap); + CoordinationRepository_.SetCoordinationContext(balancingResult.Context); +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NCoordinator::NApp::NUseCase diff --git a/src/app/use_cases/leader/coordination/coordination.hpp b/src/app/use_cases/leader/coordination/coordination.hpp new file mode 100644 index 0000000..63b02d7 --- /dev/null +++ b/src/app/use_cases/leader/coordination/coordination.hpp @@ -0,0 +1,35 @@ +#pragma once + +#include +#include +#include +#include +#include +#include + +namespace NCoordinator::NApp::NUseCase { + +//////////////////////////////////////////////////////////////////////////////// + +class TCoordinationUseCase final +{ +public: + TCoordinationUseCase( + NCore::NDomain::ICoordinationGateway& coordinationGateway_, + NCore::NDomain::ICoordinationRepository& coordinationRepository_, + NCore::NDomain::IHubGateway& hubGateway, + NCore::ILoadFactorPredictor& loadFactorPredictor); + + void Execute(const NDto::TCoordinationRequest& request) const; + +private: + NCore::NDomain::ICoordinationGateway& CoordinationGateway_; + NCore::NDomain::ICoordinationRepository& CoordinationRepository_; + NCore::NDomain::IHubGateway& HubGateway_; + + NCore::TPartitionBalancer Balancer_; +}; + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NCoordinator::NApp::NUseCase diff --git a/src/core/coordination/coordination_state.cpp b/src/core/coordination/coordination_state.cpp index 4ab25ba..0876471 100644 --- a/src/core/coordination/coordination_state.cpp +++ b/src/core/coordination/coordination_state.cpp @@ -91,8 +91,8 @@ void TCoordinationState::ApplyClusterSnapshot( { std::unordered_map expectedWeightGrowths; - for (const auto& [hub, report] : snapshot) { - THubState& state = HubStates_[hub]; + for (const auto& report : snapshot) { + THubState& state = HubStates_[report.Endpoint]; state.Endpoint = report.Endpoint; state.DC = report.DC; diff --git a/src/core/coordination/coordination_state.hpp b/src/core/coordination/coordination_state.hpp index 48af823..0d82027 100644 --- a/src/core/coordination/coordination_state.hpp +++ b/src/core/coordination/coordination_state.hpp @@ -19,7 +19,7 @@ namespace NCoordinator::NCore::NDomain { class TCoordinationState { public: - using TClusterSnapshot = std::unordered_map; + using TClusterSnapshot = std::vector; using TPartitionStates = std::unordered_map; using THubStates = std::unordered_map; diff --git a/src/core/coordination/coordination_state_ut.cpp b/src/core/coordination/coordination_state_ut.cpp index de08715..165df65 100644 --- a/src/core/coordination/coordination_state_ut.cpp +++ b/src/core/coordination/coordination_state_ut.cpp @@ -124,8 +124,7 @@ TEST(TCoordinationState, CalculateAveragePartitionWeightFromContext) TEST(TCoordinationState, CalculatesAveragePartitionWeightFromSnapshot) { TCoordinationState::TClusterSnapshot snapshot; - snapshot.emplace( - THubEndpoint("hub-a"), + snapshot.emplace_back( MakeHubReport( "hub-a", "myt", @@ -154,8 +153,7 @@ TEST(TCoordinationState, HandlesEmptySnapshot) { TEST(TCoordinationState, BuildsHubStatesFromSnapshot) { TCoordinationState::TClusterSnapshot snapshot; - snapshot.emplace( - THubEndpoint("hub-a"), + snapshot.emplace_back( MakeHubReport( "hub-a", "myt", @@ -185,8 +183,7 @@ TEST(TCoordinationState, BuildsHubStatesFromSnapshot) TEST(TCoordinationState, ComputesExpectedWeightGrowth) { TCoordinationState::TClusterSnapshot snapshot; - snapshot.emplace( - THubEndpoint("hub-a"), + snapshot.emplace_back( MakeHubReport( "hub-a", "myt", @@ -216,9 +213,7 @@ TEST(TCoordinationState, HubStatusDrainingBySettings) settings.BlockedHubs.insert(THubEndpoint("hub-a")); TCoordinationState::TClusterSnapshot snapshot; - snapshot.emplace( - THubEndpoint("hub-a"), - MakeHubReport("hub-a", "myt", 42, 10, {})); + snapshot.emplace_back(MakeHubReport("hub-a", "myt", 42, 10, {})); TCoordinationState state( MakePartitionMap(), @@ -234,9 +229,7 @@ TEST(TCoordinationState, HubStatusDrainingBySettings) TEST(TCoordinationState, HubStatusLagged) { TCoordinationState::TClusterSnapshot snapshot; - snapshot.emplace( - THubEndpoint("hub-a"), - MakeHubReport("hub-a", "myt", 41, 10, {})); + snapshot.emplace_back(MakeHubReport("hub-a", "myt", 41, 10, {})); TCoordinationState state( MakePartitionMap(), @@ -252,9 +245,7 @@ TEST(TCoordinationState, HubStatusLagged) TEST(TCoordinationState, HubStatusOverloaded) { TCoordinationState::TClusterSnapshot snapshot; - snapshot.emplace( - THubEndpoint("hub-a"), - MakeHubReport("hub-a", "myt", 42, 95, {})); + snapshot.emplace_back(MakeHubReport("hub-a", "myt", 42, 95, {})); TCoordinationState state( MakePartitionMap(), @@ -273,9 +264,7 @@ TEST(TCoordinationState, HubStatusDrainingPriority) settings.BlockedHubs.insert(THubEndpoint("hub-a")); TCoordinationState::TClusterSnapshot snapshot; - snapshot.emplace( - THubEndpoint("hub-a"), - MakeHubReport("hub-a", "myt", 42, 95, {})); + snapshot.emplace_back(MakeHubReport("hub-a", "myt", 42, 95, {})); TCoordinationState state( MakePartitionMap(), @@ -294,15 +283,9 @@ TEST(TCoordinationState, DrainingHubsByBlockedDC) settings.BlockedDCs.insert(THubDC("myt")); TCoordinationState::TClusterSnapshot snapshot; - snapshot.emplace( - THubEndpoint("hub-a"), - MakeHubReport("hub-a", "myt", 42, 10, {})); - snapshot.emplace( - THubEndpoint("hub-b"), - MakeHubReport("hub-b", "myt", 42, 35, {})); - snapshot.emplace( - THubEndpoint("hub-c"), - MakeHubReport("hub-c", "sas", 42, 24, {})); + snapshot.emplace_back(MakeHubReport("hub-a", "myt", 42, 10, {})); + snapshot.emplace_back(MakeHubReport("hub-b", "myt", 42, 35, {})); + snapshot.emplace_back(MakeHubReport("hub-c", "sas", 42, 24, {})); TCoordinationState state( MakePartitionMap(), diff --git a/src/core/partition_balancing/balancing_impl.cpp b/src/core/partition_balancing/balancing_impl.cpp index 207a3fa..71ed802 100644 --- a/src/core/partition_balancing/balancing_impl.cpp +++ b/src/core/partition_balancing/balancing_impl.cpp @@ -10,7 +10,7 @@ namespace NCoordinator::NCore::NDetail { std::pair CollectActiveHubs( const NDomain::TCoordinationState& state, - const TLoadFactorPredictorPtr& loadFactorPredictor) + const ILoadFactorPredictor& loadFactorPredictor) { TSortedHubs sortedHubs; THubEndpoints activeHubs; @@ -25,16 +25,15 @@ std::pair CollectActiveHubs( activeHubs.emplace(endpoint); TPredictionParams predictionParams { + .LoadFactor = hubState.LoadFactor, + .PartitionWeight = hubState.ExpectedWeightGrowth, .Increasing = true, .TotalPartitions = hubState.TotalPartitions, .PartitionsWeight = hubState.PartitionsWeight, .OriginalLoadFactor = hubState.LoadFactor, }; - NDomain::TLoadFactor forecastedLoad = loadFactorPredictor->PredictLoadFactor( - hubState.LoadFactor, - hubState.ExpectedWeightGrowth, - predictionParams); + NDomain::TLoadFactor forecastedLoad = loadFactorPredictor.PredictLoadFactor(predictionParams); sortedHubs.emplace(forecastedLoad, endpoint); } @@ -70,7 +69,7 @@ TSeparatedPartitions SeparatePartitions( TMigratingWeight AssignOrphanedPartitions( const NDomain::TCoordinationState& state, const TWeightedPartitions& orphanedPartitions, - const TLoadFactorPredictorPtr& loadFactorPredictor, + const ILoadFactorPredictor& loadFactorPredictor, TSortedHubs& sortedHubs, TAssignedPartitions& assignedPartitions) { @@ -90,12 +89,14 @@ TMigratingWeight AssignOrphanedPartitions( assignedPartitions.emplace_back(partitionId, hub); TPredictionParams params { + .LoadFactor = loadFactor, + .PartitionWeight = partitionWeight, .Increasing = true, .TotalPartitions = hubState.TotalPartitions + migratingWeight[hub].first, .PartitionsWeight = hubState.PartitionsWeight + migratingWeight[hub].second, .OriginalLoadFactor = hubState.LoadFactor, }; - loadFactor = loadFactorPredictor->PredictLoadFactor(loadFactor, partitionWeight, params); + loadFactor = loadFactorPredictor.PredictLoadFactor(params); migratingWeight[hub].first++; migratingWeight[hub].second += partitionWeight; @@ -122,7 +123,7 @@ void RebalancePartitions( TSortedHubs& sortedHubs, TAssignedPartitions& assignedPartitions, TMigrationContext& migrationContext, - const TLoadFactorPredictorPtr& loadFactorPredictor, + const ILoadFactorPredictor& loadFactorPredictor, const NDomain::TCoordinationState& state, const TBalancingSettings& settings) { @@ -174,7 +175,7 @@ void ExecuteRebalancingPhase( TSortedHubs& sortedHubs, THubPartitions& hubPartitions, TMigrationContext& migrationContext, - const TLoadFactorPredictorPtr& loadFactorPredictor, + const ILoadFactorPredictor& loadFactorPredictor, const NDomain::TCoordinationState& state, const TBalancingSettings& settings) { @@ -212,11 +213,23 @@ void ExecuteRebalancingPhase( } } - auto maxParams = BuildPredictionParams(false, hubPartitions[maxHub], state.GetHubState(maxHub)); - auto nextMaxLoadFactor = loadFactorPredictor->PredictLoadFactor(maxLoadFactor, partitionWeight, maxParams); + auto maxParams = BuildPredictionParams( + maxLoadFactor, + partitionWeight, + false, + hubPartitions[maxHub], + state.GetHubState(maxHub)); + + auto nextMaxLoadFactor = loadFactorPredictor.PredictLoadFactor(maxParams); - auto minParams = BuildPredictionParams(true, hubPartitions[minHub], state.GetHubState(minHub)); - auto nextMinLoadFactor = loadFactorPredictor->PredictLoadFactor(minLoadFactor, partitionWeight, minParams); + auto minParams = BuildPredictionParams( + minLoadFactor, + partitionWeight, + true, + hubPartitions[minHub], + state.GetHubState(minHub)); + + auto nextMinLoadFactor = loadFactorPredictor.PredictLoadFactor(minParams); auto currentDelta = maxLoadFactor - minLoadFactor; auto nextDelta = nextMaxLoadFactor - nextMinLoadFactor; @@ -250,11 +263,15 @@ void ExecuteRebalancingPhase( } TPredictionParams BuildPredictionParams( + const NDomain::TLoadFactor loadFactor, + const NDomain::TPartitionWeight partitionWeight, const bool increasing, const std::set& partitions, const NDomain::THubState& state) { return TPredictionParams{ + .LoadFactor = loadFactor, + .PartitionWeight = partitionWeight, .Increasing = increasing, .TotalPartitions = partitions.size(), .PartitionsWeight = diff --git a/src/core/partition_balancing/balancing_impl.hpp b/src/core/partition_balancing/balancing_impl.hpp index a12d3e6..95a4018 100644 --- a/src/core/partition_balancing/balancing_impl.hpp +++ b/src/core/partition_balancing/balancing_impl.hpp @@ -38,14 +38,14 @@ struct TMigrationContext { std::pair CollectActiveHubs( const NDomain::TCoordinationState& state, - const TLoadFactorPredictorPtr& loadFactorPredictor); + const ILoadFactorPredictor& loadFactorPredictor); TSeparatedPartitions SeparatePartitions(const NDomain::TCoordinationState& state, const THubEndpoints& activeHubs); TMigratingWeight AssignOrphanedPartitions( const NDomain::TCoordinationState& state, const TWeightedPartitions& orphanedPartitions, - const TLoadFactorPredictorPtr& loadFactorPredictor, + const ILoadFactorPredictor& loadFactorPredictor, TSortedHubs& sortedHubs, TAssignedPartitions& assignedPartitions); @@ -56,7 +56,7 @@ void RebalancePartitions( TSortedHubs& sortedHubs, TAssignedPartitions& assignedPartitions, TMigrationContext& migrationContext, - const TLoadFactorPredictorPtr& loadFactorPredictor, + const ILoadFactorPredictor& loadFactorPredictor, const NDomain::TCoordinationState& state, const TBalancingSettings& settings); @@ -64,11 +64,13 @@ void ExecuteRebalancingPhase( TSortedHubs& sortedHubs, THubPartitions& hubPartitions, TMigrationContext& migrationContext, - const TLoadFactorPredictorPtr& loadFactorPredictor, + const ILoadFactorPredictor& loadFactorPredictor, const NDomain::TCoordinationState& state, const TBalancingSettings& settings); TPredictionParams BuildPredictionParams( + const NDomain::TLoadFactor loadFactor, + const NDomain::TPartitionWeight partitionWeight, const bool increasing, const std::set& partitions, const NDomain::THubState& state); diff --git a/src/core/partition_balancing/load_factor_predictor.hpp b/src/core/partition_balancing/load_factor_predictor.hpp index 7e559df..f29e9ee 100644 --- a/src/core/partition_balancing/load_factor_predictor.hpp +++ b/src/core/partition_balancing/load_factor_predictor.hpp @@ -9,6 +9,8 @@ namespace NCoordinator::NCore { //////////////////////////////////////////////////////////////////////////////// struct TPredictionParams { + NDomain::TLoadFactor LoadFactor; + NDomain::TPartitionWeight PartitionWeight; bool Increasing{}; std::size_t TotalPartitions{}; NDomain::TPartitionWeight PartitionsWeight; @@ -17,16 +19,11 @@ struct TPredictionParams { class ILoadFactorPredictor { public: - virtual NDomain::TLoadFactor PredictLoadFactor( - const NDomain::TLoadFactor loadFactor, - const NDomain::TPartitionWeight partitionWeight, - const TPredictionParams& params) const = 0; + virtual NDomain::TLoadFactor PredictLoadFactor(const TPredictionParams& params) const = 0; virtual ~ILoadFactorPredictor() = default; }; -using TLoadFactorPredictorPtr = std::shared_ptr; - //////////////////////////////////////////////////////////////////////////////// } // namespace NCoordinator::NCore diff --git a/src/core/partition_balancing/partition_balancer.cpp b/src/core/partition_balancing/partition_balancer.cpp index a0a77bb..dbe88dc 100644 --- a/src/core/partition_balancing/partition_balancer.cpp +++ b/src/core/partition_balancing/partition_balancer.cpp @@ -6,8 +6,8 @@ namespace NCoordinator::NCore { //////////////////////////////////////////////////////////////////////////////// -TPartitionBalancer::TPartitionBalancer(TLoadFactorPredictorPtr predictor) - : LoadFactorPredictor_(std::move(predictor)) +TPartitionBalancer::TPartitionBalancer(ILoadFactorPredictor& predictor) + : LoadFactorPredictor_(predictor) { } TBalancingResult TPartitionBalancer::BalancePartitions( diff --git a/src/core/partition_balancing/partition_balancer.hpp b/src/core/partition_balancing/partition_balancer.hpp index 80d4869..00d49de 100644 --- a/src/core/partition_balancing/partition_balancer.hpp +++ b/src/core/partition_balancing/partition_balancer.hpp @@ -21,14 +21,14 @@ struct TBalancingResult { class TPartitionBalancer { public: - explicit TPartitionBalancer(TLoadFactorPredictorPtr predictor); + explicit TPartitionBalancer(ILoadFactorPredictor& predictor); TBalancingResult BalancePartitions( const NDomain::TCoordinationState& state, const TBalancingSettings& settings) const; private: - TLoadFactorPredictorPtr LoadFactorPredictor_; + ILoadFactorPredictor& LoadFactorPredictor_; }; //////////////////////////////////////////////////////////////////////////////// diff --git a/src/core/partition_balancing/tests/accumulate_migrating_weight_ut.cpp b/src/core/partition_balancing/tests/accumulate_migrating_weight_ut.cpp index 945fb3b..3532903 100644 --- a/src/core/partition_balancing/tests/accumulate_migrating_weight_ut.cpp +++ b/src/core/partition_balancing/tests/accumulate_migrating_weight_ut.cpp @@ -1,7 +1,6 @@ #include #include "test_helpers_ut.hpp" -#include "load_factor_predictor_mock.hpp" #include diff --git a/src/core/partition_balancing/tests/assign_orphaned_partitions_ut.cpp b/src/core/partition_balancing/tests/assign_orphaned_partitions_ut.cpp index 7e2e9ed..9ab08c0 100644 --- a/src/core/partition_balancing/tests/assign_orphaned_partitions_ut.cpp +++ b/src/core/partition_balancing/tests/assign_orphaned_partitions_ut.cpp @@ -21,9 +21,7 @@ class AssignOrphanedPartitionsTest { TCoordinationState::TClusterSnapshot snapshot; for (const auto& [name, load] : hubsInfo) { - snapshot.emplace(HUB(name), THubReport{ - EP(42), HUB(name), DC("myt"), load, {} - }); + snapshot.emplace_back(EP(42), HUB(name), DC("myt"), load, PWS({})); } TPartitionMap map{ @@ -54,14 +52,28 @@ TEST_F(AssignOrphanedPartitionsTest, DistributesLoadBetweenHubs) { TAssignedPartitions assignedPartitions; - EXPECT_CALL(*Predictor_, PredictLoadFactor(LF(10), PW(100), _)) + EXPECT_CALL(*Predictor_, + PredictLoadFactor(AllOf( + Field(&TPredictionParams::LoadFactor, LF(10)), + Field(&TPredictionParams::PartitionWeight, PW(100)), + Field(&TPredictionParams::Increasing, true), + Field(&TPredictionParams::TotalPartitions, 0), + Field(&TPredictionParams::PartitionsWeight, PW(0)), + Field(&TPredictionParams::OriginalLoadFactor, LF(10))))) .WillOnce(Return(LF(30))); - EXPECT_CALL(*Predictor_, PredictLoadFactor(LF(30), PW(50), _)) + EXPECT_CALL(*Predictor_, + PredictLoadFactor(AllOf( + Field(&TPredictionParams::LoadFactor, LF(30)), + Field(&TPredictionParams::PartitionWeight, PW(50)), + Field(&TPredictionParams::Increasing, true), + Field(&TPredictionParams::TotalPartitions, 1), + Field(&TPredictionParams::PartitionsWeight, PW(100)), + Field(&TPredictionParams::OriginalLoadFactor, LF(10))))) .WillOnce(Return(LF(35))); auto migratingWeight = AssignOrphanedPartitions( - state, orphaned, Predictor_, sortedHubs, assignedPartitions + state, orphaned, *Predictor_, sortedHubs, assignedPartitions ); ASSERT_EQ(assignedPartitions.size(), 2); @@ -95,25 +107,30 @@ TEST_F(AssignOrphanedPartitionsTest, CorrectlyUpdatesPredictorParams) { testing::InSequence s; // Params: Total=0, Weight=0 - EXPECT_CALL(*Predictor_, PredictLoadFactor(LF(0), PW(100), - AllOf( - Field(&TPredictionParams::TotalPartitions, 0), - Field(&TPredictionParams::PartitionsWeight, PW(0)) - ) - )).WillOnce(Return(LF(10))); - + EXPECT_CALL(*Predictor_, + PredictLoadFactor(AllOf( + Field(&TPredictionParams::LoadFactor, LF(0)), + Field(&TPredictionParams::PartitionWeight, PW(100)), + Field(&TPredictionParams::Increasing, true), + Field(&TPredictionParams::TotalPartitions, 0), + Field(&TPredictionParams::PartitionsWeight, PW(0)), + Field(&TPredictionParams::OriginalLoadFactor, LF(0))))) + .WillOnce(Return(LF(10))); // Params: Total=1, Weight=100 - EXPECT_CALL(*Predictor_, PredictLoadFactor(LF(10), PW(200), - AllOf( + EXPECT_CALL(*Predictor_, + PredictLoadFactor(AllOf( + Field(&TPredictionParams::LoadFactor, LF(10)), + Field(&TPredictionParams::PartitionWeight, PW(200)), + Field(&TPredictionParams::Increasing, true), Field(&TPredictionParams::TotalPartitions, 1), - Field(&TPredictionParams::PartitionsWeight, PW(100)) - ) - )).WillOnce(Return(LF(30))); + Field(&TPredictionParams::PartitionsWeight, PW(100)), + Field(&TPredictionParams::OriginalLoadFactor, LF(0))))) + .WillOnce(Return(LF(30))); } auto migratingWeight = AssignOrphanedPartitions( - state, orphaned, Predictor_, sortedHubs, assignedPartitions + state, orphaned, *Predictor_, sortedHubs, assignedPartitions ); ASSERT_EQ(migratingWeight.size(), 1); @@ -128,10 +145,10 @@ TEST_F(AssignOrphanedPartitionsTest, StopsWhenNoHubsAvailable) { TWeightedPartitions orphaned = {{PW(100), PID(1)}}; TAssignedPartitions assignedPartitions; - EXPECT_CALL(*Predictor_, PredictLoadFactor(_, _, _)).Times(0); + EXPECT_CALL(*Predictor_, PredictLoadFactor(_)).Times(0); auto migratingWeight = AssignOrphanedPartitions( - state, orphaned, Predictor_, sortedHubs, assignedPartitions + state, orphaned, *Predictor_, sortedHubs, assignedPartitions ); EXPECT_TRUE(assignedPartitions.empty()); @@ -155,15 +172,29 @@ TEST_F(AssignOrphanedPartitionsTest, ConsidersSortOrderAfterUpdate) { TAssignedPartitions assignedPartitions; // 1. Hub-A (10) < Hub-B (15). - EXPECT_CALL(*Predictor_, PredictLoadFactor(LF(10), PW(100), _)) + EXPECT_CALL(*Predictor_, + PredictLoadFactor(AllOf( + Field(&TPredictionParams::LoadFactor, LF(10)), + Field(&TPredictionParams::PartitionWeight, PW(100)), + Field(&TPredictionParams::Increasing, true), + Field(&TPredictionParams::TotalPartitions, 0), + Field(&TPredictionParams::PartitionsWeight, PW(0)), + Field(&TPredictionParams::OriginalLoadFactor, LF(10))))) .WillOnce(Return(LF(20))); // 2. sortedHubs: Hub-B (15), Hub-A (20) - EXPECT_CALL(*Predictor_, PredictLoadFactor(LF(15), PW(100), _)) + EXPECT_CALL(*Predictor_, + PredictLoadFactor(AllOf( + Field(&TPredictionParams::LoadFactor, LF(15)), + Field(&TPredictionParams::PartitionWeight, PW(100)), + Field(&TPredictionParams::Increasing, true), + Field(&TPredictionParams::TotalPartitions, 0), + Field(&TPredictionParams::PartitionsWeight, PW(0)), + Field(&TPredictionParams::OriginalLoadFactor, LF(15))))) .WillOnce(Return(LF(25))); AssignOrphanedPartitions( - state, orphaned, Predictor_, sortedHubs, assignedPartitions + state, orphaned, *Predictor_, sortedHubs, assignedPartitions ); ASSERT_EQ(assignedPartitions.size(), 2); diff --git a/src/core/partition_balancing/tests/build_coordination_context_ut.cpp b/src/core/partition_balancing/tests/build_coordination_context_ut.cpp index be7d2e9..f5b51a8 100644 --- a/src/core/partition_balancing/tests/build_coordination_context_ut.cpp +++ b/src/core/partition_balancing/tests/build_coordination_context_ut.cpp @@ -20,15 +20,15 @@ class BuildCoordinationContextTest TEST_F(BuildCoordinationContextTest, BuildsCoordinationContext) { TCoordinationState::TClusterSnapshot snapshot; { - snapshot.emplace(HUB("hub-active"), THubReport{ - EP(42), HUB("hub-active"), DC("myt"), LF(10), { + snapshot.emplace_back( + EP(42), HUB("hub-active"), DC("myt"), LF(10), PWS({ {PID(1), PW(50)}, - }}); + })); - snapshot.emplace(HUB("hub-draining"), THubReport{ - EP(42), HUB("hub-draining"), DC("myt"), LF(10), { + snapshot.emplace_back( + EP(42), HUB("hub-draining"), DC("myt"), LF(10), PWS({ {PID(2), PW(220)}, - }}); + })); } TPartitionMap map{ @@ -80,11 +80,11 @@ TEST_F(BuildCoordinationContextTest, BuildsCoordinationContext) { TEST_F(BuildCoordinationContextTest, UsesAverageWeightIfObservedMissing) { TCoordinationState::TClusterSnapshot snapshot; { - snapshot.emplace(HUB("hub-1"), THubReport{ - EP(10), HUB("hub-1"), DC("myt"), LF(50), { + snapshot.emplace_back( + EP(10), HUB("hub-1"), DC("myt"), LF(50), PWS({ {PID(1), PW(100)}, {PID(2), PW(200)}, - }}); + })); } TPartitionMap map{ @@ -129,7 +129,7 @@ TEST_F(BuildCoordinationContextTest, FiltersOutdatedCooldowns) { }; TCoordinationState::TClusterSnapshot snapshot; - snapshot.emplace(HUB("h"), THubReport{EP(100), HUB("h"), DC("dc"), LF(0), {}}); + snapshot.emplace_back(EP(100), HUB("h"), DC("dc"), LF(0), PWS({})); TCoordinationContext context{ .PartitionCooldowns{ @@ -154,9 +154,7 @@ TEST_F(BuildCoordinationContextTest, FiltersOutdatedCooldowns) { TEST_F(BuildCoordinationContextTest, MigrationOverwritesExistingCooldown) { TCoordinationState::TClusterSnapshot snapshot; - snapshot.emplace(HUB("h"), THubReport{ - EP(10), HUB("h"), DC("dc"), LF(0), {{PID(1), PW(100)}} - }); + snapshot.emplace_back(EP(10), HUB("h"), DC("dc"), LF(0), PWS({{PID(1), PW(100)}})); TPartitionMap map{ .Partitions{{PID(1), HUB("h")}}, @@ -187,11 +185,10 @@ TEST_F(BuildCoordinationContextTest, MigrationOverwritesExistingCooldown) { TEST_F(BuildCoordinationContextTest, UpdatesObservedWeightsInContext) { TCoordinationState::TClusterSnapshot snapshot; - snapshot.emplace(HUB("h"), THubReport{ - EP(10), HUB("h"), DC("dc"), LF(0), { + snapshot.emplace_back( + EP(10), HUB("h"), DC("dc"), LF(0), PWS({ {PID(1), PW(500)}, - } - }); + })); TPartitionMap map{ .Partitions{{PID(1), HUB("h")}}, @@ -216,9 +213,7 @@ TEST_F(BuildCoordinationContextTest, UpdatesObservedWeightsInContext) { TEST_F(BuildCoordinationContextTest, HandlesExpectedWeightGrowth) { TCoordinationState::TClusterSnapshot snapshot; - snapshot.emplace(HUB("h"), THubReport{ - EP(10), HUB("h"), DC("dc"), LF(0), {{PID(1), PW(100)}} - }); + snapshot.emplace_back(EP(10), HUB("h"), DC("dc"), LF(0), PWS({{PID(1), PW(100)}})); TPartitionMap map{ .Partitions{{PID(1), HUB("h")}}, diff --git a/src/core/partition_balancing/tests/build_prediction_params_ut.cpp b/src/core/partition_balancing/tests/build_prediction_params_ut.cpp index cd11044..c39af0e 100644 --- a/src/core/partition_balancing/tests/build_prediction_params_ut.cpp +++ b/src/core/partition_balancing/tests/build_prediction_params_ut.cpp @@ -33,8 +33,10 @@ TEST_F(BuildPredictionParamsTest, CorrectlyBuildsParams) { THubState state; state.LoadFactor = LF(35); - auto params = BuildPredictionParams(true, partitions, state); - + auto params = BuildPredictionParams(LF(15), PW(101), true, partitions, state); + + EXPECT_EQ(params.LoadFactor, LF(15)); + EXPECT_EQ(params.PartitionWeight, PW(101)); EXPECT_EQ(params.Increasing, true); EXPECT_EQ(params.TotalPartitions, 8); EXPECT_EQ(params.PartitionsWeight, PW(3082)); @@ -46,8 +48,10 @@ TEST_F(BuildPredictionParamsTest, HandlesEmptyPartitions) { THubState state; state.LoadFactor = LF(5); - auto params = BuildPredictionParams(false, partitions, state); + auto params = BuildPredictionParams(LF(5), PW(3), false, partitions, state); + EXPECT_EQ(params.LoadFactor, LF(5)); + EXPECT_EQ(params.PartitionWeight, PW(3)); EXPECT_EQ(params.Increasing, false); EXPECT_EQ(params.TotalPartitions, 0); EXPECT_EQ(params.PartitionsWeight, PW(0)); diff --git a/src/core/partition_balancing/tests/collect_active_hubs_ut.cpp b/src/core/partition_balancing/tests/collect_active_hubs_ut.cpp index fba0732..319f712 100644 --- a/src/core/partition_balancing/tests/collect_active_hubs_ut.cpp +++ b/src/core/partition_balancing/tests/collect_active_hubs_ut.cpp @@ -20,23 +20,23 @@ class CollectActiveHubsTest TEST_F(CollectActiveHubsTest, FiltersDrainingLaggedOfflineHubs) { TCoordinationState::TClusterSnapshot snapshot; { - snapshot.emplace(HUB("hub-active"), THubReport{ - EP(42), HUB("hub-active"), DC("myt"), LF(10), { + snapshot.emplace_back( + EP(42), HUB("hub-active"), DC("myt"), LF(10), PWS({ {PID(1), PW(50)}, {PID(7), PW(320)}, - }}); + })); - snapshot.emplace(HUB("hub-draining"), THubReport{ - EP(42), HUB("hub-draining"), DC("myt"), LF(10), { + snapshot.emplace_back( + EP(42), HUB("hub-draining"), DC("myt"), LF(10), PWS({ {PID(2), PW(200)}, {PID(3), PW(410)}, - }}); + })); - snapshot.emplace(HUB("hub-lagged"), THubReport{ - EP(41), HUB("hub-lagged"), DC("myt"), LF(10), { + snapshot.emplace_back( + EP(41), HUB("hub-lagged"), DC("myt"), LF(10), PWS({ {PID(5), PW(309)}, {PID(6), PW(200)}, - }}); + })); } TPartitionMap map{ @@ -67,10 +67,10 @@ TEST_F(CollectActiveHubsTest, FiltersDrainingLaggedOfflineHubs) { TCoordinationState state(map, snapshot, context, settings); - EXPECT_CALL(*Predictor_, PredictLoadFactor(_, _, _)) + EXPECT_CALL(*Predictor_, PredictLoadFactor(_)) .WillOnce(Return(TLoadFactor{15})); - auto [activeHubs, sortedHubs] = CollectActiveHubs(state, Predictor_); + auto [activeHubs, sortedHubs] = CollectActiveHubs(state, *Predictor_); EXPECT_EQ(activeHubs.size(), 1); EXPECT_TRUE(activeHubs.contains(HUB("hub-active"))); @@ -80,16 +80,16 @@ TEST_F(CollectActiveHubsTest, FiltersDrainingLaggedOfflineHubs) { TEST_F(CollectActiveHubsTest, SortsHubsByForecastedLoad) { TCoordinationState::TClusterSnapshot snapshot; { - snapshot.emplace(HUB("hub-heavy"), THubReport{ - EP(42), HUB("hub-heavy"), DC("myt"), LF(80), { + snapshot.emplace_back( + EP(42), HUB("hub-heavy"), DC("myt"), LF(80), PWS({ {PID(1), PW(150)}, {PID(3), PW(220)}, - }}); + })); - snapshot.emplace(HUB("hub-light"), THubReport{ - EP(42), HUB("hub-light"), DC("myt"), LF(20), { + snapshot.emplace_back( + EP(42), HUB("hub-light"), DC("myt"), LF(20), PWS({ {PID(2), PW(75)}, - }}); + })); } TPartitionMap map{ @@ -115,12 +115,12 @@ TEST_F(CollectActiveHubsTest, SortsHubsByForecastedLoad) { TCoordinationState state(map, snapshot, context, settings); // increasing load of light hub - EXPECT_CALL(*Predictor_, PredictLoadFactor(_, _, _)) - .WillRepeatedly(Invoke([](TLoadFactor current, auto, auto) { - return current == TLoadFactor{80} ? TLoadFactor{85} : TLoadFactor{95}; + EXPECT_CALL(*Predictor_, PredictLoadFactor(_)) + .WillRepeatedly(Invoke([](TPredictionParams params) { + return params.LoadFactor == TLoadFactor{80} ? TLoadFactor{85} : TLoadFactor{95}; })); - auto [activeHubs, sortedHubs] = CollectActiveHubs(state, Predictor_); + auto [activeHubs, sortedHubs] = CollectActiveHubs(state, *Predictor_); ASSERT_EQ(sortedHubs.size(), 2); auto it = sortedHubs.begin(); @@ -130,9 +130,9 @@ TEST_F(CollectActiveHubsTest, SortsHubsByForecastedLoad) { TEST_F(CollectActiveHubsTest, PassesCorrectParamsToPredictor) { TCoordinationState::TClusterSnapshot snapshot; - snapshot.emplace(HUB("hub-a"), THubReport{EP(42), HUB("hub-a"), DC("myt"), LF(50), { + snapshot.emplace_back(EP(42), HUB("hub-a"), DC("myt"), LF(50), PWS({ {PID(1), PW(50)} - }}); + })); TPartitionMap map{ .Partitions{{PID(1), HUB("hub-a")}}, @@ -153,13 +153,17 @@ TEST_F(CollectActiveHubsTest, PassesCorrectParamsToPredictor) { TCoordinationState state(map, snapshot, context, settings); - EXPECT_CALL(*Predictor_, PredictLoadFactor( - Eq(LF(50)), - Eq(PW(50)), - Field(&TPredictionParams::TotalPartitions, 1) - )).WillOnce(Return(LF(60))); - - auto [activeHubs, sortedHubs] = CollectActiveHubs(state, Predictor_); + EXPECT_CALL(*Predictor_, + PredictLoadFactor(AllOf( + Field(&TPredictionParams::LoadFactor, LF(50)), + Field(&TPredictionParams::PartitionWeight, PW(50)), + Field(&TPredictionParams::Increasing, true), + Field(&TPredictionParams::TotalPartitions, 1), + Field(&TPredictionParams::PartitionsWeight, PW(50)), + Field(&TPredictionParams::OriginalLoadFactor, LF(50))))) + .WillOnce(Return(LF(60))); + + auto [activeHubs, sortedHubs] = CollectActiveHubs(state, *Predictor_); ASSERT_EQ(sortedHubs.size(), 1); ASSERT_EQ(activeHubs.size(), 1); @@ -168,9 +172,9 @@ TEST_F(CollectActiveHubsTest, PassesCorrectParamsToPredictor) { TEST_F(CollectActiveHubsTest, IncludesOverloadedHubs) { TCoordinationState::TClusterSnapshot snapshot; - snapshot.emplace(HUB("hub-overloaded"), THubReport{EP(42), HUB("hub-overloaded"), DC("myt"), LF(95), { + snapshot.emplace_back(EP(42), HUB("hub-overloaded"), DC("myt"), LF(95), PWS({ {PID(1), PW(970)} - }}); + })); TPartitionMap map{ .Partitions{{PID(1), HUB("hub-overloaded")}}, @@ -190,9 +194,9 @@ TEST_F(CollectActiveHubsTest, IncludesOverloadedHubs) { TCoordinationState state(map, snapshot, context, settings); - EXPECT_CALL(*Predictor_, PredictLoadFactor(_, _, _)).WillOnce(Return(TLoadFactor{99})); + EXPECT_CALL(*Predictor_, PredictLoadFactor(_)).WillOnce(Return(TLoadFactor{99})); - auto [activeHubs, sortedHubs] = CollectActiveHubs(state, Predictor_); + auto [activeHubs, sortedHubs] = CollectActiveHubs(state, *Predictor_); ASSERT_EQ(activeHubs.size(), 1); EXPECT_TRUE(activeHubs.contains(HUB("hub-overloaded"))); diff --git a/src/core/partition_balancing/tests/execute_rebalancing_phase_ut.cpp b/src/core/partition_balancing/tests/execute_rebalancing_phase_ut.cpp index 3d806c1..fd65e0e 100644 --- a/src/core/partition_balancing/tests/execute_rebalancing_phase_ut.cpp +++ b/src/core/partition_balancing/tests/execute_rebalancing_phase_ut.cpp @@ -29,16 +29,26 @@ class ExecuteRebalancingPhaseTest hubPartitions[hub].insert({weight, pid}); } } + + TBalancingSettings GetDefaultSettings() + { + return TBalancingSettings{ + .MaxRebalancePhases = 3, + .MigratingWeightLimit = PW(1000), + .MinLoadFactorDelta = LF(5), + .MigrationBudgetThreshold = PW(5), + .BalancingThresholdCV = 25, + .BalancingTargetCV = 10, + .MinMigrationCooldown = EP(5), + .MigrationWeightPenaltyCoeff = 0.5, + }; + } }; TEST_F(ExecuteRebalancingPhaseTest, MovesPartitionWhenImbalanceIsHigh) { TCoordinationState::TClusterSnapshot snapshot; - snapshot.emplace(HUB("hub-max"), THubReport{ - EP(100), HUB("hub-max"), DC("myt"), LF(90), { {PID(1), PW(20)} } - }); - snapshot.emplace(HUB("hub-min"), THubReport{ - EP(100), HUB("hub-min"), DC("myt"), LF(10), {} - }); + snapshot.emplace_back(EP(100), HUB("hub-max"), DC("myt"), LF(90), PWS({{PID(1), PW(20)}})); + snapshot.emplace_back(EP(100), HUB("hub-min"), DC("myt"), LF(10), PWS({})); TPartitionMap map{ .Partitions{ {PID(1), HUB("hub-max")} }, @@ -59,17 +69,32 @@ TEST_F(ExecuteRebalancingPhaseTest, MovesPartitionWhenImbalanceIsHigh) { AddHubToStructures(sortedHubs, hubPartitions, HUB("hub-min"), LF(10), {}); TMigrationContext migrationContext; - TBalancingSettings settings; + + auto settings = GetDefaultSettings(); settings.MinLoadFactorDelta = LF(10); settings.MigratingWeightLimit = PW(1000); - EXPECT_CALL(*Predictor_, PredictLoadFactor(LF(90), PW(20), _)) - .WillRepeatedly(Return(LF(70))); - - EXPECT_CALL(*Predictor_, PredictLoadFactor(LF(10), PW(20), _)) - .WillRepeatedly(Return(LF(30))); - - ExecuteRebalancingPhase(sortedHubs, hubPartitions, migrationContext, Predictor_, state, settings); + EXPECT_CALL(*Predictor_, + PredictLoadFactor(AllOf( + Field(&TPredictionParams::LoadFactor, LF(90)), + Field(&TPredictionParams::PartitionWeight, PW(20)), + Field(&TPredictionParams::Increasing, false), + Field(&TPredictionParams::TotalPartitions, 1), + Field(&TPredictionParams::PartitionsWeight, PW(20)), + Field(&TPredictionParams::OriginalLoadFactor, LF(90))))) + .WillOnce(Return(LF(70))); + + EXPECT_CALL(*Predictor_, + PredictLoadFactor(AllOf( + Field(&TPredictionParams::LoadFactor, LF(10)), + Field(&TPredictionParams::PartitionWeight, PW(20)), + Field(&TPredictionParams::Increasing, true), + Field(&TPredictionParams::TotalPartitions, 0), + Field(&TPredictionParams::PartitionsWeight, PW(0)), + Field(&TPredictionParams::OriginalLoadFactor, LF(10))))) + .WillOnce(Return(LF(30))); + + ExecuteRebalancingPhase(sortedHubs, hubPartitions, migrationContext, *Predictor_, state, settings); EXPECT_TRUE(hubPartitions[HUB("hub-max")].empty()); @@ -83,12 +108,8 @@ TEST_F(ExecuteRebalancingPhaseTest, MovesPartitionWhenImbalanceIsHigh) { TEST_F(ExecuteRebalancingPhaseTest, SkipsMigrationIfCooldownIsActive) { TCoordinationState::TClusterSnapshot snapshot; - snapshot.emplace(HUB("hub-max"), THubReport{ - EP(100), HUB("hub-max"), DC("myt"), LF(90), { {PID(1), PW(20)} } - }); - snapshot.emplace(HUB("hub-min"), THubReport{ - EP(100), HUB("hub-min"), DC("myt"), LF(10), {} - }); + snapshot.emplace_back(EP(100), HUB("hub-max"), DC("myt"), LF(90), PWS({{PID(1), PW(20)}})); + snapshot.emplace_back(EP(100), HUB("hub-min"), DC("myt"), LF(10), PWS({})); TPartitionMap map{ .Partitions{ {PID(1), HUB("hub-max")} }, @@ -109,11 +130,11 @@ TEST_F(ExecuteRebalancingPhaseTest, SkipsMigrationIfCooldownIsActive) { AddHubToStructures(sortedHubs, hubPartitions, HUB("hub-min"), LF(10), {}); TMigrationContext migrationContext; - TBalancingSettings settings; + auto settings = GetDefaultSettings(); - EXPECT_CALL(*Predictor_, PredictLoadFactor(_, _, _)).Times(0); + EXPECT_CALL(*Predictor_, PredictLoadFactor(_)).Times(0); - ExecuteRebalancingPhase(sortedHubs, hubPartitions, migrationContext, Predictor_, state, settings); + ExecuteRebalancingPhase(sortedHubs, hubPartitions, migrationContext, *Predictor_, state, settings); EXPECT_EQ(hubPartitions[HUB("hub-max")].size(), 1u); EXPECT_TRUE(migrationContext.MigratingPartitions.empty()); @@ -121,8 +142,8 @@ TEST_F(ExecuteRebalancingPhaseTest, SkipsMigrationIfCooldownIsActive) { TEST_F(ExecuteRebalancingPhaseTest, StopsIfDeltaIsTooSmall) { TCoordinationState::TClusterSnapshot snapshot; - snapshot.emplace(HUB("hub-1"), THubReport{EP(100), HUB("hub-1"), DC("myt"), LF(55), {{PID(1), PW(10)}}}); - snapshot.emplace(HUB("hub-2"), THubReport{EP(100), HUB("hub-2"), DC("myt"), LF(50), {}}); + snapshot.emplace_back(EP(100), HUB("hub-1"), DC("myt"), LF(55), PWS({{PID(1), PW(10)}})); + snapshot.emplace_back(EP(100), HUB("hub-2"), DC("myt"), LF(50), PWS({})); TPartitionMap map{ .Partitions{{PID(1), HUB("hub-1")}}, .Epoch{EP(100)} }; TCoordinationContext context; @@ -135,12 +156,13 @@ TEST_F(ExecuteRebalancingPhaseTest, StopsIfDeltaIsTooSmall) { AddHubToStructures(sortedHubs, hubPartitions, HUB("hub-2"), LF(50), {}); TMigrationContext migrationContext; - TBalancingSettings settings; + + auto settings = GetDefaultSettings(); settings.MinLoadFactorDelta = LF(10); - EXPECT_CALL(*Predictor_, PredictLoadFactor(_, _, _)).Times(0); + EXPECT_CALL(*Predictor_, PredictLoadFactor(_)).Times(0); - ExecuteRebalancingPhase(sortedHubs, hubPartitions, migrationContext, Predictor_, state, settings); + ExecuteRebalancingPhase(sortedHubs, hubPartitions, migrationContext, *Predictor_, state, settings); EXPECT_EQ(hubPartitions[HUB("hub-1")].size(), 1u); EXPECT_TRUE(migrationContext.MigratingPartitions.empty()); @@ -148,8 +170,8 @@ TEST_F(ExecuteRebalancingPhaseTest, StopsIfDeltaIsTooSmall) { TEST_F(ExecuteRebalancingPhaseTest, PreventsOvershoot) { TCoordinationState::TClusterSnapshot snapshot; - snapshot.emplace(HUB("hub-max"), THubReport{EP(100), HUB("hub-max"), DC("myt"), LF(100), {{PID(1), PW(30)}}}); - snapshot.emplace(HUB("hub-min"), THubReport{EP(100), HUB("hub-min"), DC("myt"), LF(90), {}}); + snapshot.emplace_back(EP(100), HUB("hub-max"), DC("myt"), LF(100), PWS({{PID(1), PW(30)}})); + snapshot.emplace_back(EP(100), HUB("hub-min"), DC("myt"), LF(90), PWS({})); TPartitionMap map{ .Partitions{{PID(1), HUB("hub-max")}}, .Epoch{EP(100)} }; TCoordinationContext context; @@ -162,13 +184,32 @@ TEST_F(ExecuteRebalancingPhaseTest, PreventsOvershoot) { AddHubToStructures(sortedHubs, hubPartitions, HUB("hub-min"), LF(90), {}); TMigrationContext migrationContext; - TBalancingSettings settings; - settings.MinLoadFactorDelta = LF(5); - - EXPECT_CALL(*Predictor_, PredictLoadFactor(LF(100), PW(30), _)).WillRepeatedly(Return(LF(70))); - EXPECT_CALL(*Predictor_, PredictLoadFactor(LF(90), PW(30), _)).WillRepeatedly(Return(LF(120))); - ExecuteRebalancingPhase(sortedHubs, hubPartitions, migrationContext, Predictor_, state, settings); + auto settings = GetDefaultSettings(); + settings.MinLoadFactorDelta = LF(5); + settings.MigratingWeightLimit = PW(100); + + EXPECT_CALL(*Predictor_, + PredictLoadFactor(AllOf( + Field(&TPredictionParams::LoadFactor, LF(100)), + Field(&TPredictionParams::PartitionWeight, PW(30)), + Field(&TPredictionParams::Increasing, false), + Field(&TPredictionParams::TotalPartitions, 1), + Field(&TPredictionParams::PartitionsWeight, PW(30)), + Field(&TPredictionParams::OriginalLoadFactor, LF(100))))) + .WillOnce(Return(LF(70))); + + EXPECT_CALL(*Predictor_, + PredictLoadFactor(AllOf( + Field(&TPredictionParams::LoadFactor, LF(90)), + Field(&TPredictionParams::PartitionWeight, PW(30)), + Field(&TPredictionParams::Increasing, true), + Field(&TPredictionParams::TotalPartitions, 0), + Field(&TPredictionParams::PartitionsWeight, PW(0)), + Field(&TPredictionParams::OriginalLoadFactor, LF(90))))) + .WillOnce(Return(LF(120))); + + ExecuteRebalancingPhase(sortedHubs, hubPartitions, migrationContext, *Predictor_, state, settings); EXPECT_EQ(hubPartitions[HUB("hub-max")].size(), 1u); EXPECT_TRUE(migrationContext.MigratingPartitions.empty()); @@ -176,8 +217,8 @@ TEST_F(ExecuteRebalancingPhaseTest, PreventsOvershoot) { TEST_F(ExecuteRebalancingPhaseTest, UpdatesContextWhenCancellingMigration) { TCoordinationState::TClusterSnapshot snapshot; - snapshot.emplace(HUB("hub-max"), THubReport{EP(100), HUB("hub-max"), DC("myt"), LF(80), {{PID(1), PW(10)}}}); - snapshot.emplace(HUB("hub-min"), THubReport{EP(100), HUB("hub-min"), DC("myt"), LF(20), {}}); + snapshot.emplace_back(EP(100), HUB("hub-max"), DC("myt"), LF(80), PWS({{PID(1), PW(10)}})); + snapshot.emplace_back(EP(100), HUB("hub-min"), DC("myt"), LF(20), PWS({})); TPartitionMap map{ .Partitions{{PID(1), HUB("hub-max")}}, .Epoch{EP(100)} }; TCoordinationContext context; @@ -193,12 +234,29 @@ TEST_F(ExecuteRebalancingPhaseTest, UpdatesContextWhenCancellingMigration) { migrationContext.MigratingPartitions.emplace(PID(1), HUB("hub-min")); migrationContext.TotalMigratingWeight = PW(50); - TBalancingSettings settings; - - EXPECT_CALL(*Predictor_, PredictLoadFactor(LF(80), PW(10), _)).WillRepeatedly(Return(LF(70))); - EXPECT_CALL(*Predictor_, PredictLoadFactor(LF(20), PW(10), _)).WillRepeatedly(Return(LF(30))); - - ExecuteRebalancingPhase(sortedHubs, hubPartitions, migrationContext, Predictor_, state, settings); + auto settings = GetDefaultSettings(); + + EXPECT_CALL(*Predictor_, + PredictLoadFactor(AllOf( + Field(&TPredictionParams::LoadFactor, LF(80)), + Field(&TPredictionParams::PartitionWeight, PW(10)), + Field(&TPredictionParams::Increasing, false), + Field(&TPredictionParams::TotalPartitions, 1), + Field(&TPredictionParams::PartitionsWeight, PW(10)), + Field(&TPredictionParams::OriginalLoadFactor, LF(80))))) + .WillOnce(Return(LF(70))); + + EXPECT_CALL(*Predictor_, + PredictLoadFactor(AllOf( + Field(&TPredictionParams::LoadFactor, LF(20)), + Field(&TPredictionParams::PartitionWeight, PW(10)), + Field(&TPredictionParams::Increasing, true), + Field(&TPredictionParams::TotalPartitions, 0), + Field(&TPredictionParams::PartitionsWeight, PW(0)), + Field(&TPredictionParams::OriginalLoadFactor, LF(20))))) + .WillOnce(Return(LF(30))); + + ExecuteRebalancingPhase(sortedHubs, hubPartitions, migrationContext, *Predictor_, state, settings); EXPECT_EQ(hubPartitions[HUB("hub-min")].size(), 1u); diff --git a/src/core/partition_balancing/tests/load_factor_predictor_mock.hpp b/src/core/partition_balancing/tests/load_factor_predictor_mock.hpp index 4daf6c0..5ca5186 100644 --- a/src/core/partition_balancing/tests/load_factor_predictor_mock.hpp +++ b/src/core/partition_balancing/tests/load_factor_predictor_mock.hpp @@ -15,9 +15,7 @@ class TMockLoadFactorPredictor { public: MOCK_METHOD(NDomain::TLoadFactor, PredictLoadFactor, - (const NDomain::TLoadFactor loadFactor, - const NDomain::TPartitionWeight partitionWeight, - const TPredictionParams& params), + (const TPredictionParams& params), (const, override)); }; diff --git a/src/core/partition_balancing/tests/rebalance_partitions_ut.cpp b/src/core/partition_balancing/tests/rebalance_partitions_ut.cpp index 5d334d8..2ea576f 100644 --- a/src/core/partition_balancing/tests/rebalance_partitions_ut.cpp +++ b/src/core/partition_balancing/tests/rebalance_partitions_ut.cpp @@ -20,8 +20,8 @@ class RebalancePartitionsTest TEST_F(RebalancePartitionsTest, DoesNothingWhenCVBelowThreshold) { TCoordinationState::TClusterSnapshot snapshot; - snapshot.emplace(HUB("hub-1"), THubReport{EP(100), HUB("hub-1"), DC("myt"), LF(50), {}}); - snapshot.emplace(HUB("hub-2"), THubReport{EP(100), HUB("hub-2"), DC("myt"), LF(52), {}}); + snapshot.emplace_back(EP(100), HUB("hub-1"), DC("myt"), LF(50), PWS({})); + snapshot.emplace_back(EP(100), HUB("hub-2"), DC("myt"), LF(52), PWS({})); TPartitionMap map{ .Partitions{}, @@ -43,13 +43,13 @@ TEST_F(RebalancePartitionsTest, DoesNothingWhenCVBelowThreshold) { TBalancingSettings settings; settings.BalancingThresholdCV = 1.0; - EXPECT_CALL(*Predictor_, PredictLoadFactor(_, _, _)).Times(0); + EXPECT_CALL(*Predictor_, PredictLoadFactor(_)).Times(0); RebalancePartitions( sortedHubs, assignedPartitions, migrationContext, - Predictor_, + *Predictor_, state, settings); @@ -59,12 +59,8 @@ TEST_F(RebalancePartitionsTest, DoesNothingWhenCVBelowThreshold) { TEST_F(RebalancePartitionsTest, PerformsSingleRebalancingPhase) { TCoordinationState::TClusterSnapshot snapshot; - snapshot.emplace(HUB("hub-max"), THubReport{ - EP(100), HUB("hub-max"), DC("myt"), LF(90), {{PID(1), PW(20)}} - }); - snapshot.emplace(HUB("hub-min"), THubReport{ - EP(100), HUB("hub-min"), DC("myt"), LF(10), {} - }); + snapshot.emplace_back(EP(100), HUB("hub-max"), DC("myt"), LF(90), PWS({{PID(1), PW(20)}})); + snapshot.emplace_back(EP(100), HUB("hub-min"), DC("myt"), LF(10), PWS({})); TPartitionMap map{ .Partitions{{PID(1), HUB("hub-max")}}, @@ -95,17 +91,31 @@ TEST_F(RebalancePartitionsTest, PerformsSingleRebalancingPhase) { settings.MigratingWeightLimit = PW(1000); settings.MaxRebalancePhases = 1; - EXPECT_CALL(*Predictor_, PredictLoadFactor(LF(90), PW(20), _)) - .WillRepeatedly(Return(LF(70))); - - EXPECT_CALL(*Predictor_, PredictLoadFactor(LF(10), PW(20), _)) - .WillRepeatedly(Return(LF(30))); + EXPECT_CALL(*Predictor_, + PredictLoadFactor(AllOf( + Field(&TPredictionParams::LoadFactor, LF(90)), + Field(&TPredictionParams::PartitionWeight, PW(20)), + Field(&TPredictionParams::Increasing, false), + Field(&TPredictionParams::TotalPartitions, 1), + Field(&TPredictionParams::PartitionsWeight, PW(20)), + Field(&TPredictionParams::OriginalLoadFactor, LF(90))))) + .WillOnce(Return(LF(70))); + + EXPECT_CALL(*Predictor_, + PredictLoadFactor(AllOf( + Field(&TPredictionParams::LoadFactor, LF(10)), + Field(&TPredictionParams::PartitionWeight, PW(20)), + Field(&TPredictionParams::Increasing, true), + Field(&TPredictionParams::TotalPartitions, 0), + Field(&TPredictionParams::PartitionsWeight, PW(0)), + Field(&TPredictionParams::OriginalLoadFactor, LF(10))))) + .WillOnce(Return(LF(30))); RebalancePartitions( sortedHubs, assignedPartitions, migrationContext, - Predictor_, + *Predictor_, state, settings); @@ -119,12 +129,8 @@ TEST_F(RebalancePartitionsTest, PerformsSingleRebalancingPhase) { TEST_F(RebalancePartitionsTest, StopsWhenMigrationBudgetExceeded) { TCoordinationState::TClusterSnapshot snapshot; - snapshot.emplace(HUB("hub-max"), THubReport{ - EP(100), HUB("hub-max"), DC("myt"), LF(90), {{PID(1), PW(50)}} - }); - snapshot.emplace(HUB("hub-min"), THubReport{ - EP(100), HUB("hub-min"), DC("myt"), LF(10), {} - }); + snapshot.emplace_back(EP(100), HUB("hub-max"), DC("myt"), LF(90), PWS({{PID(1), PW(50)}})); + snapshot.emplace_back(EP(100), HUB("hub-min"), DC("myt"), LF(10), PWS({})); TPartitionMap map{ .Partitions{{PID(1), HUB("hub-max")}}, @@ -156,13 +162,13 @@ TEST_F(RebalancePartitionsTest, StopsWhenMigrationBudgetExceeded) { settings.MigratingWeightLimit = PW(50); settings.MaxRebalancePhases = 10; - EXPECT_CALL(*Predictor_, PredictLoadFactor(_, _, _)).Times(0); + EXPECT_CALL(*Predictor_, PredictLoadFactor(_)).Times(0); RebalancePartitions( sortedHubs, assignedPartitions, migrationContext, - Predictor_, + *Predictor_, state, settings); @@ -171,8 +177,8 @@ TEST_F(RebalancePartitionsTest, StopsWhenMigrationBudgetExceeded) { TEST_F(RebalancePartitionsTest, RespectsMaxRebalancePhases) { TCoordinationState::TClusterSnapshot snapshot; - snapshot.emplace(HUB("hub-1"), THubReport{EP(100), HUB("hub-1"), DC("myt"), LF(80), {{PID(1), PW(10)}}}); - snapshot.emplace(HUB("hub-2"), THubReport{EP(100), HUB("hub-2"), DC("myt"), LF(20), {}}); + snapshot.emplace_back(EP(100), HUB("hub-1"), DC("myt"), LF(80), PWS({{PID(1), PW(10)}})); + snapshot.emplace_back(EP(100), HUB("hub-2"), DC("myt"), LF(20), PWS({})); TPartitionMap map{ .Partitions{{PID(1), HUB("hub-1")}}, @@ -201,13 +207,13 @@ TEST_F(RebalancePartitionsTest, RespectsMaxRebalancePhases) { settings.BalancingTargetCV = 0.0; settings.MaxRebalancePhases = 0; - EXPECT_CALL(*Predictor_, PredictLoadFactor(_, _, _)).Times(0); + EXPECT_CALL(*Predictor_, PredictLoadFactor(_)).Times(0); RebalancePartitions( sortedHubs, assignedPartitions, migrationContext, - Predictor_, + *Predictor_, state, settings); diff --git a/src/core/partition_balancing/tests/separate_partitions_ut.cpp b/src/core/partition_balancing/tests/separate_partitions_ut.cpp index 93e5408..67b56d6 100644 --- a/src/core/partition_balancing/tests/separate_partitions_ut.cpp +++ b/src/core/partition_balancing/tests/separate_partitions_ut.cpp @@ -19,11 +19,11 @@ class SeparatePartitionsTest TEST_F(SeparatePartitionsTest, CorrectLabelsAndSortsOrphanedPartitions) { TCoordinationState::TClusterSnapshot snapshot; - snapshot.emplace(HUB("hub-active"), THubReport{ - EP(42), HUB("hub-active"), DC("myt"), LF(25), { + snapshot.emplace_back( + EP(42), HUB("hub-active"), DC("myt"), LF(25), PWS({ {PID(1), PW(100)}, {PID(3), PW(150)}, - }}); + })); TPartitionMap map{ .Partitions{ @@ -49,10 +49,10 @@ TEST_F(SeparatePartitionsTest, CorrectLabelsAndSortsOrphanedPartitions) { }; TCoordinationState state(map, snapshot, context, settings); - EXPECT_CALL(*Predictor_, PredictLoadFactor(_, _, _)) + EXPECT_CALL(*Predictor_, PredictLoadFactor(_)) .WillOnce(Return(TLoadFactor{25})); - auto [activeHubs, sortedHubs] = CollectActiveHubs(state, Predictor_); + auto [activeHubs, sortedHubs] = CollectActiveHubs(state, *Predictor_); ASSERT_EQ(activeHubs.size(), 1); auto result = SeparatePartitions(state, activeHubs); @@ -71,10 +71,10 @@ TEST_F(SeparatePartitionsTest, CorrectLabelsAndSortsOrphanedPartitions) { TEST_F(SeparatePartitionsTest, FallbackToAverageWeightWhenObservedMissing) { TCoordinationState::TClusterSnapshot snapshot; - snapshot.emplace(HUB("hub-active"), THubReport{ - EP(42), HUB("hub-active"), DC("myt"), LF(25), { + snapshot.emplace_back( + EP(42), HUB("hub-active"), DC("myt"), LF(25), PWS({ {PID(1), PW(200)}, - }}); + })); TPartitionMap map{ .Partitions{ @@ -159,9 +159,7 @@ TEST_F(SeparatePartitionsTest, SortsOrphansByCalculatedWeightDescending) { TEST_F(SeparatePartitionsTest, AllHubsActiveNoOrphans) { TCoordinationState::TClusterSnapshot snapshot; - snapshot.emplace(HUB("hub-1"), THubReport{ - EP(10), HUB("hub-1"), DC("dc1"), LF(10), {{PID(1), PW(100)}} - }); + snapshot.emplace_back(EP(10), HUB("hub-1"), DC("dc1"), LF(10), PWS({{PID(1), PW(100)}})); TPartitionMap map{ .Partitions{{PID(1), HUB("hub-1")}}, @@ -187,9 +185,7 @@ TEST_F(SeparatePartitionsTest, AllHubsActiveNoOrphans) { TEST_F(SeparatePartitionsTest, AddsExpectedGrowthToWeight) { TCoordinationState::TClusterSnapshot snapshot; - snapshot.emplace(HUB("hub-1"), THubReport{ - EP(10), HUB("hub-1"), DC("dc1"), LF(10), {{PID(1), PW(100)}} - }); + snapshot.emplace_back(EP(10), HUB("hub-1"), DC("dc1"), LF(10), PWS({{PID(1), PW(100)}})); TPartitionMap map{ .Partitions{{PID(1), HUB("hub-1")}}, diff --git a/src/core/partition_balancing/tests/test_helpers_ut.cpp b/src/core/partition_balancing/tests/test_helpers_ut.cpp index 18b1a9d..b0d1df6 100644 --- a/src/core/partition_balancing/tests/test_helpers_ut.cpp +++ b/src/core/partition_balancing/tests/test_helpers_ut.cpp @@ -7,7 +7,7 @@ namespace NCoordinator::NCore { //////////////////////////////////////////////////////////////////////////////// void TBalancingTestBase::SetUp() { - Predictor_ = std::make_shared(); + Predictor_ = std::make_unique(); } TEpoch TBalancingTestBase::EP(TEpoch::UnderlyingType epoch) const @@ -25,6 +25,12 @@ TPartitionWeight TBalancingTestBase::PW(TPartitionWeight::UnderlyingType partiti return TPartitionWeight(partitionWeight); } +std::unordered_map TBalancingTestBase::PWS( + std::initializer_list> list) const +{ + return std::unordered_map(list); +} + THubEndpoint TBalancingTestBase::HUB(THubEndpoint::UnderlyingType hub) const { return THubEndpoint{std::move(hub)}; diff --git a/src/core/partition_balancing/tests/test_helpers_ut.hpp b/src/core/partition_balancing/tests/test_helpers_ut.hpp index caab8c8..bbd1ff7 100644 --- a/src/core/partition_balancing/tests/test_helpers_ut.hpp +++ b/src/core/partition_balancing/tests/test_helpers_ut.hpp @@ -28,13 +28,15 @@ class TBalancingTestBase TEpoch EP(TEpoch::UnderlyingType epoch) const; TPartitionId PID(TPartitionId::UnderlyingType partitionId) const; TPartitionWeight PW(TPartitionWeight::UnderlyingType partitionWeight) const; + std::unordered_map PWS( + std::initializer_list> list) const; THubEndpoint HUB(THubEndpoint::UnderlyingType hub) const; THubDC DC(THubDC::UnderlyingType dc) const; TLoadFactor LF(TLoadFactor::UnderlyingType lf) const; protected: - std::shared_ptr Predictor_; + std::unique_ptr Predictor_; }; //////////////////////////////////////////////////////////////////////////////// diff --git a/src/hello.cpp b/src/hello.cpp index e920ee5..3071cf7 100644 --- a/src/hello.cpp +++ b/src/hello.cpp @@ -4,7 +4,7 @@ #include -#include +#include #include #include @@ -12,19 +12,16 @@ namespace coordinator { Hello::Hello(const userver::components::ComponentConfig& config, const userver::components::ComponentContext& context) - : HttpHandlerJsonBase(config, context), - ydb_client_(context.FindComponent().GetCoordinationClient("chat_db")) -{ - gateway = std::make_unique( - ydb_client_, "chat-coordination", "partition-map-lock", "discovery-lock", false); -} + : HttpHandlerJsonBase(config, context) + , gateway(context.FindComponent().GetGateway()) +{ } userver::formats::json::Value Hello::HandleRequestJsonThrow( const userver::server::http::HttpRequest& , const userver::formats::json::Value& , userver::server::request::RequestContext&) const { - auto result = gateway->GetPartitionMap(); + auto result = gateway.GetPartitionMap(); return NCoordinator::NInfra::SerializePartitionMap(result.value()); } diff --git a/src/hello.hpp b/src/hello.hpp index ef9ce44..3b23267 100644 --- a/src/hello.hpp +++ b/src/hello.hpp @@ -4,7 +4,7 @@ #include #include #include -#include +#include #include #include @@ -27,8 +27,7 @@ class Hello final : public userver::server::handlers::HttpHandlerJsonBase { private: NYdb::NCoordination::TSessionSettings settings; - std::shared_ptr ydb_client_; - std::unique_ptr gateway; + NCoordinator::NCore::NDomain::ICoordinationGateway& gateway; }; } // namespace coordinator \ No newline at end of file diff --git a/src/infra/components/components.cpp b/src/infra/components/components.cpp index 9b7c0b9..c2c1636 100644 --- a/src/infra/components/components.cpp +++ b/src/infra/components/components.cpp @@ -1,8 +1,11 @@ #include "components.hpp" -#include +#include +#include #include #include +#include +#include #include @@ -45,7 +48,15 @@ void RegisterInfraComponents(userver::components::ComponentList& list) { list.Append() .Append() - .Append(); + .Append() + .Append() + .Append(); +} + +// Services +void RegisterServices(userver::components::ComponentList& list) +{ + list.Append(); } // Handlers diff --git a/src/infra/components/components.hpp b/src/infra/components/components.hpp index f9912c9..997cdcd 100644 --- a/src/infra/components/components.hpp +++ b/src/infra/components/components.hpp @@ -14,6 +14,9 @@ void RegisterYdbComponents(userver::components::ComponentList&); // Infra void RegisterInfraComponents(userver::components::ComponentList&); +// Services +void RegisterServices(userver::components::ComponentList&); + // Handlers void RegisterHandlers(userver::components::ComponentList& list); diff --git a/src/infra/components/coordination/coordination_dist_lock_component.cpp b/src/infra/components/coordination/coordination_dist_lock_component.cpp deleted file mode 100644 index 77536a5..0000000 --- a/src/infra/components/coordination/coordination_dist_lock_component.cpp +++ /dev/null @@ -1,52 +0,0 @@ -#include "coordination_dist_lock_component.hpp" - -#include - -#include -#include -#include - -#include - -namespace { - -//////////////////////////////////////////////////////////////////////////////// - -const auto DEFAULT_COORDINATION_PAUSE_SECONDS = std::chrono::seconds(10); // TODO replace with dynamic config - -//////////////////////////////////////////////////////////////////////////////// - -} // anonymous namespace - -namespace NCoordinator::NInfra::NComponents { - -//////////////////////////////////////////////////////////////////////////////// - -TCoordinationDistLockComponent::TCoordinationDistLockComponent( - const userver::components::ComponentConfig& config, - const userver::components::ComponentContext& context) - : DistLockComponentBase(config, context) - , Gateway_(context.FindComponent().GetGateway()) -{ - Start(); -} - -TCoordinationDistLockComponent::~TCoordinationDistLockComponent() -{ - Stop(); -} - -void TCoordinationDistLockComponent::DoWork() { - while (!userver::engine::current_task::ShouldCancel()) { - auto partitionMap = Gateway_.GetPartitionMap().value_or(NCore::NDomain::TPartitionMap{}); - - ++(partitionMap.Epoch.GetUnderlying()); - Gateway_.BroadcastPartitionMap(partitionMap); - - userver::engine::InterruptibleSleepFor(DEFAULT_COORDINATION_PAUSE_SECONDS); // TODO replace with dynamic config - } -} - -//////////////////////////////////////////////////////////////////////////////// - -} // namespace NCoordinator::NInfra::NComponents diff --git a/src/infra/components/hub/hub_gateway_component.cpp b/src/infra/components/hub/hub_gateway_component.cpp new file mode 100644 index 0000000..a608f27 --- /dev/null +++ b/src/infra/components/hub/hub_gateway_component.cpp @@ -0,0 +1,30 @@ +#include "hub_gateway_component.hpp" + +#include + +#include +#include +#include + +namespace NCoordinator::NInfra::NComponents { + +//////////////////////////////////////////////////////////////////////////////// + +THubGatewayComponent::THubGatewayComponent( + const userver::components::ComponentConfig& config, + const userver::components::ComponentContext& context) + : LoggableComponentBase(config, context) +{ + auto& httpClient = context.FindComponent().GetHttpClient(); + + Gateway_ = std::make_unique(httpClient); +} + +NCore::NDomain::IHubGateway& THubGatewayComponent::GetGateway() +{ + return *Gateway_; +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NCoordinator::NInfra::NComponents diff --git a/src/infra/components/hub/hub_gateway_component.hpp b/src/infra/components/hub/hub_gateway_component.hpp new file mode 100644 index 0000000..817f0a8 --- /dev/null +++ b/src/infra/components/hub/hub_gateway_component.hpp @@ -0,0 +1,31 @@ +#pragma once + +#include + +#include + +#include + +namespace NCoordinator::NInfra::NComponents { + +//////////////////////////////////////////////////////////////////////////////// + +class THubGatewayComponent + : public userver::components::LoggableComponentBase +{ +public: + static constexpr std::string_view kName = "hub-gateway"; + + THubGatewayComponent( + const userver::components::ComponentConfig& config, + const userver::components::ComponentContext& context); + + NCore::NDomain::IHubGateway& GetGateway(); + +private: + std::unique_ptr Gateway_; +}; + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NCoordinator::NInfra::NComponents diff --git a/src/infra/components/leader/leader_dist_lock_component.cpp b/src/infra/components/leader/leader_dist_lock_component.cpp new file mode 100644 index 0000000..880edf3 --- /dev/null +++ b/src/infra/components/leader/leader_dist_lock_component.cpp @@ -0,0 +1,72 @@ +#include "leader_dist_lock_component.hpp" + +#include "leader_service_component.hpp" + +#include +#include +#include + +#include + +namespace { + +//////////////////////////////////////////////////////////////////////////////// + +const auto DEFAULT_COORDINATION_PAUSE_SECONDS = std::chrono::seconds(10); // TODO replace with dynamic config + +//////////////////////////////////////////////////////////////////////////////// + +} // anonymous namespace + +namespace NCoordinator::NInfra::NComponents { + +//////////////////////////////////////////////////////////////////////////////// + +TLeaderDistLockComponent::TLeaderDistLockComponent( + const userver::components::ComponentConfig& config, + const userver::components::ComponentContext& context) + : DistLockComponentBase(config, context) + , Service_(context.FindComponent().GetService()) +{ + Start(); +} + +TLeaderDistLockComponent::~TLeaderDistLockComponent() +{ + Stop(); +} + +void TLeaderDistLockComponent::DoWork() { + while (!userver::engine::current_task::ShouldCancel()) { + + // TODO replace to settings from dynconfig + NCore::NDomain::TStateBuildingSettings stateBuildingSettings{ + .BlockedDCs = {}, + .BlockedHubs = {}, + .OverloadThreshold = NCore::NDomain::TLoadFactor{90}, + }; + + NCore::TBalancingSettings balancingSettings{ + .MaxRebalancePhases = 5, + .MigratingWeightLimit = NCore::NDomain::TPartitionWeight{1500}, + .MinLoadFactorDelta = NCore::NDomain::TLoadFactor{10}, + .MigrationBudgetThreshold = NCore::NDomain::TPartitionWeight{100}, + .BalancingThresholdCV = 25, + .BalancingTargetCV = 5, + .MinMigrationCooldown = NCore::NDomain::TEpoch{5}, + .MigrationWeightPenaltyCoeff = 0.5, + }; + + NApp::NDto::TCoordinationRequest request{ + .StateBuildingSettings = std::move(stateBuildingSettings), + .BalancingSettings = std::move(balancingSettings), + }; + Service_.Coordinate(std::move(request)); + + userver::engine::InterruptibleSleepFor(DEFAULT_COORDINATION_PAUSE_SECONDS); // TODO replace with dynamic config + } +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NCoordinator::NInfra::NComponents diff --git a/src/infra/components/coordination/coordination_dist_lock_component.hpp b/src/infra/components/leader/leader_dist_lock_component.hpp similarity index 63% rename from src/infra/components/coordination/coordination_dist_lock_component.hpp rename to src/infra/components/leader/leader_dist_lock_component.hpp index 5c9738b..a76a22a 100644 --- a/src/infra/components/coordination/coordination_dist_lock_component.hpp +++ b/src/infra/components/leader/leader_dist_lock_component.hpp @@ -1,6 +1,6 @@ #pragma once -#include +#include #include @@ -8,22 +8,22 @@ namespace NCoordinator::NInfra::NComponents { //////////////////////////////////////////////////////////////////////////////// -class TCoordinationDistLockComponent +class TLeaderDistLockComponent : public userver::ydb::DistLockComponentBase { public: - static constexpr std::string_view kName = "coordination-dist-lock"; + static constexpr std::string_view kName = "leader-dist-lock"; - TCoordinationDistLockComponent( + TLeaderDistLockComponent( const userver::components::ComponentConfig& config, const userver::components::ComponentContext& context); - ~TCoordinationDistLockComponent(); + ~TLeaderDistLockComponent(); void DoWork() final; private: - NCore::NDomain::ICoordinationGateway& Gateway_; // Replace with service with core + NApp::NService::TLeaderService& Service_; }; //////////////////////////////////////////////////////////////////////////////// diff --git a/src/infra/components/leader/leader_service_component.cpp b/src/infra/components/leader/leader_service_component.cpp new file mode 100644 index 0000000..098fb2d --- /dev/null +++ b/src/infra/components/leader/leader_service_component.cpp @@ -0,0 +1,41 @@ +#include "leader_service_component.hpp" + +#include +#include +#include +#include + +#include +#include +#include + +namespace NCoordinator::NInfra::NComponents { + +//////////////////////////////////////////////////////////////////////////////// + +TLeaderServiceComponent::TLeaderServiceComponent( + const userver::components::ComponentConfig& config, + const userver::components::ComponentContext& context) + : LoggableComponentBase(config, context) +{ + auto& coordinationGateway = context.FindComponent().GetGateway(); + auto& coordinationRepository = + context.FindComponent().GetRepository(); + auto& hubGateway = context.FindComponent().GetGateway(); + auto& predictor = context.FindComponent().GetPredictor(); + + Service_ = std::make_unique( + coordinationGateway, + coordinationRepository, + hubGateway, + predictor); +} + +NApp::NService::TLeaderService& TLeaderServiceComponent::GetService() +{ + return *Service_; +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NCoordinator::NInfra::NComponents diff --git a/src/infra/components/leader/leader_service_component.hpp b/src/infra/components/leader/leader_service_component.hpp new file mode 100644 index 0000000..d4e0642 --- /dev/null +++ b/src/infra/components/leader/leader_service_component.hpp @@ -0,0 +1,31 @@ +#pragma once + +#include + +#include + +#include + +namespace NCoordinator::NInfra::NComponents { + +//////////////////////////////////////////////////////////////////////////////// + +class TLeaderServiceComponent + : public userver::components::LoggableComponentBase +{ +public: + static constexpr std::string_view kName = "leader-service"; + + TLeaderServiceComponent( + const userver::components::ComponentConfig& config, + const userver::components::ComponentContext& context); + + NApp::NService::TLeaderService& GetService(); + +private: + std::unique_ptr Service_; +}; + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NCoordinator::NInfra::NComponents diff --git a/src/infra/components/partition_balancing/load_factor_predictor_component.cpp b/src/infra/components/partition_balancing/load_factor_predictor_component.cpp new file mode 100644 index 0000000..612b198 --- /dev/null +++ b/src/infra/components/partition_balancing/load_factor_predictor_component.cpp @@ -0,0 +1,44 @@ +#include "load_factor_predictor_component.hpp" + +#include + +#include +#include +#include + +namespace NCoordinator::NInfra::NComponents { + +//////////////////////////////////////////////////////////////////////////////// + +TLoadFactorPredictorComponent::TLoadFactorPredictorComponent( + const userver::components::ComponentConfig& config, + const userver::components::ComponentContext& context) + : LoggableComponentBase(config, context) +{ + // TODO other types of predictors + // auto type = config["type"].As(); + + Predictor_ = std::make_unique(); +} + +NCore::ILoadFactorPredictor& TLoadFactorPredictorComponent::GetPredictor() +{ + return *Predictor_; +} + +userver::yaml_config::Schema TLoadFactorPredictorComponent::GetStaticConfigSchema() { + return userver::yaml_config::MergeSchemas( + R"( +type: object +description: Component for load factor predictor +additionalProperties: false +properties: + type: + type: string + description: the type of predictor (heuristic, ...) +)"); +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NCoordinator::NInfra::NComponents diff --git a/src/infra/components/partition_balancing/load_factor_predictor_component.hpp b/src/infra/components/partition_balancing/load_factor_predictor_component.hpp new file mode 100644 index 0000000..488ded7 --- /dev/null +++ b/src/infra/components/partition_balancing/load_factor_predictor_component.hpp @@ -0,0 +1,33 @@ +#pragma once + +#include + +#include + +#include + +namespace NCoordinator::NInfra::NComponents { + +//////////////////////////////////////////////////////////////////////////////// + +class TLoadFactorPredictorComponent + : public userver::components::LoggableComponentBase +{ +public: + static constexpr std::string_view kName = "load-factor-predictor"; + + TLoadFactorPredictorComponent( + const userver::components::ComponentConfig& config, + const userver::components::ComponentContext& context); + + NCore::ILoadFactorPredictor& GetPredictor(); + + static userver::yaml_config::Schema GetStaticConfigSchema(); + +private: + std::unique_ptr Predictor_; +}; + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NCoordinator::NInfra::NComponents diff --git a/src/infra/coordination_gateway/kesus_coordination_gateway.cpp b/src/infra/coordination_gateway/kesus_coordination_gateway.cpp index 54fc7ef..f07e2cd 100644 --- a/src/infra/coordination_gateway/kesus_coordination_gateway.cpp +++ b/src/infra/coordination_gateway/kesus_coordination_gateway.cpp @@ -113,6 +113,7 @@ void TKesusCoordinationGateway::InitialSetup( PartitionMapSemaphore_, PARTITION_MAP_SEMAPHORE_LIMIT); // std::string{DEFAULT_PARTITION_MAP_DATA}); + session.UpdateSemaphore(PartitionMapSemaphore_, std::string{DEFAULT_PARTITION_MAP_DATA}); } catch (const userver::ydb::YdbResponseError& ex) { LOG_WARNING() << "Could not create partition map semaphore: " << ex; } diff --git a/src/infra/load_factor_predictor/heuristic_predictor.cpp b/src/infra/load_factor_predictor/heuristic_predictor.cpp new file mode 100644 index 0000000..b8dfaab --- /dev/null +++ b/src/infra/load_factor_predictor/heuristic_predictor.cpp @@ -0,0 +1,54 @@ +#include "heuristic_predictor.hpp" + +#include + +#include +#include +#include + +namespace NCoordinator::NInfra { + +//////////////////////////////////////////////////////////////////////////////// + +NCore::NDomain::TLoadFactor THeuristicPredictor::PredictLoadFactor( + const NCore::TPredictionParams& params) const +{ + const double currentTotalWeight = static_cast(params.PartitionsWeight.GetUnderlying()); + + if (currentTotalWeight <= std::numeric_limits::epsilon() || params.TotalPartitions == 0) { + if (params.Increasing) { + return DefaultFirstLoadFactor; + } else { + return NCore::NDomain::TLoadFactor{0}; + } + } + + const double currentLoadFactor = static_cast(params.LoadFactor.GetUnderlying()); + const double deltaWeight = static_cast(params.PartitionWeight.GetUnderlying()); + + double newTotalWeight = currentTotalWeight; + if (params.Increasing) { + newTotalWeight += deltaWeight; + } else { + newTotalWeight -= deltaWeight; + } + + if (newTotalWeight < 0.0) { + newTotalWeight = 0.0; + } + + double predictedVal = currentLoadFactor * (newTotalWeight / currentTotalWeight); + + predictedVal = std::ceil(predictedVal); + + const auto result = std::clamp( + static_cast(predictedVal), + std::uint64_t{0}, + std::uint64_t{100}); + + return NCore::NDomain::TLoadFactor{result}; +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NCoordinator::NInfra diff --git a/src/infra/load_factor_predictor/heuristic_predictor.hpp b/src/infra/load_factor_predictor/heuristic_predictor.hpp new file mode 100644 index 0000000..12cd8ca --- /dev/null +++ b/src/infra/load_factor_predictor/heuristic_predictor.hpp @@ -0,0 +1,22 @@ +#pragma once + +#include +#include + +namespace NCoordinator::NInfra { + +//////////////////////////////////////////////////////////////////////////////// + +class THeuristicPredictor + : public NCore::ILoadFactorPredictor +{ +public: + NCore::NDomain::TLoadFactor PredictLoadFactor(const NCore::TPredictionParams& params) const override; + +private: + NCore::NDomain::TLoadFactor DefaultFirstLoadFactor{5}; // TODO replace with dynconfig +}; + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NCoordinator::NInfra diff --git a/src/infra/load_factor_predictor/heuristic_predictor_ut.cpp b/src/infra/load_factor_predictor/heuristic_predictor_ut.cpp new file mode 100644 index 0000000..749363e --- /dev/null +++ b/src/infra/load_factor_predictor/heuristic_predictor_ut.cpp @@ -0,0 +1,131 @@ +#include "heuristic_predictor.hpp" + +#include + +namespace { + +//////////////////////////////////////////////////////////////////////////////// + +using namespace NCoordinator::NCore; +using namespace NCoordinator::NCore::NDomain; +using namespace NCoordinator::NInfra; + +TPredictionParams MakeParams( + uint64_t currentLf, + uint64_t totalWeight, + uint64_t deltaWeight, + bool increasing, + size_t totalPartitions = 10) +{ + TPredictionParams p; + p.LoadFactor = TLoadFactor{currentLf}; + p.PartitionsWeight = TPartitionWeight{totalWeight}; + p.PartitionWeight = TPartitionWeight{deltaWeight}; + p.Increasing = increasing; + p.TotalPartitions = totalPartitions; + + p.OriginalLoadFactor = TLoadFactor{currentLf}; + return p; +} + +//////////////////////////////////////////////////////////////////////////////// + +} // anonymous namespace + +namespace NCoordinator::NInfra { + +//////////////////////////////////////////////////////////////////////////////// + +TEST(THeuristicPredictor, LinearGrowth) +{ + const THeuristicPredictor predictor; + + auto params = MakeParams(50, 100, 20, true); + auto result = predictor.PredictLoadFactor(params); + + EXPECT_EQ(result.GetUnderlying(), 60); +} + +TEST(THeuristicPredictor, LinearDecline) +{ + const THeuristicPredictor predictor; + + auto params = MakeParams(50, 100, 20, false); + auto result = predictor.PredictLoadFactor(params); + + EXPECT_EQ(result.GetUnderlying(), 40); +} + +TEST(THeuristicPredictor, FirstPartitionInsertionUsesDefault) +{ + const THeuristicPredictor predictor; + + TPredictionParams params; + params.LoadFactor = TLoadFactor{0}; + params.PartitionsWeight = TPartitionWeight{0}; + params.PartitionWeight = TPartitionWeight{10}; + params.Increasing = true; + params.TotalPartitions = 0; + + auto result = predictor.PredictLoadFactor(params); + + EXPECT_EQ(result.GetUnderlying(), 5); +} + +TEST(THeuristicPredictor, ClampsToMaxLoadFactor) +{ + const THeuristicPredictor predictor; + + auto params = MakeParams(80, 100, 50, true); + auto result = predictor.PredictLoadFactor(params); + + EXPECT_EQ(result.GetUnderlying(), 100); +} + +TEST(THeuristicPredictor, ClampsToZeroLoadFactor) +{ + const THeuristicPredictor predictor; + + auto params = MakeParams(10, 100, 100, false); + auto result = predictor.PredictLoadFactor(params); + + EXPECT_EQ(result.GetUnderlying(), 0); +} + +TEST(THeuristicPredictor, RoundsUpPessimistically) +{ + const THeuristicPredictor predictor; + + auto params = MakeParams(10, 100, 1, true); + auto result = predictor.PredictLoadFactor(params); + + EXPECT_EQ(result.GetUnderlying(), 11); +} + +TEST(THeuristicPredictor, HandlesZeroDeltaWeight) +{ + const THeuristicPredictor predictor; + + auto params = MakeParams(50, 100, 0, true); + auto result = predictor.PredictLoadFactor(params); + + EXPECT_EQ(result.GetUnderlying(), 50); +} + +TEST(THeuristicPredictor, HandlesRemovalFromEmptySafeGuard) +{ + const THeuristicPredictor predictor; + + TPredictionParams params; + params.PartitionsWeight = TPartitionWeight{0}; + params.Increasing = false; + params.TotalPartitions = 0; + + auto result = predictor.PredictLoadFactor(params); + + EXPECT_EQ(result.GetUnderlying(), 0); +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NCoordinator::NInfra diff --git a/src/infra/serializer/serializer_ut.cpp b/src/infra/serializer/serializer_ut.cpp index 5a9ff63..c242792 100644 --- a/src/infra/serializer/serializer_ut.cpp +++ b/src/infra/serializer/serializer_ut.cpp @@ -33,6 +33,10 @@ TPartitionMap MakePartitionMap() } // anonymous namespace +namespace NCoordinator::NInfra { + +//////////////////////////////////////////////////////////////////////////////// + TEST(PartitionMapSerializer, SerializeProducesValidJson) { const auto map = MakePartitionMap(); @@ -124,3 +128,7 @@ TEST(HubReportDeserializer, HandlesEmptyPartitions) EXPECT_TRUE(report.PartitionWeights.empty()); } + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NCoordinator::NInfra diff --git a/src/main.cpp b/src/main.cpp index c18ca38..18f31de 100644 --- a/src/main.cpp +++ b/src/main.cpp @@ -19,6 +19,7 @@ int main(int argc, char* argv[]) { NCoordinator::NInfra::NComponents::RegisterYdbComponents(component_list); NCoordinator::NInfra::NComponents::RegisterInfraComponents(component_list); + NCoordinator::NInfra::NComponents::RegisterServices(component_list); NCoordinator::NInfra::NComponents::RegisterHandlers(component_list); return userver::utils::DaemonMain(argc, argv, component_list); From a26a3143b98669c3bb3b6d8e268f2f70fde100c8 Mon Sep 17 00:00:00 2001 From: Ilya Repin Date: Sat, 31 Jan 2026 11:14:27 +0000 Subject: [PATCH 2/2] fix --- configs/static_config.yaml | 10 ++++++++++ .../use_cases/leader/coordination/coordination.cpp | 11 +++++++++-- 2 files changed, 19 insertions(+), 2 deletions(-) diff --git a/configs/static_config.yaml b/configs/static_config.yaml index 3a0ab13..099e0da 100644 --- a/configs/static_config.yaml +++ b/configs/static_config.yaml @@ -43,6 +43,7 @@ components_manager: fs-task-processor: fs-task-processor dns-client: + load-enabled: true fs-task-processor: fs-task-processor tests-control: @@ -52,6 +53,7 @@ components_manager: task_processor: main-task-processor ydb: + load-enabled: true operation-settings: cancel-after: 1000ms client-timeout: 1100ms @@ -64,9 +66,11 @@ components_manager: min_pool_size: 5 coordination-repository: + load-enabled: true dbname: chat_db coordination-gateway: + load-enabled: true dbname: chat_db coordination-node: chat-coordination partition-map-semaphore: partition-map-lock @@ -74,19 +78,24 @@ components_manager: initial-setup: true hub-gateway: + load-enabled: true load-factor-predictor: + load-enabled: true type: heuristic leader-dist-lock: + load-enabled: true database-settings: dbname: chat_db coordination-node: chat-coordination semaphore-name: partition-map-lock leader-service: + load-enabled: true handler-ping: + load-enabled: true path: /ping method: GET task_processor: main-task-processor @@ -94,6 +103,7 @@ components_manager: url_trailing_slash: strict-match handler-hello: # Finally! Our handler. + load-enabled: true path: /hello # Registering handler by URL '/hello'. method: GET,POST # It will only reply to GET (HEAD) and POST requests. task_processor: main-task-processor # Run it on CPU bound task processor \ No newline at end of file diff --git a/src/app/use_cases/leader/coordination/coordination.cpp b/src/app/use_cases/leader/coordination/coordination.cpp index a1cbca7..5aca6d1 100644 --- a/src/app/use_cases/leader/coordination/coordination.cpp +++ b/src/app/use_cases/leader/coordination/coordination.cpp @@ -2,6 +2,8 @@ #include +#include + namespace NCoordinator::NApp::NUseCase { //////////////////////////////////////////////////////////////////////////////// @@ -22,8 +24,13 @@ void TCoordinationUseCase::Execute(const NDto::TCoordinationRequest& request) co auto hubDiscovery = CoordinationGateway_.GetHubDiscovery(); auto hubReports = HubGateway_.GetHubReports(hubDiscovery); - auto partitionMap = CoordinationGateway_.GetPartitionMap().value(); - + auto partitionMapOpt = CoordinationGateway_.GetPartitionMap(); + if (!partitionMapOpt.has_value()) { + LOG_ERROR() << "Failed to get partition map"; + return; + } + auto partitionMap = partitionMapOpt.value(); + auto coordinationContext = CoordinationRepository_.GetCoordinationContext(); auto coordinationState = NCore::NDomain::TCoordinationState(