Add leader service

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,29 +66,44 @@ 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
             discovery-semaphore: discovery-lock
             initial-setup: true
 
-        coordination-dist-lock:
+        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
             throttling_enabled: false
             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

src/app/dto/leader/coordination.hpp

@@ -0,0 +1,17 @@
+#pragma once
+
+#include <core/coordination/coordination_settings.hpp>
+#include <core/partition_balancing/balancing_settings.hpp>
+
+namespace NCoordinator::NApp::NDto {
+
+////////////////////////////////////////////////////////////////////////////////
+
+struct TCoordinationRequest {
+    NCore::NDomain::TStateBuildingSettings StateBuildingSettings;
+    NCore::TBalancingSettings BalancingSettings;
+};
+
+////////////////////////////////////////////////////////////////////////////////
+
+} // namespace NCoordinator::NApp::NDto

src/app/services/leader/leader_service.cpp

@@ -0,0 +1,24 @@
+#include "leader_service.hpp"
+
+#include <userver/logging/log.hpp>
+
+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

src/app/services/leader/leader_service.hpp

@@ -0,0 +1,30 @@
+#pragma once
+
+#include <app/dto/leader/coordination.hpp>
+#include <app/use_cases/leader/coordination/coordination.hpp>
+#include <core/coordination/coordination_gateway.hpp>
+#include <core/coordination/coordination_repository.hpp>
+#include <core/hub/hub_gateway.hpp>
+
+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

src/app/use_cases/leader/coordination/coordination.cpp

@@ -0,0 +1,50 @@
+#include "coordination.hpp"
+
+#include <core/coordination/coordination_state.hpp>
+
+#include <userver/logging/log.hpp>
+
+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 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(
+        partitionMap,
+        hubReports,
+        coordinationContext,
+        request.StateBuildingSettings);
+
+    auto balancingResult = Balancer_.BalancePartitions(coordinationState, request.BalancingSettings);
+
+    CoordinationGateway_.BroadcastPartitionMap(balancingResult.PartitionMap);
+    CoordinationRepository_.SetCoordinationContext(balancingResult.Context);
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
+} // namespace NCoordinator::NApp::NUseCase

src/app/use_cases/leader/coordination/coordination.hpp

@@ -0,0 +1,35 @@
+#pragma once
+
+#include <app/dto/leader/coordination.hpp>
+#include <core/coordination/coordination_gateway.hpp>
+#include <core/coordination/coordination_repository.hpp>
+#include <core/partition_balancing/partition_balancer.hpp>
+#include <core/partition_balancing/load_factor_predictor.hpp>
+#include <core/hub/hub_gateway.hpp>
+
+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

src/core/coordination/coordination_state.cpp

@@ -91,8 +91,8 @@ void TCoordinationState::ApplyClusterSnapshot(
 {
     std::unordered_map<THubEndpoint, TPartitionWeight> 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;

src/core/coordination/coordination_state.hpp

@@ -19,7 +19,7 @@ namespace NCoordinator::NCore::NDomain {
 
 class TCoordinationState {
 public:
-    using TClusterSnapshot = std::unordered_map<THubEndpoint, THubReport>;
+    using TClusterSnapshot = std::vector<THubReport>;
     using TPartitionStates = std::unordered_map<TPartitionId, TPartitionState>;
     using THubStates = std::unordered_map<THubEndpoint, THubState>;
 

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(),