Add dynconfig and get partition

configs/config_vars.testing.yaml

@@ -4,4 +4,7 @@ logger-level: debug
 
 is-testing: true
 
-server-port: 8080
+server-port: 8080
+
+config-cache: ~/cache/
+config-server-url: http://localhost:8083

configs/config_vars.yaml

@@ -4,4 +4,7 @@ logger-level: info
 
 is-testing: false
 
-server-port: 8080
+server-port: 8080
+
+config-cache: ~/cache/
+config-server-url: http://localhost:8083

configs/static_config.yaml

@@ -24,13 +24,23 @@ components_manager:
                     level: $logger-level
                     overflow_behavior: discard  # Drop logs if the system is too busy to write them down.
 
-        # Dynamic config options. Cache is disabled, updates are disabled.
         dynamic-config:
-            # For most of userver dynamic configs, defaults are used, some are overridden here.
-            # See userver "dynamic config" docs for what configs exist.
-            defaults:
-                HTTP_CLIENT_CONNECTION_POOL_SIZE: 1000
-
+            updates-enabled: true
+            fs-cache-path: $config-cache
+            fs-task-processor: fs-task-processor
+
+        dynamic-config-client:
+            config-url: $config-server-url
+            http-retries: 5
+            http-timeout: 20s
+            service-name: chat
+
+        dynamic-config-client-updater:
+            config-settings: false
+            first-update-fail-ok: true
+            full-update-interval: 1m
+            update-interval: 5s
+
         default-secdist-provider:
             config: secure_data.json
 
@@ -119,6 +129,12 @@ components_manager:
             method: GET
             task_processor: main-task-processor
 
+        handler-get-partition:
+            load-enabled: true
+            path: /admin/partition
+            method: GET
+            task_processor: main-task-processor
+
         handler-get-hub-reports:
             load-enabled: true
             path: /admin/hub_reports

src/api/http/v1/admin/get_partition.cpp

@@ -0,0 +1,47 @@
+#include "get_partition.hpp"
+
+#include <app/dto/admin/get_partition.hpp>
+
+#include <infra/components/admin/admin_service_component.hpp>
+#include <infra/serializer/serializer.hpp>
+
+#include <api/http/exceptions/handler_exceptions.hpp>
+
+#include <userver/logging/log.hpp>
+
+namespace NCoordinator::NApi::NHandlers {
+
+////////////////////////////////////////////////////////////////////////////////
+
+TGetPartitionHandler::TGetPartitionHandler(
+    const userver::components::ComponentConfig& config,
+    const userver::components::ComponentContext& context)
+    : HttpHandlerJsonBase(config, context)
+    , AdminService_(context.FindComponent<NInfra::NComponents::TAdminServiceComponent>().GetService())
+{ }
+
+userver::formats::json::Value TGetPartitionHandler::HandleRequestJsonThrow(
+    const userver::server::http::HttpRequest& request,
+    const userver::formats::json::Value& /*request_json*/,
+    userver::server::request::RequestContext& /*request_context*/) const
+{
+    NApp::NDto::TGetPartitionResponse result;
+
+    const auto& channel = request.GetPathArg("channel");
+
+    try {
+        result = AdminService_.GetPartition(NApp::NDto::TGetPartitionRequest{channel});
+    } catch (const NApp::NUseCase::TGetPartitionTemporaryUnavailable& ex) {
+        LOG_ERROR() << "Get partition unavailable: " << ex.what();
+        throw TServerException("Get partition temporary unavailable");
+    }
+
+    userver::formats::json::ValueBuilder builder;
+    builder["partition"] = result.Partition.GetUnderlying();
+
+    return builder.ExtractValue();
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
+}  // namespace NChat::NApi::NHandlers

src/api/http/v1/admin/get_partition.hpp

@@ -0,0 +1,34 @@
+#pragma once
+
+#include <app/services/admin/admin_service.hpp>
+
+#include <userver/components/component_context.hpp>
+#include <userver/server/handlers/http_handler_json_base.hpp>
+#include <userver/ydb/component.hpp>
+
+namespace NCoordinator::NApi::NHandlers {
+
+////////////////////////////////////////////////////////////////////////////////
+
+class TGetPartitionHandler final
+    : public userver::server::handlers::HttpHandlerJsonBase
+{
+public:
+    static constexpr std::string_view kName = "handler-get-partition";
+
+    TGetPartitionHandler(
+        const userver::components::ComponentConfig&,
+        const userver::components::ComponentContext&);
+
+    userver::formats::json::Value HandleRequestJsonThrow(
+        const userver::server::http::HttpRequest& request,
+        const userver::formats::json::Value& request_json,
+        userver::server::request::RequestContext& context) const override;
+
+private:
+    NApp::NService::TAdminService& AdminService_;
+};
+
+////////////////////////////////////////////////////////////////////////////////
+
+}  // namespace NChat::NApi::NHandlers

src/app/dto/admin/get_partition.hpp

@@ -0,0 +1,19 @@
+#pragma once
+
+#include <core/common/partition_params.hpp>
+
+namespace NCoordinator::NApp::NDto {
+
+////////////////////////////////////////////////////////////////////////////////
+
+struct TGetPartitionRequest {
+    std::string ChannelId;
+};
+
+struct TGetPartitionResponse {
+    NCore::NDomain::TPartitionId Partition;
+};
+
+////////////////////////////////////////////////////////////////////////////////
+
+} // namespace NCoordinator::NApp::NDto

src/app/dto/leader/coordination.hpp

@@ -8,6 +8,7 @@ namespace NCoordinator::NApp::NDto {
 ////////////////////////////////////////////////////////////////////////////////
 
 struct TCoordinationRequest {
+    std::size_t DefaultPartitionsAmount;
     NCore::NDomain::TStateBuildingSettings StateBuildingSettings;
     NCore::TBalancingSettings BalancingSettings;
 };

src/app/services/admin/admin_service.cpp

@@ -10,6 +10,7 @@ TAdminService::TAdminService(
     NCore::NDomain::IHubGateway& hubGateway)
     : GetContextUseCase_(coordinationRepository)
     , GetPartitionMapUseCase_(coordinationGateway)
+    , GetPartitionUseCase_(coordinationGateway)
     , GetHubReportsUseCase_(coordinationGateway, hubGateway)
 { }
 
@@ -23,6 +24,11 @@ NDto::TGetPartitionMapResponse TAdminService::GetPartitionMap() const
     return GetPartitionMapUseCase_.Execute();
 }
 
+NDto::TGetPartitionResponse TAdminService::GetPartition(const NDto::TGetPartitionRequest& request) const
+{
+    return GetPartitionUseCase_.Execute(request);
+}
+
 NDto::TGetHubReportsResponse TAdminService::GetHubReports() const
 {
     return GetHubReportsUseCase_.Execute();

src/app/services/admin/admin_service.hpp

@@ -3,8 +3,10 @@
 #include <app/dto/admin/get_context.hpp>
 #include <app/dto/admin/get_hub_reports.hpp>
 #include <app/dto/admin/get_partition_map.hpp>
+#include <app/dto/admin/get_partition.hpp>
 #include <app/use_cases/admin/get_context/get_context.hpp>
 #include <app/use_cases/admin/get_partition_map/get_partition_map.hpp>
+#include <app/use_cases/admin/get_partition/get_partition.hpp>
 #include <app/use_cases/admin/get_hub_reports/get_hub_reports.hpp>
 #include <core/coordination/coordination_gateway.hpp>
 #include <core/coordination/coordination_repository.hpp>
@@ -24,11 +26,13 @@ class TAdminService final
 
    NDto::TGetContextResponse GetCoordinationContext() const;
    NDto::TGetPartitionMapResponse GetPartitionMap() const;
+   NDto::TGetPartitionResponse GetPartition(const NDto::TGetPartitionRequest& request) const;
    NDto::TGetHubReportsResponse GetHubReports() const;
 
 private:
     NUseCase::TGetContextUseCase GetContextUseCase_;
     NUseCase::TGetPartitionMapUseCase GetPartitionMapUseCase_;
+    NUseCase::TGetPartitionUseCase GetPartitionUseCase_;
     NUseCase::TGetHubReportsUseCase GetHubReportsUseCase_;
 };
 

src/app/use_cases/admin/get_partition/get_partition.cpp

@@ -0,0 +1,49 @@
+#include "get_partition.hpp"
+
+#include <core/hash_ring/hash_ring.hpp>
+#include <core/partition/partition_map.hpp>
+
+#include <userver/logging/log.hpp>
+
+#include <fmt/format.h>
+
+#include <vector>
+
+namespace NCoordinator::NApp::NUseCase {
+
+////////////////////////////////////////////////////////////////////////////////
+
+TGetPartitionUseCase::TGetPartitionUseCase(NCore::NDomain::ICoordinationGateway& coordinationGateway)
+    : CoordinationGateway_(coordinationGateway)
+{ }
+
+NDto::TGetPartitionResponse TGetPartitionUseCase::Execute(const NDto::TGetPartitionRequest& request) const
+{
+    NCore::NDomain::TPartitionMap partitionMap;
+
+    try {
+        partitionMap = CoordinationGateway_.GetPartitionMap();
+    } catch (std::exception& ex) {
+        throw TGetPartitionTemporaryUnavailable(fmt::format("Failed to get partition map: {}", ex.what()));
+    }
+
+    std::vector<NCore::NDomain::TPartitionId> partitions;
+    partitions.reserve(partitionMap.Partitions.size());
+
+    for (const auto& [partition, hub] : partitionMap.Partitions) {
+        partitions.push_back(partition);
+    }
+
+    NCore::NDomain::THashRing hashRing(partitions);
+    auto partition = hashRing.GetPartition(request.ChannelId);
+
+    NDto::TGetPartitionResponse response{
+        .Partition = partition,
+    };
+
+    return response;
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
+} // namespace NCoordinator::NApp::NUseCase

src/app/use_cases/admin/get_partition/get_partition.hpp

@@ -0,0 +1,30 @@
+#pragma once
+
+#include <app/exceptions.hpp>
+#include <app/dto/admin/get_partition.hpp>
+#include <core/coordination/coordination_gateway.hpp>
+
+namespace NCoordinator::NApp::NUseCase {
+
+////////////////////////////////////////////////////////////////////////////////
+
+class TGetPartitionTemporaryUnavailable
+    : public TApplicationException
+{
+  using TApplicationException::TApplicationException;
+};
+
+class TGetPartitionUseCase final
+{
+public:
+   TGetPartitionUseCase(NCore::NDomain::ICoordinationGateway& coordinationGateway);
+
+   NDto::TGetPartitionResponse Execute(const NDto::TGetPartitionRequest& request) const;
+
+private:
+    NCore::NDomain::ICoordinationGateway& CoordinationGateway_;
+};
+
+////////////////////////////////////////////////////////////////////////////////
+
+} // namespace NCoordinator::NApp::NUseCase

src/app/use_cases/admin/get_partition_map/get_partition_map.cpp

@@ -1,7 +1,5 @@
 #include "get_partition_map.hpp"
 
-#include <core/coordination/coordination_state.hpp>
-
 #include <userver/logging/log.hpp>
 
 #include <fmt/format.h>

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

@@ -1,6 +1,7 @@
 #include "coordination.hpp"
 
 #include <core/coordination/coordination_state.hpp>
+#include <core/partition/partition_map.hpp>
 
 #include <userver/logging/log.hpp>
 
@@ -33,6 +34,11 @@ void TCoordinationUseCase::Execute(const NDto::TCoordinationRequest& request) co
         return;
     }
 
+    if (partitionMap.Partitions.empty()) {
+        auto startingPartitionMap = NCore::NDomain::BuildStartingPartitionMap(request.DefaultPartitionsAmount);
+        std::swap(partitionMap.Partitions, startingPartitionMap.Partitions);
+    }
+
     auto coordinationContext = CoordinationRepository_.GetCoordinationContext();
 
     auto coordinationState = NCore::NDomain::TCoordinationState(

src/core/common/hub_params.hpp

@@ -21,7 +21,7 @@ using THubDC = userver::utils::StrongTypedef<
 
 using TLoadFactor = userver::utils::StrongTypedef<
     struct TLoadFactorTag,
-    std::uint64_t,
+    std::uint32_t,
     userver::utils::StrongTypedefOps::kCompareTransparent>;
 
 enum class EHubStatus {

src/core/hash_ring/hash_ring.cpp

@@ -13,6 +13,31 @@ THashRing::THashRing(const std::vector<TPartitionId>& partitions, HashFunction h
     LoadPartitions(partitions);
 }
 
+THashRing::THashRing(std::size_t partitionCount, HashFunction hasher)
+    : Hasher_(std::move(hasher))
+{
+    if (partitionCount == 0) {
+        throw std::invalid_argument("Partition count must be at least 1");
+    }
+
+    Partitions_.reserve(partitionCount);
+
+    const std::uint64_t maxVal = std::numeric_limits<std::uint64_t>::max();
+    const std::uint64_t step = maxVal / partitionCount; 
+
+    for (std::size_t i = 1; i <= partitionCount; ++i) {
+        std::uint64_t boundary;
+
+        if (i == partitionCount) {
+            boundary = maxVal;
+        } else {
+            boundary = step * i;
+        }
+
+        Partitions_.emplace_back(boundary);
+    }
+}
+
 void THashRing::LoadPartitions(const std::vector<TPartitionId>& partitions)
 {
     if (partitions.empty()) {
@@ -25,7 +50,7 @@ void THashRing::LoadPartitions(const std::vector<TPartitionId>& partitions)
 
 TPartitionId THashRing::GetPartition(const std::string& key) const
 {
-    const uint64_t hash = Hasher_(key);
+    const std::uint64_t hash = Hasher_(key);
 
     auto it = std::lower_bound(Partitions_.begin(), Partitions_.end(), hash);
 

src/core/hash_ring/hash_ring.hpp

@@ -16,6 +16,7 @@ class THashRing {
     using HashFunction = std::function<std::uint64_t(const std::string&)>;
 
     explicit THashRing(const std::vector<TPartitionId>& partitions, HashFunction hasher = std::hash<std::string>{});
+    explicit THashRing(std::size_t partitionCount, HashFunction hasher = std::hash<std::string>{});
 
     void LoadPartitions(const std::vector<TPartitionId>& partitions);