From b1b49a18ead785f7afaf64b11c72fb71cfc9158f Mon Sep 17 00:00:00 2001 From: Ilya Repin Date: Mon, 19 Jan 2026 11:06:58 +0000 Subject: [PATCH 1/5] add dist lock --- CMakeLists.txt | 17 +++- CMakePresets.json | 2 + configs/secure_data.json | 1 + configs/static_config.yaml | 27 ++++++ requirements.txt | 3 +- src/core/common/coordination_params.hpp | 14 +++ src/core/common/hub_params.hpp | 34 +++++++ src/core/common/partition_params.hpp | 29 ++++++ .../coordination/coordination_context.hpp | 15 ++++ src/core/coordination_gateway.hpp | 19 ++++ src/core/coordination_repository.hpp | 13 +++ src/core/partitions/partition_map.hpp | 17 ++++ src/hello.cpp | 29 +++++- src/hello.hpp | 28 ++++-- src/infra/components/components.cpp | 51 +++++++++++ src/infra/components/components.hpp | 22 +++++ .../coordination_dist_lock_component.cpp | 54 +++++++++++ .../coordination_dist_lock_component.hpp | 31 +++++++ .../coordination_gateway_component.cpp | 65 ++++++++++++++ .../coordination_gateway_component.hpp | 33 +++++++ .../coordination_service_component.cpp | 0 .../coordination_service_component.hpp | 0 src/infra/kesus_gateway/kesus_gateway.cpp | 90 +++++++++++++++++++ src/infra/kesus_gateway/kesus_gateway.hpp | 39 ++++++++ src/infra/serializer/serializer.cpp | 67 ++++++++++++++ src/infra/serializer/serializer.hpp | 20 +++++ src/main.cpp | 22 ++--- tests/conftest.py | 6 +- tests/requirements.txt | 1 + tests/test_basic.py | 5 +- 30 files changed, 721 insertions(+), 33 deletions(-) create mode 100644 configs/secure_data.json create mode 100644 src/core/common/coordination_params.hpp create mode 100644 src/core/common/hub_params.hpp create mode 100644 src/core/common/partition_params.hpp create mode 100644 src/core/coordination/coordination_context.hpp create mode 100644 src/core/coordination_gateway.hpp create mode 100644 src/core/coordination_repository.hpp create mode 100644 src/core/partitions/partition_map.hpp create mode 100644 src/infra/components/components.cpp create mode 100644 src/infra/components/components.hpp create mode 100644 src/infra/components/coordination/coordination_dist_lock_component.cpp create mode 100644 src/infra/components/coordination/coordination_dist_lock_component.hpp create mode 100644 src/infra/components/coordination/coordination_gateway_component.cpp create mode 100644 src/infra/components/coordination/coordination_gateway_component.hpp create mode 100644 src/infra/components/coordination/coordination_service_component.cpp create mode 100644 src/infra/components/coordination/coordination_service_component.hpp create mode 100644 src/infra/kesus_gateway/kesus_gateway.cpp create mode 100644 src/infra/kesus_gateway/kesus_gateway.hpp create mode 100644 src/infra/serializer/serializer.cpp create mode 100644 src/infra/serializer/serializer.hpp create mode 100644 tests/requirements.txt diff --git a/CMakeLists.txt b/CMakeLists.txt index 9c83786..a6d2918 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -17,16 +17,23 @@ endif() userver_setup_environment() # Common sources + +file(GLOB_RECURSE SOURCES ${CMAKE_CURRENT_SOURCE_DIR}/src/*.[hc]pp) +file(GLOB_RECURSE UNIT_TEST_SOURCES ${CMAKE_CURRENT_SOURCE_DIR}/src/*_test.cpp) +file(GLOB_RECURSE BENCHMARK_SOURCES ${CMAKE_CURRENT_SOURCE_DIR}/src/*_benchmark.cpp) +file(GLOB_RECURSE LIBUBENCH_SOURCES ${CMAKE_CURRENT_SOURCE_DIR}/src/benchmarks_main.cpp) +list(REMOVE_ITEM SOURCES ${BENCH_SOURCES} ${LIBUBENCH_SOURCES}) +list(REMOVE_ITEM SOURCES ${UNIT_TEST_SOURCES} ${BENCHMARK_SOURCES} ${LIBUBENCH_SOURCES} ${CMAKE_CURRENT_SOURCE_DIR}/src/main.cpp) + include_directories(src) add_library( ${PROJECT_NAME}_objs OBJECT - src/greeting.cpp # - src/hello.cpp # + ${SOURCES} ) target_link_libraries( ${PROJECT_NAME}_objs - PUBLIC userver::core # + PUBLIC userver::core userver::ydb ) @@ -45,7 +52,9 @@ target_link_libraries(${PROJECT_NAME}_benchmark PRIVATE ${PROJECT_NAME}_objs use add_google_benchmark_tests(${PROJECT_NAME}_benchmark) # Functional testing -userver_testsuite_add_simple() +userver_testsuite_add_simple( + REQUIREMENTS ${CMAKE_CURRENT_SOURCE_DIR}/tests/requirements.txt +) # Install include(GNUInstallDirs) diff --git a/CMakePresets.json b/CMakePresets.json index b2c3da4..6b06f8c 100644 --- a/CMakePresets.json +++ b/CMakePresets.json @@ -36,6 +36,8 @@ "hidden": true, "generator": "Ninja", "cacheVariables": { + "USERVER_FEATURE_GRPC": "ON", + "USERVER_FEATURE_YDB": "ON", "CMAKE_EXPORT_COMPILE_COMMANDS": "ON" } } diff --git a/configs/secure_data.json b/configs/secure_data.json new file mode 100644 index 0000000..9e26dfe --- /dev/null +++ b/configs/secure_data.json @@ -0,0 +1 @@ +{} \ No newline at end of file diff --git a/configs/static_config.yaml b/configs/static_config.yaml index f4f6606..a0f17cf 100644 --- a/configs/static_config.yaml +++ b/configs/static_config.yaml @@ -29,6 +29,9 @@ components_manager: # See userver "dynamic config" docs for what configs exist. defaults: HTTP_CLIENT_CONNECTION_POOL_SIZE: 1000 + + default-secdist-provider: + config: secure_data.json testsuite-support: {} @@ -48,6 +51,30 @@ components_manager: method: POST task_processor: main-task-processor + ydb: + operation-settings: + cancel-after: 1000ms + client-timeout: 1100ms + operation-timeout: 1000ms + retries: 3 + databases: + db: + database: chat_db + max_pool_size: 10 + min_pool_size: 5 + + coordination-gateway: + dbname: chat_db + coordination-node: chat-coordination + semaphore-name: partition-map-lock + initial-setup: true + + coordination-dist-lock: + database-settings: + dbname: chat_db + coordination-node: chat-coordination + semaphore-name: partition-map-lock + handler-ping: path: /ping method: GET diff --git a/requirements.txt b/requirements.txt index a8e7b5b..e64445f 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1 +1,2 @@ -pycodestyle \ No newline at end of file +pycodestyle +ydb >= 3.5.2 \ No newline at end of file diff --git a/src/core/common/coordination_params.hpp b/src/core/common/coordination_params.hpp new file mode 100644 index 0000000..4172bce --- /dev/null +++ b/src/core/common/coordination_params.hpp @@ -0,0 +1,14 @@ +#pragma once + +#include + +#include + +namespace NCoordinator::NCore::NDomain { + +using TEpoch = userver::utils::StrongTypedef< + struct TEpochTag, + std::uint64_t, + userver::utils::StrongTypedefOps::kCompareTransparent>; + +} // namespace NCoordinator::NCore::NDomain diff --git a/src/core/common/hub_params.hpp b/src/core/common/hub_params.hpp new file mode 100644 index 0000000..0bc2def --- /dev/null +++ b/src/core/common/hub_params.hpp @@ -0,0 +1,34 @@ +#pragma once + +#include + +#include +#include + +namespace NCoordinator::NCore::NDomain { + +using THubEndpoint = userver::utils::StrongTypedef< + struct THubEndpointTag, + std::string, + userver::utils::StrongTypedefOps::kCompareStrong>; + +using THubDC = userver::utils::StrongTypedef< + struct THubDCTag, + std::string, + userver::utils::StrongTypedefOps::kCompareStrong>; + +using TLoadFactor = userver::utils::StrongTypedef< + struct TLoadFactorTag, + std::uint64_t, + userver::utils::StrongTypedefOps::kCompareTransparent>; + +enum class EHubStatus { + HEALTHY, + DRAINING, + OVERLOADED, + // SUSPECTED, + // UNHEALTHY, + LAGGED, +}; + +} // namespace NCoordinator::NCore::NDomain diff --git a/src/core/common/partition_params.hpp b/src/core/common/partition_params.hpp new file mode 100644 index 0000000..e96166e --- /dev/null +++ b/src/core/common/partition_params.hpp @@ -0,0 +1,29 @@ +#pragma once + +#include + +#include + +namespace NCoordinator::NCore::NDomain { + +using TPartitionId = userver::utils::StrongTypedef< + struct TPartitionIdTag, + std::uint64_t, + userver::utils::StrongTypedefOps::kCompareTransparent>; + +using TPartitionLoad = userver::utils::StrongTypedef< + struct TPartitionLoadTag, + std::uint64_t, + userver::utils::StrongTypedefOps::kCompareTransparent>; + +using TPartitionImpact = userver::utils::StrongTypedef< + struct TPartitionLoadTag, + double, + userver::utils::StrongTypedefOps::kCompareTransparent>; + +enum class EPartitionStatus { + ACTIVE, + MIGRATED, +}; + +} // namespace NCoordinator::NCore::NDomain diff --git a/src/core/coordination/coordination_context.hpp b/src/core/coordination/coordination_context.hpp new file mode 100644 index 0000000..a18685f --- /dev/null +++ b/src/core/coordination/coordination_context.hpp @@ -0,0 +1,15 @@ +#pragma once + +#include +#include + +// TODO to delete + +namespace NCoordinator::NCore::NDomain { + +struct TCoordinationContext { + std::unordered_map PartitionFreezeEpoch; + std::unordered_map PartitionLoads; +}; + +} \ No newline at end of file diff --git a/src/core/coordination_gateway.hpp b/src/core/coordination_gateway.hpp new file mode 100644 index 0000000..e71fdec --- /dev/null +++ b/src/core/coordination_gateway.hpp @@ -0,0 +1,19 @@ +#pragma once + +#include + +namespace NCoordinator::NCore { + +//////////////////////////////////////////////////////////////////////////////// + +class ICoordinationGateway { + public: + virtual NDomain::TPartitionMap GetPartitionMap() const = 0; + virtual void BroadcastPartitionMap(const NDomain::TPartitionMap& partitionMap) const = 0; + + virtual ~ICoordinationGateway() = default; +}; + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NCoordinator::NCore diff --git a/src/core/coordination_repository.hpp b/src/core/coordination_repository.hpp new file mode 100644 index 0000000..e96d07d --- /dev/null +++ b/src/core/coordination_repository.hpp @@ -0,0 +1,13 @@ +#pragma once + +namespace NCoordinator::NCore { + +// class ICoordinationrRepository { +// public: +// virtual TCoordinationContext GetCoordinationContext() const = 0; +// virtual void SetCoordinationContext(const TCoordinationContext& context) const = 0; + +// virtual ~ICoordinationrRepository() = default; +// }; + +} // namespace NCoordinator::NCore diff --git a/src/core/partitions/partition_map.hpp b/src/core/partitions/partition_map.hpp new file mode 100644 index 0000000..a1b655b --- /dev/null +++ b/src/core/partitions/partition_map.hpp @@ -0,0 +1,17 @@ +#pragma once + +#include +#include +#include + +#include +#include +// TODO to delete +namespace NCoordinator::NCore::NDomain { + +struct TPartitionMap { + TEpoch Epoch; + std::vector> Partitions; +}; // TODO Delete + +} \ No newline at end of file diff --git a/src/hello.cpp b/src/hello.cpp index c6cf79d..4337c77 100644 --- a/src/hello.cpp +++ b/src/hello.cpp @@ -2,12 +2,33 @@ #include +#include + +#include +#include + +#include + namespace coordinator { -std::string -Hello::HandleRequestThrow(const userver::server::http::HttpRequest& request, userver::server::request::RequestContext&) - const { - return SayHelloTo(request.GetArg("name"), UserType::kFirstTime); +Hello::Hello(const userver::components::ComponentConfig& config, const userver::components::ComponentContext& context) + : HttpHandlerJsonBase(config, context), + ydb_client_(context.FindComponent().GetCoordinationClient("chat_db")) + , dist_(context.FindComponent()) +{ + gateway = std::make_unique( + ydb_client_, "chat-coordination", "partition-map-lock", false); +} + +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(); + + return NCoordinator::NInfra::SerializePartitionMap(result); } } // namespace coordinator \ No newline at end of file diff --git a/src/hello.hpp b/src/hello.hpp index cb47519..396b80c 100644 --- a/src/hello.hpp +++ b/src/hello.hpp @@ -1,17 +1,35 @@ #pragma once -#include +#include +#include +#include +#include +#include + +#include +#include +#include + namespace coordinator { -class Hello final : public userver::server::handlers::HttpHandlerBase { +class Hello final : public userver::server::handlers::HttpHandlerJsonBase { public: static constexpr std::string_view kName = "handler-hello"; - using HttpHandlerBase::HttpHandlerBase; + Hello(const userver::components::ComponentConfig& config, const userver::components::ComponentContext& context); + + 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; - std::string HandleRequestThrow(const userver::server::http::HttpRequest&, userver::server::request::RequestContext&) - const override; +private: + NYdb::NCoordination::TSessionSettings settings; + std::shared_ptr ydb_client_; + std::unique_ptr gateway; + NCoordinator::NInfra::NComponents::TCoordinationDistLockComponent& dist_; }; } // namespace coordinator \ No newline at end of file diff --git a/src/infra/components/components.cpp b/src/infra/components/components.cpp new file mode 100644 index 0000000..c1e09a8 --- /dev/null +++ b/src/infra/components/components.cpp @@ -0,0 +1,51 @@ +#include "components.hpp" + +#include +#include + +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace NCoordinator::NInfra::NComponents { + +void RegisterUserverComponents(userver::components::ComponentList& list) { + list.Append() + .Append() + .AppendComponentList(userver::clients::http::ComponentList()) + .Append() + .Append() + .Append(); +} + +// Clients +void RegisterYdbComponent(userver::components::ComponentList& list) { + list.Append() + .Append() + .Append(); +} + +// Components +void RegisterServiceComponents(userver::components::ComponentList& list) { + list.Append() + .Append(); +} + +// Handlers +void RegisterHandlers(userver::components::ComponentList& list) { + list.Append(); +} + +} // namespace NCoordinator::NInfra::NComponents diff --git a/src/infra/components/components.hpp b/src/infra/components/components.hpp new file mode 100644 index 0000000..0273674 --- /dev/null +++ b/src/infra/components/components.hpp @@ -0,0 +1,22 @@ +#pragma once + +#include + +namespace NCoordinator::NInfra::NComponents { + +//////////////////////////////////////////////////////////////////////////////// + +void RegisterUserverComponents(userver::components::ComponentList&); + +// Clients +void RegisterYdbComponent(userver::components::ComponentList&); + +// Components +void RegisterServiceComponents(userver::components::ComponentList&); + +// Handlers +void RegisterHandlers(userver::components::ComponentList& list); + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NChat::NInfra::NComponents diff --git a/src/infra/components/coordination/coordination_dist_lock_component.cpp b/src/infra/components/coordination/coordination_dist_lock_component.cpp new file mode 100644 index 0000000..e381632 --- /dev/null +++ b/src/infra/components/coordination/coordination_dist_lock_component.cpp @@ -0,0 +1,54 @@ +#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() { + LOG_CRITICAL() << "DO WORK 1"; + + while (!userver::engine::current_task::ShouldCancel()) { + LOG_CRITICAL() << "DO WORK 2"; + auto partitionMap = Gateway_.GetPartitionMap(); + ++(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/coordination/coordination_dist_lock_component.hpp b/src/infra/components/coordination/coordination_dist_lock_component.hpp new file mode 100644 index 0000000..26bf787 --- /dev/null +++ b/src/infra/components/coordination/coordination_dist_lock_component.hpp @@ -0,0 +1,31 @@ +#pragma once + +#include + +#include + +namespace NCoordinator::NInfra::NComponents { + +//////////////////////////////////////////////////////////////////////////////// + +class TCoordinationDistLockComponent + : public userver::ydb::DistLockComponentBase +{ +public: + static constexpr std::string_view kName = "coordination-dist-lock"; + + TCoordinationDistLockComponent( + const userver::components::ComponentConfig& config, + const userver::components::ComponentContext& context); + + ~TCoordinationDistLockComponent(); + + void DoWork() final; + +private: + NCore::ICoordinationGateway& Gateway_; // Replace with service with core +}; + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NCoordinator::NInfra::NComponents diff --git a/src/infra/components/coordination/coordination_gateway_component.cpp b/src/infra/components/coordination/coordination_gateway_component.cpp new file mode 100644 index 0000000..65c0d86 --- /dev/null +++ b/src/infra/components/coordination/coordination_gateway_component.cpp @@ -0,0 +1,65 @@ +#include "coordination_gateway_component.hpp" + +#include + +#include +#include +#include +#include + +namespace NCoordinator::NInfra::NComponents { + +//////////////////////////////////////////////////////////////////////////////// + +TCoordinationGatewayComponent::TCoordinationGatewayComponent( + const userver::components::ComponentConfig& config, + const userver::components::ComponentContext& context) + : LoggableComponentBase(config, context) +{ + auto dbname = config["dbname"].As(); + auto coordinationNode = config["coordination-node"].As(); + auto semaphoreName = config["semaphore-name"].As(); + auto initialSetup = config["initial-setup"].As(true); + + auto coordinationClient = context.FindComponent().GetCoordinationClient(dbname); + + Gateway_ = std::make_unique( + std::move(coordinationClient), + coordinationNode, + semaphoreName, + initialSetup); +} + +NCore::ICoordinationGateway& TCoordinationGatewayComponent::GetGateway() +{ + return *Gateway_; +} + +userver::yaml_config::Schema TCoordinationGatewayComponent::GetStaticConfigSchema() { + return userver::yaml_config::MergeSchemas( + R"( +type: object +description: Component for coordination gateway +additionalProperties: false +properties: + dbname: + type: string + description: the key of the database within ydb component (NOT the + actual database path) + coordination-node: + type: string + description: name of the coordination node within the database + semaphore-name: + type: string + description: name of the semaphore within the coordination node + initial-setup: + type: boolean + description: if true, then create the coordination node and the semaphore + unless they already exist + default: true +)"); +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NCoordinator::NInfra::NComponents diff --git a/src/infra/components/coordination/coordination_gateway_component.hpp b/src/infra/components/coordination/coordination_gateway_component.hpp new file mode 100644 index 0000000..b21cfc6 --- /dev/null +++ b/src/infra/components/coordination/coordination_gateway_component.hpp @@ -0,0 +1,33 @@ +#pragma once + +#include + +#include + +#include + +namespace NCoordinator::NInfra::NComponents { + +//////////////////////////////////////////////////////////////////////////////// + +class TCoordinationGatewayComponent + : public userver::components::LoggableComponentBase +{ +public: + static constexpr std::string_view kName = "coordination-gateway"; + + TCoordinationGatewayComponent( + const userver::components::ComponentConfig& config, + const userver::components::ComponentContext& context); + + NCore::ICoordinationGateway& GetGateway(); + + static userver::yaml_config::Schema GetStaticConfigSchema(); + +private: + std::unique_ptr Gateway_; +}; + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NCoordinator::NInfra::NComponents diff --git a/src/infra/components/coordination/coordination_service_component.cpp b/src/infra/components/coordination/coordination_service_component.cpp new file mode 100644 index 0000000..e69de29 diff --git a/src/infra/components/coordination/coordination_service_component.hpp b/src/infra/components/coordination/coordination_service_component.hpp new file mode 100644 index 0000000..e69de29 diff --git a/src/infra/kesus_gateway/kesus_gateway.cpp b/src/infra/kesus_gateway/kesus_gateway.cpp new file mode 100644 index 0000000..2db4dd9 --- /dev/null +++ b/src/infra/kesus_gateway/kesus_gateway.cpp @@ -0,0 +1,90 @@ +#include "kesus_gateway.hpp" + +#include +#include +#include +#include + +#include +#include +#include + +#include + +namespace { + +//////////////////////////////////////////////////////////////////////////////// + +const std::uint64_t SEMAPHORE_LIMIT = 1; + +//////////////////////////////////////////////////////////////////////////////// + +} // anonymous namespace + +namespace NCoordinator::NInfra::NGateway { + +//////////////////////////////////////////////////////////////////////////////// + +TKesusGateway::TKesusGateway( + std::shared_ptr coordinationClient, + const std::string& coordinationNode, + const std::string& semaphoreName, + const bool initialSetup) + : SemaphoreName_(semaphoreName) + , CoordinationSession_(nullptr) +{ + if (initialSetup) { + InitialSetup(*coordinationClient, coordinationNode); + } + + auto session = coordinationClient->StartSession(coordinationNode, {}); + CoordinationSession_ = std::make_unique(std::move(session)); +} + +NCore::NDomain::TPartitionMap TKesusGateway::GetPartitionMap() const +{ + NYdb::NCoordination::TDescribeSemaphoreSettings describeSettings; + auto description = CoordinationSession_->DescribeSemaphore(SemaphoreName_, describeSettings); + + userver::formats::json::Value json; + try { + json = userver::formats::json::FromString(description.GetData()); + } catch(std::exception& ex) { + LOG_ERROR() << "Can't parse json: " << ex; + return {}; + } + + return DeserializePartitionMap(json); +} + +void TKesusGateway::BroadcastPartitionMap(const NCore::NDomain::TPartitionMap& partitionMap) const +{ + auto json = SerializePartitionMap(partitionMap); + CoordinationSession_->UpdateSemaphore( + SemaphoreName_, + userver::formats::json::ToStableString(json)); +} + +void TKesusGateway::InitialSetup( + userver::ydb::CoordinationClient& coordinationClient, + const std::string& coordinationNode) const +{ + NYdb::NCoordination::TCreateNodeSettings createNodeSettings; + try { + coordinationClient.CreateNode(coordinationNode, createNodeSettings); + } catch (const userver::ydb::YdbResponseError& ex) { + LOG_WARNING() << "Could not create coordination node: " << ex; + } + + auto session = coordinationClient.StartSession(coordinationNode, {}); + try { + session.CreateSemaphore(SemaphoreName_, SEMAPHORE_LIMIT); + } catch (const userver::ydb::YdbResponseError& ex) { + LOG_WARNING() << "Could not create semaphore: " << ex; + } + session.Close(); +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NCoordinator::NInfra::NGateway diff --git a/src/infra/kesus_gateway/kesus_gateway.hpp b/src/infra/kesus_gateway/kesus_gateway.hpp new file mode 100644 index 0000000..d61891b --- /dev/null +++ b/src/infra/kesus_gateway/kesus_gateway.hpp @@ -0,0 +1,39 @@ +#pragma once + +#include +#include + +#include + +#include + +namespace NCoordinator::NInfra::NGateway { + +//////////////////////////////////////////////////////////////////////////////// + +class TKesusGateway + : public NCore::ICoordinationGateway +{ +public: + TKesusGateway( + std::shared_ptr coordinationClient, + const std::string& coordinationNode, + const std::string& semaphoreName, + const bool initialSetup); + + NCore::NDomain::TPartitionMap GetPartitionMap() const override; + void BroadcastPartitionMap(const NCore::NDomain::TPartitionMap& partitionMap) const override; + +private: + void InitialSetup( + userver::ydb::CoordinationClient& coordinationClient, + const std::string& coordinationNode) const; + +private: + const std::string SemaphoreName_; + std::unique_ptr CoordinationSession_; +}; + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NCoordinator::NInfra::NGateway diff --git a/src/infra/serializer/serializer.cpp b/src/infra/serializer/serializer.cpp new file mode 100644 index 0000000..191caf9 --- /dev/null +++ b/src/infra/serializer/serializer.cpp @@ -0,0 +1,67 @@ +#include "serializer.hpp" + +#include +#include +#include + +#include + +namespace { + +//////////////////////////////////////////////////////////////////////////////// + +constexpr inline std::string_view EPOCH_KEY = "epoch"; +constexpr inline std::string_view PARTITIONS_KEY = "partitions"; +constexpr inline std::string_view PARTITION_ID_KEY = "id"; +constexpr inline std::string_view HUB_ENDPOINT_KEY = "hub"; + +//////////////////////////////////////////////////////////////////////////////// + +} // anonymous namespace + +namespace NCoordinator::NInfra { + +//////////////////////////////////////////////////////////////////////////////// + +userver::formats::json::Value SerializePartitionMap( + const NCore::NDomain::TPartitionMap& partitionMap) +{ + userver::formats::json::ValueBuilder result; + result[PARTITIONS_KEY.data()] = userver::formats::common::Type::kArray; + result[EPOCH_KEY.data()] = partitionMap.Epoch.GetUnderlying(); + + for (const auto& [partition, hub] : partitionMap.Partitions) { + userver::formats::json::ValueBuilder pair; + pair[PARTITION_ID_KEY.data()] = partition.GetUnderlying(); + pair[HUB_ENDPOINT_KEY.data()] = hub.GetUnderlying(); + + result[PARTITIONS_KEY.data()].PushBack(pair.ExtractValue()); + } + + return result.ExtractValue(); +} + +NCore::NDomain::TPartitionMap DeserializePartitionMap(const userver::formats::json::Value& jsonValue) +{ + NCore::NDomain::TPartitionMap result; + result.Epoch = NCore::NDomain::TEpoch{ + jsonValue[EPOCH_KEY.data()].As() + }; + + for (const auto& pair : jsonValue[PARTITIONS_KEY.data()]) { + NCore::NDomain::TPartitionId partitionId{ + pair[PARTITION_ID_KEY.data()].As() + }; + NCore::NDomain::THubEndpoint hubEndpoint{ + pair[HUB_ENDPOINT_KEY.data()].As() + }; + + result.Partitions.emplace_back(std::move(partitionId), std::move(hubEndpoint)); + } + + return result; +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NCoordinator::NInfra diff --git a/src/infra/serializer/serializer.hpp b/src/infra/serializer/serializer.hpp new file mode 100644 index 0000000..78d3942 --- /dev/null +++ b/src/infra/serializer/serializer.hpp @@ -0,0 +1,20 @@ +#pragma once + +#include +#include +#include +#include + +#include + +namespace NCoordinator::NInfra { + +//////////////////////////////////////////////////////////////////////////////// + +userver::formats::json::Value SerializePartitionMap(const NCore::NDomain::TPartitionMap& partitionMap); + +NCore::NDomain::TPartitionMap DeserializePartitionMap(const userver::formats::json::Value& jsonValue); + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NCoordinator::NInfra diff --git a/src/main.cpp b/src/main.cpp index 8ef9820..e0a6105 100644 --- a/src/main.cpp +++ b/src/main.cpp @@ -1,3 +1,5 @@ +#include + #include #include #include @@ -8,22 +10,16 @@ #include #include - #include -#include - int main(int argc, char* argv[]) { - auto component_list = - userver::components::MinimalServerComponentList() - .Append() - .Append() - .AppendComponentList(userver::clients::http::ComponentList()) - .Append() - .Append() - .Append() - .Append() - ; + auto component_list = userver::components::MinimalServerComponentList(); + + NCoordinator::NInfra::NComponents::RegisterUserverComponents(component_list); + NCoordinator::NInfra::NComponents::RegisterYdbComponent(component_list); + + NCoordinator::NInfra::NComponents::RegisterServiceComponents(component_list); + NCoordinator::NInfra::NComponents::RegisterHandlers(component_list); return userver::utils::DaemonMain(argc, argv, component_list); } \ No newline at end of file diff --git a/tests/conftest.py b/tests/conftest.py index 4b56309..4a4480e 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -3,9 +3,5 @@ pytest_plugins = [ 'pytest_userver.plugins.core', + 'pytest_userver.plugins.ydb', ] - - - - - diff --git a/tests/requirements.txt b/tests/requirements.txt new file mode 100644 index 0000000..299a01b --- /dev/null +++ b/tests/requirements.txt @@ -0,0 +1 @@ +ydb >= 3.5.2 diff --git a/tests/test_basic.py b/tests/test_basic.py index f568a97..a8a4a42 100644 --- a/tests/test_basic.py +++ b/tests/test_basic.py @@ -2,6 +2,9 @@ async def test_basic(service_client): + await asyncio.sleep(35) + response = await service_client.post('/hello', params={'name': 'Tester'}) assert response.status == 200 - assert response.text == 'Hello, Tester!\n' \ No newline at end of file + assert 'application/json' in response.headers['Content-Type'] + assert response.text == '{"description":"Semaphore data"}' \ No newline at end of file From b6029cd5227c05ed8d3533c36cbaa14ada576b37 Mon Sep 17 00:00:00 2001 From: Ilya Repin Date: Sun, 25 Jan 2026 19:35:10 +0000 Subject: [PATCH 2/5] fix --- configs/static_config.yaml | 2 +- src/core/coordination_gateway.hpp | 2 +- src/core/partitions/partition_map.hpp | 17 ----------------- src/infra/kesus_gateway/kesus_gateway.hpp | 2 +- src/infra/serializer/serializer.hpp | 2 +- 5 files changed, 4 insertions(+), 21 deletions(-) delete mode 100644 src/core/partitions/partition_map.hpp diff --git a/configs/static_config.yaml b/configs/static_config.yaml index a0f17cf..9b03796 100644 --- a/configs/static_config.yaml +++ b/configs/static_config.yaml @@ -58,7 +58,7 @@ components_manager: operation-timeout: 1000ms retries: 3 databases: - db: + chat_db: database: chat_db max_pool_size: 10 min_pool_size: 5 diff --git a/src/core/coordination_gateway.hpp b/src/core/coordination_gateway.hpp index e71fdec..4a96ea9 100644 --- a/src/core/coordination_gateway.hpp +++ b/src/core/coordination_gateway.hpp @@ -1,6 +1,6 @@ #pragma once -#include +#include namespace NCoordinator::NCore { diff --git a/src/core/partitions/partition_map.hpp b/src/core/partitions/partition_map.hpp deleted file mode 100644 index a1b655b..0000000 --- a/src/core/partitions/partition_map.hpp +++ /dev/null @@ -1,17 +0,0 @@ -#pragma once - -#include -#include -#include - -#include -#include -// TODO to delete -namespace NCoordinator::NCore::NDomain { - -struct TPartitionMap { - TEpoch Epoch; - std::vector> Partitions; -}; // TODO Delete - -} \ No newline at end of file diff --git a/src/infra/kesus_gateway/kesus_gateway.hpp b/src/infra/kesus_gateway/kesus_gateway.hpp index d61891b..218875a 100644 --- a/src/infra/kesus_gateway/kesus_gateway.hpp +++ b/src/infra/kesus_gateway/kesus_gateway.hpp @@ -1,7 +1,7 @@ #pragma once #include -#include +#include #include diff --git a/src/infra/serializer/serializer.hpp b/src/infra/serializer/serializer.hpp index 78d3942..23ba6f2 100644 --- a/src/infra/serializer/serializer.hpp +++ b/src/infra/serializer/serializer.hpp @@ -3,7 +3,7 @@ #include #include #include -#include +#include #include From 3182f5d10cc2ee676e8105edcb468738988d4867 Mon Sep 17 00:00:00 2001 From: Ilya Repin Date: Thu, 29 Jan 2026 13:57:06 +0000 Subject: [PATCH 3/5] add hub_gateway, coordination_repository --- CMakeLists.txt | 7 +- configs/static_config.yaml | 6 +- .../coordination/coordination_gateway.hpp | 25 +++ .../coordination/coordination_repository.hpp | 15 ++ src/core/coordination_gateway.hpp | 19 -- src/core/coordination_repository.hpp | 13 -- src/core/hub/hub_gateway.hpp | 20 +++ src/hello.cpp | 12 +- src/hello.hpp | 7 +- src/infra/components/components.cpp | 16 +- src/infra/components/components.hpp | 6 +- .../coordination_dist_lock_component.cpp | 6 +- .../coordination_dist_lock_component.hpp | 4 +- .../coordination_gateway_component.cpp | 19 +- .../coordination_gateway_component.hpp | 6 +- .../coordination_repository_component.cpp | 47 +++++ .../coordination_repository_component.hpp | 33 ++++ .../coordination_service_component.cpp | 0 .../coordination_service_component.hpp | 0 .../kesus_coordination_gateway.cpp | 131 ++++++++++++++ .../kesus_coordination_gateway.hpp} | 20 ++- .../ydb_coordination_repository.cpp | 168 ++++++++++++++++++ .../ydb_coordination_repository.hpp | 28 +++ src/infra/hub_gateway/hub_gateway.cpp | 80 +++++++++ src/infra/hub_gateway/hub_gateway.hpp | 29 +++ src/infra/kesus_gateway/kesus_gateway.cpp | 90 ---------- src/infra/serializer/serializer.cpp | 32 ++++ src/infra/serializer/serializer.hpp | 3 + src/infra/serializer/serializer_ut.cpp | 1 + src/main.cpp | 4 +- tests/test_basic.py | 4 +- ydb/migrations/coordination_context.yql | 7 + 32 files changed, 685 insertions(+), 173 deletions(-) create mode 100644 src/core/coordination/coordination_gateway.hpp create mode 100644 src/core/coordination/coordination_repository.hpp delete mode 100644 src/core/coordination_gateway.hpp delete mode 100644 src/core/coordination_repository.hpp create mode 100644 src/core/hub/hub_gateway.hpp create mode 100644 src/infra/components/coordination/coordination_repository_component.cpp create mode 100644 src/infra/components/coordination/coordination_repository_component.hpp delete mode 100644 src/infra/components/coordination/coordination_service_component.cpp delete mode 100644 src/infra/components/coordination/coordination_service_component.hpp create mode 100644 src/infra/coordination_gateway/kesus_coordination_gateway.cpp rename src/infra/{kesus_gateway/kesus_gateway.hpp => coordination_gateway/kesus_coordination_gateway.hpp} (60%) create mode 100644 src/infra/coordination_repository/ydb_coordination_repository.cpp create mode 100644 src/infra/coordination_repository/ydb_coordination_repository.hpp create mode 100644 src/infra/hub_gateway/hub_gateway.cpp create mode 100644 src/infra/hub_gateway/hub_gateway.hpp delete mode 100644 src/infra/kesus_gateway/kesus_gateway.cpp create mode 100644 src/infra/serializer/serializer_ut.cpp create mode 100644 ydb/migrations/coordination_context.yql diff --git a/CMakeLists.txt b/CMakeLists.txt index 63672c0..abb5fa6 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -7,7 +7,8 @@ include(DownloadUserver) find_package( userver - COMPONENTS core # + COMPONENTS + core QUIET ) if(NOT userver_FOUND) @@ -33,7 +34,7 @@ list(REMOVE_ITEM SOURCES ${UNIT_TEST_SOURCES} ${BENCHMARK_SOURCES} ${LIBUBENCH_S add_library(${PROJECT_NAME}_objs OBJECT ${SOURCES}) - +include_directories(src) target_link_libraries( ${PROJECT_NAME}_objs PUBLIC userver::core userver::ydb @@ -74,4 +75,4 @@ install( FILES ${CONFIGS_FILES} DESTINATION ${CMAKE_INSTALL_SYSCONFDIR}/${PROJECT_NAME} COMPONENT ${PROJECT_NAME} -) \ No newline at end of file +) diff --git a/configs/static_config.yaml b/configs/static_config.yaml index 9b03796..8dcf650 100644 --- a/configs/static_config.yaml +++ b/configs/static_config.yaml @@ -63,10 +63,14 @@ components_manager: max_pool_size: 10 min_pool_size: 5 + coordination-repository: + dbname: chat_db + coordination-gateway: dbname: chat_db coordination-node: chat-coordination - semaphore-name: partition-map-lock + partition-map-semaphore: partition-map-lock + discovery-semaphore: discovery-lock initial-setup: true coordination-dist-lock: diff --git a/src/core/coordination/coordination_gateway.hpp b/src/core/coordination/coordination_gateway.hpp new file mode 100644 index 0000000..32ad2e6 --- /dev/null +++ b/src/core/coordination/coordination_gateway.hpp @@ -0,0 +1,25 @@ +#pragma once + +#include +#include + +#include +#include + +namespace NCoordinator::NCore::NDomain { + +//////////////////////////////////////////////////////////////////////////////// + +class ICoordinationGateway { +public: + virtual std::optional GetPartitionMap() const = 0; + virtual void BroadcastPartitionMap(const TPartitionMap& partitionMap) const = 0; + + virtual std::vector GetHubDiscovery() const = 0; + + virtual ~ICoordinationGateway() = default; +}; + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NCoordinator::NCore::NDomain diff --git a/src/core/coordination/coordination_repository.hpp b/src/core/coordination/coordination_repository.hpp new file mode 100644 index 0000000..3e050eb --- /dev/null +++ b/src/core/coordination/coordination_repository.hpp @@ -0,0 +1,15 @@ +#pragma once + +#include "coordination_context.hpp" + +namespace NCoordinator::NCore::NDomain { + +class ICoordinationRepository { +public: + virtual TCoordinationContext GetCoordinationContext() const = 0; + virtual void SetCoordinationContext(const TCoordinationContext& context) const = 0; + + virtual ~ICoordinationRepository() = default; +}; + +} // namespace NCoordinator::NCore::NDomain diff --git a/src/core/coordination_gateway.hpp b/src/core/coordination_gateway.hpp deleted file mode 100644 index 4a96ea9..0000000 --- a/src/core/coordination_gateway.hpp +++ /dev/null @@ -1,19 +0,0 @@ -#pragma once - -#include - -namespace NCoordinator::NCore { - -//////////////////////////////////////////////////////////////////////////////// - -class ICoordinationGateway { - public: - virtual NDomain::TPartitionMap GetPartitionMap() const = 0; - virtual void BroadcastPartitionMap(const NDomain::TPartitionMap& partitionMap) const = 0; - - virtual ~ICoordinationGateway() = default; -}; - -//////////////////////////////////////////////////////////////////////////////// - -} // namespace NCoordinator::NCore diff --git a/src/core/coordination_repository.hpp b/src/core/coordination_repository.hpp deleted file mode 100644 index e96d07d..0000000 --- a/src/core/coordination_repository.hpp +++ /dev/null @@ -1,13 +0,0 @@ -#pragma once - -namespace NCoordinator::NCore { - -// class ICoordinationrRepository { -// public: -// virtual TCoordinationContext GetCoordinationContext() const = 0; -// virtual void SetCoordinationContext(const TCoordinationContext& context) const = 0; - -// virtual ~ICoordinationrRepository() = default; -// }; - -} // namespace NCoordinator::NCore diff --git a/src/core/hub/hub_gateway.hpp b/src/core/hub/hub_gateway.hpp new file mode 100644 index 0000000..fa168a2 --- /dev/null +++ b/src/core/hub/hub_gateway.hpp @@ -0,0 +1,20 @@ +#pragma once + +#include "hub_report.hpp" + +#include + +namespace NCoordinator::NCore::NDomain { + +//////////////////////////////////////////////////////////////////////////////// + +class IHubGateway { +public: + virtual std::vector GetHubReports(const std::vector& hubs) const = 0; + + virtual ~IHubGateway() = default; +}; + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NCoordinator::NCore::NDomain diff --git a/src/hello.cpp b/src/hello.cpp index 4337c77..e920ee5 100644 --- a/src/hello.cpp +++ b/src/hello.cpp @@ -4,7 +4,7 @@ #include -#include +#include #include #include @@ -14,21 +14,19 @@ namespace coordinator { Hello::Hello(const userver::components::ComponentConfig& config, const userver::components::ComponentContext& context) : HttpHandlerJsonBase(config, context), ydb_client_(context.FindComponent().GetCoordinationClient("chat_db")) - , dist_(context.FindComponent()) { - gateway = std::make_unique( - ydb_client_, "chat-coordination", "partition-map-lock", false); + gateway = std::make_unique( + ydb_client_, "chat-coordination", "partition-map-lock", "discovery-lock", false); } userver::formats::json::Value Hello::HandleRequestJsonThrow( const userver::server::http::HttpRequest& , const userver::formats::json::Value& , - userver::server::request::RequestContext& -) const + userver::server::request::RequestContext&) const { auto result = gateway->GetPartitionMap(); - return NCoordinator::NInfra::SerializePartitionMap(result); + return NCoordinator::NInfra::SerializePartitionMap(result.value()); } } // namespace coordinator \ No newline at end of file diff --git a/src/hello.hpp b/src/hello.hpp index 396b80c..ef9ce44 100644 --- a/src/hello.hpp +++ b/src/hello.hpp @@ -1,14 +1,14 @@ #pragma once -#include +#include #include #include #include #include +#include #include #include -#include namespace coordinator { @@ -28,8 +28,7 @@ class Hello final : public userver::server::handlers::HttpHandlerJsonBase { private: NYdb::NCoordination::TSessionSettings settings; std::shared_ptr ydb_client_; - std::unique_ptr gateway; - NCoordinator::NInfra::NComponents::TCoordinationDistLockComponent& dist_; + std::unique_ptr gateway; }; } // namespace coordinator \ No newline at end of file diff --git a/src/infra/components/components.cpp b/src/infra/components/components.cpp index c1e09a8..9b7c0b9 100644 --- a/src/infra/components/components.cpp +++ b/src/infra/components/components.cpp @@ -2,6 +2,7 @@ #include #include +#include #include @@ -21,7 +22,8 @@ namespace NCoordinator::NInfra::NComponents { -void RegisterUserverComponents(userver::components::ComponentList& list) { +void RegisterUserverComponents(userver::components::ComponentList& list) +{ list.Append() .Append() .AppendComponentList(userver::clients::http::ComponentList()) @@ -31,20 +33,24 @@ void RegisterUserverComponents(userver::components::ComponentList& list) { } // Clients -void RegisterYdbComponent(userver::components::ComponentList& list) { +void RegisterYdbComponents(userver::components::ComponentList& list) +{ list.Append() .Append() .Append(); } -// Components -void RegisterServiceComponents(userver::components::ComponentList& list) { +// Infra +void RegisterInfraComponents(userver::components::ComponentList& list) +{ list.Append() + .Append() .Append(); } // Handlers -void RegisterHandlers(userver::components::ComponentList& list) { +void RegisterHandlers(userver::components::ComponentList& list) +{ list.Append(); } diff --git a/src/infra/components/components.hpp b/src/infra/components/components.hpp index 0273674..f9912c9 100644 --- a/src/infra/components/components.hpp +++ b/src/infra/components/components.hpp @@ -9,10 +9,10 @@ namespace NCoordinator::NInfra::NComponents { void RegisterUserverComponents(userver::components::ComponentList&); // Clients -void RegisterYdbComponent(userver::components::ComponentList&); +void RegisterYdbComponents(userver::components::ComponentList&); -// Components -void RegisterServiceComponents(userver::components::ComponentList&); +// Infra +void RegisterInfraComponents(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 index e381632..77536a5 100644 --- a/src/infra/components/coordination/coordination_dist_lock_component.cpp +++ b/src/infra/components/coordination/coordination_dist_lock_component.cpp @@ -37,11 +37,9 @@ TCoordinationDistLockComponent::~TCoordinationDistLockComponent() } void TCoordinationDistLockComponent::DoWork() { - LOG_CRITICAL() << "DO WORK 1"; - while (!userver::engine::current_task::ShouldCancel()) { - LOG_CRITICAL() << "DO WORK 2"; - auto partitionMap = Gateway_.GetPartitionMap(); + auto partitionMap = Gateway_.GetPartitionMap().value_or(NCore::NDomain::TPartitionMap{}); + ++(partitionMap.Epoch.GetUnderlying()); Gateway_.BroadcastPartitionMap(partitionMap); diff --git a/src/infra/components/coordination/coordination_dist_lock_component.hpp b/src/infra/components/coordination/coordination_dist_lock_component.hpp index 26bf787..5c9738b 100644 --- a/src/infra/components/coordination/coordination_dist_lock_component.hpp +++ b/src/infra/components/coordination/coordination_dist_lock_component.hpp @@ -1,6 +1,6 @@ #pragma once -#include +#include #include @@ -23,7 +23,7 @@ class TCoordinationDistLockComponent void DoWork() final; private: - NCore::ICoordinationGateway& Gateway_; // Replace with service with core + NCore::NDomain::ICoordinationGateway& Gateway_; // Replace with service with core }; //////////////////////////////////////////////////////////////////////////////// diff --git a/src/infra/components/coordination/coordination_gateway_component.cpp b/src/infra/components/coordination/coordination_gateway_component.cpp index 65c0d86..edaedc3 100644 --- a/src/infra/components/coordination/coordination_gateway_component.cpp +++ b/src/infra/components/coordination/coordination_gateway_component.cpp @@ -1,6 +1,6 @@ #include "coordination_gateway_component.hpp" -#include +#include #include #include @@ -18,19 +18,21 @@ TCoordinationGatewayComponent::TCoordinationGatewayComponent( { auto dbname = config["dbname"].As(); auto coordinationNode = config["coordination-node"].As(); - auto semaphoreName = config["semaphore-name"].As(); + auto partitionMapSemaphore = config["partition-map-semaphore"].As(); + auto discoverySemaphore = config["discovery-semaphore"].As(); auto initialSetup = config["initial-setup"].As(true); auto coordinationClient = context.FindComponent().GetCoordinationClient(dbname); - Gateway_ = std::make_unique( + Gateway_ = std::make_unique( std::move(coordinationClient), coordinationNode, - semaphoreName, + partitionMapSemaphore, + discoverySemaphore, initialSetup); } -NCore::ICoordinationGateway& TCoordinationGatewayComponent::GetGateway() +NCore::NDomain::ICoordinationGateway& TCoordinationGatewayComponent::GetGateway() { return *Gateway_; } @@ -49,9 +51,12 @@ additionalProperties: false coordination-node: type: string description: name of the coordination node within the database - semaphore-name: + partition-map-semaphore: type: string - description: name of the semaphore within the coordination node + description: name of the partition map semaphore within the coordination node + discovery-semaphore: + type: string + description: name of the discovery semaphore within the coordination node initial-setup: type: boolean description: if true, then create the coordination node and the semaphore diff --git a/src/infra/components/coordination/coordination_gateway_component.hpp b/src/infra/components/coordination/coordination_gateway_component.hpp index b21cfc6..b7ba6c6 100644 --- a/src/infra/components/coordination/coordination_gateway_component.hpp +++ b/src/infra/components/coordination/coordination_gateway_component.hpp @@ -1,6 +1,6 @@ #pragma once -#include +#include #include @@ -20,12 +20,12 @@ class TCoordinationGatewayComponent const userver::components::ComponentConfig& config, const userver::components::ComponentContext& context); - NCore::ICoordinationGateway& GetGateway(); + NCore::NDomain::ICoordinationGateway& GetGateway(); static userver::yaml_config::Schema GetStaticConfigSchema(); private: - std::unique_ptr Gateway_; + std::unique_ptr Gateway_; }; //////////////////////////////////////////////////////////////////////////////// diff --git a/src/infra/components/coordination/coordination_repository_component.cpp b/src/infra/components/coordination/coordination_repository_component.cpp new file mode 100644 index 0000000..2d31f9a --- /dev/null +++ b/src/infra/components/coordination/coordination_repository_component.cpp @@ -0,0 +1,47 @@ +#include "coordination_repository_component.hpp" + +#include + +#include +#include +#include +#include + +namespace NCoordinator::NInfra::NComponents { + +//////////////////////////////////////////////////////////////////////////////// + +TCoordinationRepositoryComponent::TCoordinationRepositoryComponent( + const userver::components::ComponentConfig& config, + const userver::components::ComponentContext& context) + : LoggableComponentBase(config, context) +{ + auto dbname = config["dbname"].As(); + + auto tableClient = context.FindComponent().GetTableClient(dbname); + + Repository_ = std::make_unique(std::move(tableClient)); +} + +NCore::NDomain::ICoordinationRepository& TCoordinationRepositoryComponent::GetRepository() +{ + return *Repository_; +} + +userver::yaml_config::Schema TCoordinationRepositoryComponent::GetStaticConfigSchema() { + return userver::yaml_config::MergeSchemas( + R"( +type: object +description: Component for coordination repository +additionalProperties: false +properties: + dbname: + type: string + description: the key of the database within ydb component (NOT the + actual database path) +)"); +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NCoordinator::NInfra::NComponents diff --git a/src/infra/components/coordination/coordination_repository_component.hpp b/src/infra/components/coordination/coordination_repository_component.hpp new file mode 100644 index 0000000..4aacd68 --- /dev/null +++ b/src/infra/components/coordination/coordination_repository_component.hpp @@ -0,0 +1,33 @@ +#pragma once + +#include + +#include + +#include + +namespace NCoordinator::NInfra::NComponents { + +//////////////////////////////////////////////////////////////////////////////// + +class TCoordinationRepositoryComponent + : public userver::components::LoggableComponentBase +{ +public: + static constexpr std::string_view kName = "coordination-repository"; + + TCoordinationRepositoryComponent( + const userver::components::ComponentConfig& config, + const userver::components::ComponentContext& context); + + NCore::NDomain::ICoordinationRepository& GetRepository(); + + static userver::yaml_config::Schema GetStaticConfigSchema(); + +private: + std::unique_ptr Repository_; +}; + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NCoordinator::NInfra::NComponents diff --git a/src/infra/components/coordination/coordination_service_component.cpp b/src/infra/components/coordination/coordination_service_component.cpp deleted file mode 100644 index e69de29..0000000 diff --git a/src/infra/components/coordination/coordination_service_component.hpp b/src/infra/components/coordination/coordination_service_component.hpp deleted file mode 100644 index e69de29..0000000 diff --git a/src/infra/coordination_gateway/kesus_coordination_gateway.cpp b/src/infra/coordination_gateway/kesus_coordination_gateway.cpp new file mode 100644 index 0000000..54fc7ef --- /dev/null +++ b/src/infra/coordination_gateway/kesus_coordination_gateway.cpp @@ -0,0 +1,131 @@ +#include "kesus_coordination_gateway.hpp" + +#include +#include +#include +#include + +#include +#include +#include + +#include +#include +#include +#include + +namespace { + +//////////////////////////////////////////////////////////////////////////////// + +constexpr inline std::string_view DEFAULT_PARTITION_MAP_DATA = "{\"partitions\": [],\"epoch\": 0}"; +constexpr inline std::uint64_t PARTITION_MAP_SEMAPHORE_LIMIT = 1; +constexpr inline std::uint64_t DISCOVERY_SEMAPHORE_LIMIT = std::numeric_limits::max(); + +//////////////////////////////////////////////////////////////////////////////// + +} // anonymous namespace + +namespace NCoordinator::NInfra::NGateway { + +//////////////////////////////////////////////////////////////////////////////// + +TKesusCoordinationGateway::TKesusCoordinationGateway( + std::shared_ptr coordinationClient, + const std::string& coordinationNode, + const std::string& partitionMapSemaphore, + const std::string& discoverySemaphore, + const bool initialSetup) + : PartitionMapSemaphore_(partitionMapSemaphore) + , DiscoverySemaphore_(discoverySemaphore) + , CoordinationSession_(nullptr) +{ + if (initialSetup) { + InitialSetup(*coordinationClient, coordinationNode); + } + + auto session = coordinationClient->StartSession(coordinationNode, {}); + CoordinationSession_ = std::make_unique(std::move(session)); +} + +std::optional TKesusCoordinationGateway::GetPartitionMap() const +{ + NYdb::NCoordination::TDescribeSemaphoreSettings describeSettings; + auto description = CoordinationSession_->DescribeSemaphore(PartitionMapSemaphore_, describeSettings); + + userver::formats::json::Value json; + try { + json = userver::formats::json::FromString(description.GetData()); + } catch(std::exception& ex) { + LOG_ERROR() << "Can't parse json: " << ex; + return std::nullopt; + } + + try { + return DeserializePartitionMap(json); + } catch (std::exception& ex) { + LOG_ERROR() << "Can't deserialize partition map: " << ex; + return std::nullopt; + } +} + +void TKesusCoordinationGateway::BroadcastPartitionMap(const NCore::NDomain::TPartitionMap& partitionMap) const +{ + auto json = SerializePartitionMap(partitionMap); + CoordinationSession_->UpdateSemaphore( + PartitionMapSemaphore_, + userver::formats::json::ToStableString(json)); +} + +std::vector TKesusCoordinationGateway::GetHubDiscovery() const +{ + NYdb::NCoordination::TDescribeSemaphoreSettings describeSettings; + describeSettings.IncludeOwners(true); + + auto description = CoordinationSession_->DescribeSemaphore(DiscoverySemaphore_, describeSettings); + + std::vector hubs; + hubs.reserve(description.GetCount()); + + for (const auto& owner : description.GetOwners()) { + auto hub = NCore::NDomain::THubEndpoint{owner.GetData()}; + hubs.emplace_back(std::move(hub)); + } + + return hubs; +} + +void TKesusCoordinationGateway::InitialSetup( + userver::ydb::CoordinationClient& coordinationClient, + const std::string& coordinationNode) const +{ + NYdb::NCoordination::TCreateNodeSettings createNodeSettings; + try { + coordinationClient.CreateNode(coordinationNode, createNodeSettings); + } catch (const userver::ydb::YdbResponseError& ex) { + LOG_WARNING() << "Could not create coordination node: " << ex; + } + + auto session = coordinationClient.StartSession(coordinationNode, {}); + + try { + session.CreateSemaphore( + PartitionMapSemaphore_, + PARTITION_MAP_SEMAPHORE_LIMIT); + // std::string{DEFAULT_PARTITION_MAP_DATA}); + } catch (const userver::ydb::YdbResponseError& ex) { + LOG_WARNING() << "Could not create partition map semaphore: " << ex; + } + + try { + session.CreateSemaphore(DiscoverySemaphore_, DISCOVERY_SEMAPHORE_LIMIT); + } catch (const userver::ydb::YdbResponseError& ex) { + LOG_WARNING() << "Could not create discovery semaphore: " << ex; + } + + session.Close(); +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NCoordinator::NInfra::NGateway diff --git a/src/infra/kesus_gateway/kesus_gateway.hpp b/src/infra/coordination_gateway/kesus_coordination_gateway.hpp similarity index 60% rename from src/infra/kesus_gateway/kesus_gateway.hpp rename to src/infra/coordination_gateway/kesus_coordination_gateway.hpp index 218875a..f4c3db4 100644 --- a/src/infra/kesus_gateway/kesus_gateway.hpp +++ b/src/infra/coordination_gateway/kesus_coordination_gateway.hpp @@ -1,36 +1,42 @@ #pragma once -#include +#include +#include #include #include #include +#include namespace NCoordinator::NInfra::NGateway { //////////////////////////////////////////////////////////////////////////////// -class TKesusGateway - : public NCore::ICoordinationGateway +class TKesusCoordinationGateway + : public NCore::NDomain::ICoordinationGateway { public: - TKesusGateway( + TKesusCoordinationGateway( std::shared_ptr coordinationClient, const std::string& coordinationNode, - const std::string& semaphoreName, + const std::string& partitionMapSemaphore, + const std::string& discoverySemaphore, const bool initialSetup); - NCore::NDomain::TPartitionMap GetPartitionMap() const override; + std::optional GetPartitionMap() const override; void BroadcastPartitionMap(const NCore::NDomain::TPartitionMap& partitionMap) const override; + std::vector GetHubDiscovery() const override; + private: void InitialSetup( userver::ydb::CoordinationClient& coordinationClient, const std::string& coordinationNode) const; private: - const std::string SemaphoreName_; + const std::string PartitionMapSemaphore_; + const std::string DiscoverySemaphore_; std::unique_ptr CoordinationSession_; }; diff --git a/src/infra/coordination_repository/ydb_coordination_repository.cpp b/src/infra/coordination_repository/ydb_coordination_repository.cpp new file mode 100644 index 0000000..725b757 --- /dev/null +++ b/src/infra/coordination_repository/ydb_coordination_repository.cpp @@ -0,0 +1,168 @@ +#include "ydb_coordination_repository.hpp" + +#include +#include + +#include + +#include +#include +#include +#include +#include + +namespace { + +//////////////////////////////////////////////////////////////////////////////// + +const userver::ydb::Query LOAD_COORDINATION_CONTEXT_QUERY = { +R"-( +SELECT + partition_id, + cooldown_epoch, + weight +FROM + coordination_context; +)-", + userver::ydb::Query::NameLiteral("load_coordination_context"), + userver::ydb::Query::LogMode::kFull, +}; + +const userver::ydb::Query SET_COORDINATION_CONTEXT_QUERY = { +R"-( +DECLARE $rows AS List>; +DECLARE $active_ids AS List; + +DELETE FROM coordination_context +WHERE partition_id NOT IN $active_ids; + +REPLACE INTO coordination_context +SELECT * FROM AS_TABLE($rows); +)-", + userver::ydb::Query::NameLiteral("update_coordination_context"), + userver::ydb::Query::LogMode::kFull, +}; + +struct TSetCoordinationContextRequest { + static constexpr userver::ydb::StructMemberNames<3> kYdbMemberNames{{ + {"PartitionId", "partition_id"}, + {"CooldownEpoch", "cooldown_epoch"}, + {"Weight", "weight"}, + }}; + + std::uint64_t PartitionId; + std::optional CooldownEpoch; + std::optional Weight; +}; + +//////////////////////////////////////////////////////////////////////////////// + +} // anonymous namespace + +namespace NCoordinator::NInfra::NRepository { + +//////////////////////////////////////////////////////////////////////////////// + +TYdbCoordinationRepository::TYdbCoordinationRepository(std::shared_ptr ydbClient) + : YdbClient_(std::move(ydbClient)) +{ } + +NCore::NDomain::TCoordinationContext TYdbCoordinationRepository::GetCoordinationContext() const +{ + const userver::ydb::OperationSettings queryParams = { + 3, // retries - TODO take from dynconfig + std::chrono::milliseconds(1000), // operation_timeout - TODO take from dynconfig + std::chrono::milliseconds(1000), // cancel_after - TODO take from dynconfig + std::chrono::milliseconds(1100), // client_timeout - TODO take from dynconfig + userver::ydb::TransactionMode::kOnlineRO + }; + + auto response = YdbClient_->ExecuteDataQuery( + queryParams, + LOAD_COORDINATION_CONTEXT_QUERY); + + if (response.GetCursorCount() != 1) { + throw std::runtime_error("Unexpected response data"); + } + + NCore::NDomain::TCoordinationContext context; + for (auto row : response.GetSingleCursor()) { + auto partitionId = NCore::NDomain::TPartitionId{ + row.Get("partition_id") + }; + auto cooldownEpoch = row.Get>("cooldown_epoch"); + auto weight = row.Get>("weight"); + + if (cooldownEpoch.has_value()) { + context.PartitionCooldowns[partitionId] = NCore::NDomain::TEpoch{cooldownEpoch.value()}; + } + if (weight.has_value()) { + context.PartitionWeights[partitionId] = NCore::NDomain::TPartitionWeight{weight.value()}; + } + } + + return context; +} + +void TYdbCoordinationRepository::SetCoordinationContext(const NCore::NDomain::TCoordinationContext& context) const +{ + std::vector rows; + std::vector activeIds; + + std::unordered_set partitionIds; + for (const auto& [partitionId, _] : context.PartitionCooldowns) { + partitionIds.insert(partitionId); + } + for (const auto& [partitionId, _] : context.PartitionWeights) { + partitionIds.insert(partitionId); + } + + rows.reserve(partitionIds.size()); + activeIds.reserve(partitionIds.size()); + + for (auto partitionId : partitionIds) { + TSetCoordinationContextRequest row; + row.PartitionId = partitionId.GetUnderlying(); + + if (auto it = context.PartitionCooldowns.find(partitionId); + it != context.PartitionCooldowns.end()) + { + row.CooldownEpoch = it->second.GetUnderlying(); + } + + if (auto it = context.PartitionWeights.find(partitionId); + it != context.PartitionWeights.end()) + { + row.Weight = it->second.GetUnderlying(); + } + + rows.emplace_back(std::move(row)); + activeIds.emplace_back(partitionId); + } + + const userver::ydb::OperationSettings queryParams = { + 3, // retries - TODO take from dynconfig + std::chrono::milliseconds(1000), // operation_timeout - TODO take from dynconfig + std::chrono::milliseconds(1000), // cancel_after - TODO take from dynconfig + std::chrono::milliseconds(1100), // client_timeout - TODO take from dynconfig + userver::ydb::TransactionMode::kSerializableRW + }; + + auto response = YdbClient_->ExecuteDataQuery( + queryParams, + SET_COORDINATION_CONTEXT_QUERY, + "$rows", + rows, + "$active_ids", + activeIds); + + // TODO handling errors + + if (response.GetCursorCount()) { + throw std::runtime_error("Unexpected response data"); + } +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NCoordinator::NInfra::NRepository diff --git a/src/infra/coordination_repository/ydb_coordination_repository.hpp b/src/infra/coordination_repository/ydb_coordination_repository.hpp new file mode 100644 index 0000000..998c274 --- /dev/null +++ b/src/infra/coordination_repository/ydb_coordination_repository.hpp @@ -0,0 +1,28 @@ +#pragma once + +#include +#include + +#include + +namespace NCoordinator::NInfra::NRepository { + +//////////////////////////////////////////////////////////////////////////////// + +class TYdbCoordinationRepository + : public NCore::NDomain::ICoordinationRepository +{ +public: + TYdbCoordinationRepository(std::shared_ptr ydbClient); + + NCore::NDomain::TCoordinationContext GetCoordinationContext() const override; + void SetCoordinationContext(const NCore::NDomain::TCoordinationContext& context) const override; + +private: + std::shared_ptr YdbClient_; + // TODO addd dynconfig +}; + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NCoordinator::NInfra::NRepository diff --git a/src/infra/hub_gateway/hub_gateway.cpp b/src/infra/hub_gateway/hub_gateway.cpp new file mode 100644 index 0000000..c3abae5 --- /dev/null +++ b/src/infra/hub_gateway/hub_gateway.cpp @@ -0,0 +1,80 @@ +#include "hub_gateway.hpp" + +#include + +#include +#include +#include + +#include + +namespace { + +//////////////////////////////////////////////////////////////////////////////// + +inline constexpr auto DEFAULT_TIMEOUT = std::chrono::seconds(5); + +inline constexpr auto DEFAULT_RETRIES = 3; + +//////////////////////////////////////////////////////////////////////////////// + +} + +namespace NCoordinator::NInfra::NGateway { + +//////////////////////////////////////////////////////////////////////////////// + +THubGateway::THubGateway(userver::clients::http::Client& client) + : Client_(client) +{ } + +std::vector THubGateway::GetHubReports( + const std::vector& hubs) const +{ + std::vector requests; + requests.reserve(hubs.size()); + + for (const auto& hub : hubs) { + auto request = Client_.CreateRequest() + .follow_redirects(false) + .get(hub.GetUnderlying()) + .timeout(DEFAULT_TIMEOUT) + .retry(DEFAULT_RETRIES) + .async_perform(); + + requests.emplace_back(std::move(request)); + } + + std::vector reports; + reports.reserve(requests.size()); + + auto deadline = userver::engine::Deadline::FromDuration(DEFAULT_TIMEOUT); + + while (auto index = userver::engine::WaitAnyUntil(deadline, requests)) { + const std::shared_ptr response = requests[*index].Get(); + + if (response->IsOk()) { + userver::formats::json::Value json; + try { + json = userver::formats::json::FromString(response->body()); + } catch(std::exception& ex) { + LOG_ERROR() << "Can't parse json: " << ex; + continue; + } + + try { + auto report = DeserializeHubReport(json); + reports.emplace_back(std::move(report)); + } catch(std::exception& ex) { + LOG_ERROR() << "Can't deserialize hub report " << ex; + continue; + } + } + } + + return reports; +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NCoordinator::NInfra::NGateway diff --git a/src/infra/hub_gateway/hub_gateway.hpp b/src/infra/hub_gateway/hub_gateway.hpp new file mode 100644 index 0000000..055fd92 --- /dev/null +++ b/src/infra/hub_gateway/hub_gateway.hpp @@ -0,0 +1,29 @@ +#pragma once + +#include +#include + +#include + +#include + +namespace NCoordinator::NInfra::NGateway { + +//////////////////////////////////////////////////////////////////////////////// + +class THubGateway + : public NCore::NDomain::IHubGateway +{ +public: + THubGateway(userver::clients::http::Client& client); + + std::vector GetHubReports( + const std::vector& hubs) const override; + +private: + userver::clients::http::Client& Client_; +}; + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NCoordinator::NInfra::NGateway diff --git a/src/infra/kesus_gateway/kesus_gateway.cpp b/src/infra/kesus_gateway/kesus_gateway.cpp deleted file mode 100644 index 2db4dd9..0000000 --- a/src/infra/kesus_gateway/kesus_gateway.cpp +++ /dev/null @@ -1,90 +0,0 @@ -#include "kesus_gateway.hpp" - -#include -#include -#include -#include - -#include -#include -#include - -#include - -namespace { - -//////////////////////////////////////////////////////////////////////////////// - -const std::uint64_t SEMAPHORE_LIMIT = 1; - -//////////////////////////////////////////////////////////////////////////////// - -} // anonymous namespace - -namespace NCoordinator::NInfra::NGateway { - -//////////////////////////////////////////////////////////////////////////////// - -TKesusGateway::TKesusGateway( - std::shared_ptr coordinationClient, - const std::string& coordinationNode, - const std::string& semaphoreName, - const bool initialSetup) - : SemaphoreName_(semaphoreName) - , CoordinationSession_(nullptr) -{ - if (initialSetup) { - InitialSetup(*coordinationClient, coordinationNode); - } - - auto session = coordinationClient->StartSession(coordinationNode, {}); - CoordinationSession_ = std::make_unique(std::move(session)); -} - -NCore::NDomain::TPartitionMap TKesusGateway::GetPartitionMap() const -{ - NYdb::NCoordination::TDescribeSemaphoreSettings describeSettings; - auto description = CoordinationSession_->DescribeSemaphore(SemaphoreName_, describeSettings); - - userver::formats::json::Value json; - try { - json = userver::formats::json::FromString(description.GetData()); - } catch(std::exception& ex) { - LOG_ERROR() << "Can't parse json: " << ex; - return {}; - } - - return DeserializePartitionMap(json); -} - -void TKesusGateway::BroadcastPartitionMap(const NCore::NDomain::TPartitionMap& partitionMap) const -{ - auto json = SerializePartitionMap(partitionMap); - CoordinationSession_->UpdateSemaphore( - SemaphoreName_, - userver::formats::json::ToStableString(json)); -} - -void TKesusGateway::InitialSetup( - userver::ydb::CoordinationClient& coordinationClient, - const std::string& coordinationNode) const -{ - NYdb::NCoordination::TCreateNodeSettings createNodeSettings; - try { - coordinationClient.CreateNode(coordinationNode, createNodeSettings); - } catch (const userver::ydb::YdbResponseError& ex) { - LOG_WARNING() << "Could not create coordination node: " << ex; - } - - auto session = coordinationClient.StartSession(coordinationNode, {}); - try { - session.CreateSemaphore(SemaphoreName_, SEMAPHORE_LIMIT); - } catch (const userver::ydb::YdbResponseError& ex) { - LOG_WARNING() << "Could not create semaphore: " << ex; - } - session.Close(); -} - -//////////////////////////////////////////////////////////////////////////////// - -} // namespace NCoordinator::NInfra::NGateway diff --git a/src/infra/serializer/serializer.cpp b/src/infra/serializer/serializer.cpp index 191caf9..eb4920b 100644 --- a/src/infra/serializer/serializer.cpp +++ b/src/infra/serializer/serializer.cpp @@ -62,6 +62,38 @@ NCore::NDomain::TPartitionMap DeserializePartitionMap(const userver::formats::js return result; } +NCore::NDomain::THubReport DeserializeHubReport(const userver::formats::json::Value& jsonValue) +{ + NCore::NDomain::THubReport report; + + report.Epoch = NCore::NDomain::TEpoch{ + std::stoull(jsonValue["epoch"].As()) + }; + report.Endpoint = NCore::NDomain::THubEndpoint{ + jsonValue["endpoint"].As() + }; + report.DC = NCore::NDomain::THubDC{ + jsonValue["dc"].As() + }; + report.LoadFactor = NCore::NDomain::TLoadFactor{ + jsonValue["load_factor"].As() + }; + + const auto& partitionsJson = jsonValue["partitions"]; + for (auto it = partitionsJson.begin(); it != partitionsJson.end(); ++it) { + NCore::NDomain::TPartitionId partitionId{ + std::stoull(it.GetName()) + }; + NCore::NDomain::TPartitionWeight partitionWeight{ + std::stoull(it->As()) + }; + + report.PartitionWeights.emplace(partitionId, partitionWeight); + } + + return report; +} + //////////////////////////////////////////////////////////////////////////////// } // namespace NCoordinator::NInfra diff --git a/src/infra/serializer/serializer.hpp b/src/infra/serializer/serializer.hpp index 23ba6f2..bbfd94f 100644 --- a/src/infra/serializer/serializer.hpp +++ b/src/infra/serializer/serializer.hpp @@ -4,6 +4,7 @@ #include #include #include +#include #include @@ -15,6 +16,8 @@ userver::formats::json::Value SerializePartitionMap(const NCore::NDomain::TParti NCore::NDomain::TPartitionMap DeserializePartitionMap(const userver::formats::json::Value& jsonValue); +NCore::NDomain::THubReport DeserializeHubReport(const userver::formats::json::Value& jsonValue); + //////////////////////////////////////////////////////////////////////////////// } // namespace NCoordinator::NInfra diff --git a/src/infra/serializer/serializer_ut.cpp b/src/infra/serializer/serializer_ut.cpp new file mode 100644 index 0000000..0ffdd02 --- /dev/null +++ b/src/infra/serializer/serializer_ut.cpp @@ -0,0 +1 @@ +// TODO \ No newline at end of file diff --git a/src/main.cpp b/src/main.cpp index e0a6105..c18ca38 100644 --- a/src/main.cpp +++ b/src/main.cpp @@ -16,9 +16,9 @@ int main(int argc, char* argv[]) { auto component_list = userver::components::MinimalServerComponentList(); NCoordinator::NInfra::NComponents::RegisterUserverComponents(component_list); - NCoordinator::NInfra::NComponents::RegisterYdbComponent(component_list); + NCoordinator::NInfra::NComponents::RegisterYdbComponents(component_list); - NCoordinator::NInfra::NComponents::RegisterServiceComponents(component_list); + NCoordinator::NInfra::NComponents::RegisterInfraComponents(component_list); NCoordinator::NInfra::NComponents::RegisterHandlers(component_list); return userver::utils::DaemonMain(argc, argv, component_list); diff --git a/tests/test_basic.py b/tests/test_basic.py index a8a4a42..1f4a9fa 100644 --- a/tests/test_basic.py +++ b/tests/test_basic.py @@ -2,9 +2,7 @@ async def test_basic(service_client): - await asyncio.sleep(35) - response = await service_client.post('/hello', params={'name': 'Tester'}) assert response.status == 200 assert 'application/json' in response.headers['Content-Type'] - assert response.text == '{"description":"Semaphore data"}' \ No newline at end of file + assert response.text == '{"partitions":[],"epoch":0}' \ No newline at end of file diff --git a/ydb/migrations/coordination_context.yql b/ydb/migrations/coordination_context.yql new file mode 100644 index 0000000..0be5e86 --- /dev/null +++ b/ydb/migrations/coordination_context.yql @@ -0,0 +1,7 @@ +-- +goose Up +CREATE TABLE coordination_context ( + partition_id Uint64, + cooldown_epoch Uint64, + weight Uint64, + PRIMARY KEY (partition_id) +); \ No newline at end of file From d25b53cb51951ed5230568ce135c8017b759f9a7 Mon Sep 17 00:00:00 2001 From: Ilya Repin Date: Thu, 29 Jan 2026 14:06:17 +0000 Subject: [PATCH 4/5] fix --- .../components/coordination/coordination_gateway_component.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/infra/components/coordination/coordination_gateway_component.cpp b/src/infra/components/coordination/coordination_gateway_component.cpp index edaedc3..7a91cbd 100644 --- a/src/infra/components/coordination/coordination_gateway_component.cpp +++ b/src/infra/components/coordination/coordination_gateway_component.cpp @@ -3,8 +3,8 @@ #include #include -#include #include +#include #include namespace NCoordinator::NInfra::NComponents { From 52f1437ce460e9ef53d4110842ea47d727b11e3b Mon Sep 17 00:00:00 2001 From: Ilya Repin Date: Thu, 29 Jan 2026 14:07:09 +0000 Subject: [PATCH 5/5] remove CI --- .github/workflows/ci.yml | 122 +++++++++++++++++----------------- .github/workflows/docker.yaml | 66 +++++++++--------- 2 files changed, 94 insertions(+), 94 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index bce2a92..3b9bd78 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -1,61 +1,61 @@ -# NOTE: if this CI check breaks, just remove it, docker checks are enough anyway. - -name: CI - -'on': - schedule: - - cron: '30 5 * * 1' # Every Monday at 5:30 - pull_request: - push: - branches: - - develop - -env: - UBSAN_OPTIONS: print_stacktrace=1 - -jobs: - posix: - strategy: - fail-fast: false - matrix: - include: - - os: ubuntu-24.04 - make: test-debug - info: g++-11 + test-debug - - - os: ubuntu-24.04 - make: test-release - info: g++-11 + test-release - - name: '${{matrix.os}}: ${{matrix.info}}' - runs-on: ${{matrix.os}} - - steps: - - uses: actions/checkout@v4 - with: - submodules: true - - - name: Reuse ccache directory - uses: actions/cache@v4 - with: - path: ~/.cache/ccache - key: '${{matrix.os}} ${{matrix.info}} ccache-dir ${{github.ref}} run-${{github.run_number}}' - restore-keys: | - ${{matrix.os}} ${{matrix.info}} ccache-dir ${{github.ref}} run-' - ${{matrix.os}} ${{matrix.info}} ccache- - - - name: Install packages - run: | - DEPS_FILE="https://raw.githubusercontent.com/userver-framework/userver/refs/heads/develop/scripts/docs/en/deps/${{matrix.os}}.md" - sudo apt update - sudo apt install --allow-downgrades -y $(wget -q -O - ${DEPS_FILE}) - python3 -m pip install -r requirements.txt - - - name: Setup ccache - run: | - ccache -M 2.0GB - ccache -s - - - name: Run ${{matrix.make}} - run: | - make ${{matrix.make}} \ No newline at end of file +# # NOTE: if this CI check breaks, just remove it, docker checks are enough anyway. + +# name: CI + +# 'on': +# schedule: +# - cron: '30 5 * * 1' # Every Monday at 5:30 +# pull_request: +# push: +# branches: +# - develop + +# env: +# UBSAN_OPTIONS: print_stacktrace=1 + +# jobs: +# posix: +# strategy: +# fail-fast: false +# matrix: +# include: +# - os: ubuntu-24.04 +# make: test-debug +# info: g++-11 + test-debug + +# - os: ubuntu-24.04 +# make: test-release +# info: g++-11 + test-release + +# name: '${{matrix.os}}: ${{matrix.info}}' +# runs-on: ${{matrix.os}} + +# steps: +# - uses: actions/checkout@v4 +# with: +# submodules: true + +# - name: Reuse ccache directory +# uses: actions/cache@v4 +# with: +# path: ~/.cache/ccache +# key: '${{matrix.os}} ${{matrix.info}} ccache-dir ${{github.ref}} run-${{github.run_number}}' +# restore-keys: | +# ${{matrix.os}} ${{matrix.info}} ccache-dir ${{github.ref}} run-' +# ${{matrix.os}} ${{matrix.info}} ccache- + +# - name: Install packages +# run: | +# DEPS_FILE="https://raw.githubusercontent.com/userver-framework/userver/refs/heads/develop/scripts/docs/en/deps/${{matrix.os}}.md" +# sudo apt update +# sudo apt install --allow-downgrades -y $(wget -q -O - ${DEPS_FILE}) +# python3 -m pip install -r requirements.txt + +# - name: Setup ccache +# run: | +# ccache -M 2.0GB +# ccache -s + +# - name: Run ${{matrix.make}} +# run: | +# make ${{matrix.make}} \ No newline at end of file diff --git a/.github/workflows/docker.yaml b/.github/workflows/docker.yaml index 8145082..133267b 100644 --- a/.github/workflows/docker.yaml +++ b/.github/workflows/docker.yaml @@ -1,40 +1,40 @@ -name: Docker build +# name: Docker build -'on': - schedule: - - cron: '30 5 * * 1' # Every Monday at 5:30 - pull_request: - push: - branches: - - master - - develop - - feature/** +# 'on': +# schedule: +# - cron: '30 5 * * 1' # Every Monday at 5:30 +# pull_request: +# push: +# branches: +# - master +# - develop +# - feature/** -env: - UBSAN_OPTIONS: print_stacktrace=1 +# env: +# UBSAN_OPTIONS: print_stacktrace=1 -jobs: - tests: - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v4 - with: - submodules: true +# jobs: +# tests: +# runs-on: ubuntu-latest +# steps: +# - uses: actions/checkout@v4 +# with: +# submodules: true - - name: Reuse ccache directory - uses: actions/cache@v4 - with: - path: .ccache - key: 'ccache-dir-${{github.ref}}_run-${{github.run_number}}' - restore-keys: | - ccache-dir-${{github.ref}}_run- - ccache- +# - name: Reuse ccache directory +# uses: actions/cache@v4 +# with: +# path: .ccache +# key: 'ccache-dir-${{github.ref}}_run-${{github.run_number}}' +# restore-keys: | +# ccache-dir-${{github.ref}}_run- +# ccache- - - name: Cmake - run: make docker-cmake-debug +# - name: Cmake +# run: make docker-cmake-debug - - name: Build - run: make docker-build-debug +# - name: Build +# run: make docker-build-debug - - name: Test - run: make docker-test-debug \ No newline at end of file +# - name: Test +# run: make docker-test-debug \ No newline at end of file