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 diff --git a/CMakeLists.txt b/CMakeLists.txt index c1535cc..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) @@ -18,6 +19,14 @@ 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) file(GLOB_RECURSE SOURCES ${CMAKE_CURRENT_SOURCE_DIR}/src/*.[hc]pp) file(GLOB_RECURSE UNIT_TEST_SOURCES ${CMAKE_CURRENT_SOURCE_DIR}/src/*_ut.cpp) @@ -25,10 +34,10 @@ 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 # + PUBLIC userver::core userver::ydb ) @@ -42,7 +51,9 @@ target_link_libraries(${PROJECT_NAME}_unittest PRIVATE ${PROJECT_NAME}_objs user add_google_tests(${PROJECT_NAME}_unittest) # Functional testing -userver_testsuite_add_simple() +userver_testsuite_add_simple( + REQUIREMENTS ${CMAKE_CURRENT_SOURCE_DIR}/tests/requirements.txt +) # Install include(GNUInstallDirs) @@ -64,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/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..8dcf650 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,34 @@ components_manager: method: POST task_processor: main-task-processor + ydb: + operation-settings: + cancel-after: 1000ms + client-timeout: 1100ms + operation-timeout: 1000ms + retries: 3 + databases: + chat_db: + database: chat_db + max_pool_size: 10 + min_pool_size: 5 + + coordination-repository: + dbname: chat_db + + coordination-gateway: + dbname: chat_db + coordination-node: chat-coordination + partition-map-semaphore: partition-map-lock + discovery-semaphore: discovery-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/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/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 c6cf79d..e920ee5 100644 --- a/src/hello.cpp +++ b/src/hello.cpp @@ -2,12 +2,31 @@ #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")) +{ + 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 +{ + auto result = gateway->GetPartitionMap(); + + 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 cb47519..ef9ce44 100644 --- a/src/hello.hpp +++ b/src/hello.hpp @@ -1,17 +1,34 @@ #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; }; } // 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..9b7c0b9 --- /dev/null +++ b/src/infra/components/components.cpp @@ -0,0 +1,57 @@ +#include "components.hpp" + +#include +#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 RegisterYdbComponents(userver::components::ComponentList& list) +{ + list.Append() + .Append() + .Append(); +} + +// Infra +void RegisterInfraComponents(userver::components::ComponentList& list) +{ + list.Append() + .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..f9912c9 --- /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 RegisterYdbComponents(userver::components::ComponentList&); + +// Infra +void RegisterInfraComponents(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..77536a5 --- /dev/null +++ b/src/infra/components/coordination/coordination_dist_lock_component.cpp @@ -0,0 +1,52 @@ +#include "coordination_dist_lock_component.hpp" + +#include + +#include +#include +#include + +#include + +namespace { + +//////////////////////////////////////////////////////////////////////////////// + +const auto DEFAULT_COORDINATION_PAUSE_SECONDS = std::chrono::seconds(10); // TODO replace with dynamic config + +//////////////////////////////////////////////////////////////////////////////// + +} // anonymous namespace + +namespace NCoordinator::NInfra::NComponents { + +//////////////////////////////////////////////////////////////////////////////// + +TCoordinationDistLockComponent::TCoordinationDistLockComponent( + const userver::components::ComponentConfig& config, + const userver::components::ComponentContext& context) + : DistLockComponentBase(config, context) + , Gateway_(context.FindComponent().GetGateway()) +{ + Start(); +} + +TCoordinationDistLockComponent::~TCoordinationDistLockComponent() +{ + Stop(); +} + +void TCoordinationDistLockComponent::DoWork() { + while (!userver::engine::current_task::ShouldCancel()) { + auto partitionMap = Gateway_.GetPartitionMap().value_or(NCore::NDomain::TPartitionMap{}); + + ++(partitionMap.Epoch.GetUnderlying()); + Gateway_.BroadcastPartitionMap(partitionMap); + + userver::engine::InterruptibleSleepFor(DEFAULT_COORDINATION_PAUSE_SECONDS); // TODO replace with dynamic config + } +} + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NCoordinator::NInfra::NComponents diff --git a/src/infra/components/coordination/coordination_dist_lock_component.hpp b/src/infra/components/coordination/coordination_dist_lock_component.hpp new file mode 100644 index 0000000..5c9738b --- /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::NDomain::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..7a91cbd --- /dev/null +++ b/src/infra/components/coordination/coordination_gateway_component.cpp @@ -0,0 +1,70 @@ +#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 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( + std::move(coordinationClient), + coordinationNode, + partitionMapSemaphore, + discoverySemaphore, + initialSetup); +} + +NCore::NDomain::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 + partition-map-semaphore: + type: string + 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 + 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..b7ba6c6 --- /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::NDomain::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_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/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/coordination_gateway/kesus_coordination_gateway.hpp b/src/infra/coordination_gateway/kesus_coordination_gateway.hpp new file mode 100644 index 0000000..f4c3db4 --- /dev/null +++ b/src/infra/coordination_gateway/kesus_coordination_gateway.hpp @@ -0,0 +1,45 @@ +#pragma once + +#include +#include +#include + +#include + +#include +#include + +namespace NCoordinator::NInfra::NGateway { + +//////////////////////////////////////////////////////////////////////////////// + +class TKesusCoordinationGateway + : public NCore::NDomain::ICoordinationGateway +{ +public: + TKesusCoordinationGateway( + std::shared_ptr coordinationClient, + const std::string& coordinationNode, + const std::string& partitionMapSemaphore, + const std::string& discoverySemaphore, + const bool initialSetup); + + 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 PartitionMapSemaphore_; + const std::string DiscoverySemaphore_; + std::unique_ptr CoordinationSession_; +}; + +//////////////////////////////////////////////////////////////////////////////// + +} // namespace NCoordinator::NInfra::NGateway 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/serializer/serializer.cpp b/src/infra/serializer/serializer.cpp new file mode 100644 index 0000000..eb4920b --- /dev/null +++ b/src/infra/serializer/serializer.cpp @@ -0,0 +1,99 @@ +#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; +} + +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 new file mode 100644 index 0000000..bbfd94f --- /dev/null +++ b/src/infra/serializer/serializer.hpp @@ -0,0 +1,23 @@ +#pragma once + +#include +#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); + +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 8ef9820..c18ca38 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::RegisterYdbComponents(component_list); + + NCoordinator::NInfra::NComponents::RegisterInfraComponents(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..1f4a9fa 100644 --- a/tests/test_basic.py +++ b/tests/test_basic.py @@ -4,4 +4,5 @@ async def test_basic(service_client): 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 == '{"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