diff --git a/.mapping.json b/.mapping.json index 19cc4b6b105e..afe241e9c1f9 100644 --- a/.mapping.json +++ b/.mapping.json @@ -217,7 +217,7 @@ "chaotic/include/userver/chaotic/dynamic_config_variable_bundle.hpp":"taxi/uservices/userver/chaotic/include/userver/chaotic/dynamic_config_variable_bundle.hpp", "chaotic/include/userver/chaotic/exception.hpp":"taxi/uservices/userver/chaotic/include/userver/chaotic/exception.hpp", "chaotic/include/userver/chaotic/io/boost/uuids/uuid.hpp":"taxi/uservices/userver/chaotic/include/userver/chaotic/io/boost/uuids/uuid.hpp", - "chaotic/include/userver/chaotic/io/crypto/base64/string64.hpp":"taxi/uservices/userver/chaotic/include/userver/chaotic/io/crypto/base64/string64.hpp", + "chaotic/include/userver/chaotic/io/userver/crypto/base64/string64.hpp":"taxi/uservices/userver/chaotic/include/userver/chaotic/io/crypto/base64/string64.hpp", "chaotic/include/userver/chaotic/io/decimal64/decimal.hpp":"taxi/uservices/userver/chaotic/include/userver/chaotic/io/decimal64/decimal.hpp", "chaotic/include/userver/chaotic/io/std/chrono/days.hpp":"taxi/uservices/userver/chaotic/include/userver/chaotic/io/std/chrono/days.hpp", "chaotic/include/userver/chaotic/io/std/chrono/duration.hpp":"taxi/uservices/userver/chaotic/include/userver/chaotic/io/std/chrono/duration.hpp", diff --git a/CMakeLists.txt b/CMakeLists.txt index fe708131cdfd..23cbf07ba037 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -148,6 +148,7 @@ option(USERVER_FEATURE_YDB "Provide asynchronous driver for YDB" "${USERVER_YDB_ option(USERVER_FEATURE_OTLP "Provide asynchronous OTLP exporters" "${USERVER_LIB_ENABLED_DEFAULT}") option(USERVER_FEATURE_SQLITE "Provide asynchronous driver for SQLite" "${USERVER_LIB_ENABLED_DEFAULT}") option(USERVER_FEATURE_ODBC "Provide asynchronous wrapper around ODBC" "${USERVER_LIB_ENABLED_DEFAULT}") +option(USERVER_FEATURE_ETCD "Provide asynchronous driver for etcd" "${USERVER_LIB_ENABLED_DEFAULT}") set(CMAKE_DEBUG_POSTFIX d) @@ -312,6 +313,11 @@ if (USERVER_FEATURE_ODBC) list(APPEND USERVER_AVAILABLE_COMPONENTS odbc) endif() +if (USERVER_FEATURE_ETCD) + _require_userver_core("USERVER_FEATURE_ETCD") + add_subdirectory(etcd) +endif() + add_subdirectory(libraries) if (USERVER_BUILD_TESTS) diff --git a/chaotic/chaotic/back/cpp/translator.py b/chaotic/chaotic/back/cpp/translator.py index 08ab028bb6e9..2e9fca6862d7 100644 --- a/chaotic/chaotic/back/cpp/translator.py +++ b/chaotic/chaotic/back/cpp/translator.py @@ -444,7 +444,7 @@ def _gen_string( if schema.format and schema.format != types.StringFormat.BINARY: if schema.format == types.StringFormat.BYTE: - format_cpp_type = 'crypto::base64::String64' + format_cpp_type = 'userver::crypto::base64::String64' elif schema.format == types.StringFormat.UUID: format_cpp_type = 'boost::uuids::uuid' elif schema.format == types.StringFormat.DATE: diff --git a/chaotic/include/userver/chaotic/io/crypto/base64/string64.hpp b/chaotic/include/userver/chaotic/io/userver/crypto/base64/string64.hpp similarity index 100% rename from chaotic/include/userver/chaotic/io/crypto/base64/string64.hpp rename to chaotic/include/userver/chaotic/io/userver/crypto/base64/string64.hpp index 7384dbebce64..cf1dfd27a2ae 100644 --- a/chaotic/include/userver/chaotic/io/crypto/base64/string64.hpp +++ b/chaotic/include/userver/chaotic/io/userver/crypto/base64/string64.hpp @@ -5,6 +5,8 @@ #include +USERVER_NAMESPACE_BEGIN + namespace crypto::base64 { // RFC4648 @@ -14,8 +16,6 @@ class String64 : public USERVER_NAMESPACE::utils::StrongTypedef); diff --git a/chaotic/src/chaotic/io/crypto/base64/string64.cpp b/chaotic/src/chaotic/io/crypto/base64/string64.cpp index f05dc10bfe02..1a6b1dc50d5b 100644 --- a/chaotic/src/chaotic/io/crypto/base64/string64.cpp +++ b/chaotic/src/chaotic/io/crypto/base64/string64.cpp @@ -1,4 +1,4 @@ -#include +#include #include @@ -6,11 +6,11 @@ USERVER_NAMESPACE_BEGIN namespace chaotic::convert { -::crypto::base64::String64 Convert(const std::string& str, chaotic::convert::To<::crypto::base64::String64>) { - return ::crypto::base64::String64(crypto::base64::Base64Decode(str)); +crypto::base64::String64 Convert(const std::string& str, chaotic::convert::To) { + return crypto::base64::String64(crypto::base64::Base64Decode(str)); } -std::string Convert(const ::crypto::base64::String64& str64, chaotic::convert::To) { +std::string Convert(const crypto::base64::String64& str64, chaotic::convert::To) { return crypto::base64::Base64Encode(str64.GetUnderlying()); } diff --git a/chaotic/src/chaotic/io/userver/crypto/base64/string64.cpp b/chaotic/src/chaotic/io/userver/crypto/base64/string64.cpp new file mode 100644 index 000000000000..1a6b1dc50d5b --- /dev/null +++ b/chaotic/src/chaotic/io/userver/crypto/base64/string64.cpp @@ -0,0 +1,19 @@ +#include + +#include + +USERVER_NAMESPACE_BEGIN + +namespace chaotic::convert { + +crypto::base64::String64 Convert(const std::string& str, chaotic::convert::To) { + return crypto::base64::String64(crypto::base64::Base64Decode(str)); +} + +std::string Convert(const crypto::base64::String64& str64, chaotic::convert::To) { + return crypto::base64::Base64Encode(str64.GetUnderlying()); +} + +} // namespace chaotic::convert + +USERVER_NAMESPACE_END diff --git a/chaotic/tests/back/cpp/test_tr_string.py b/chaotic/tests/back/cpp/test_tr_string.py index 28617124637d..dc019ff617e8 100644 --- a/chaotic/tests/back/cpp/test_tr_string.py +++ b/chaotic/tests/back/cpp/test_tr_string.py @@ -54,7 +54,7 @@ def test_byte(simple_gen): assert types == { '::type': cpp_types.CppStringWithFormat( raw_cpp_type=type_name.TypeName('std::string'), - format_cpp_type='crypto::base64::String64', + format_cpp_type='userver::crypto::base64::String64', user_cpp_type=None, json_schema=None, nullable=False, diff --git a/cmake/install/userver-etcd-config.cmake b/cmake/install/userver-etcd-config.cmake new file mode 100644 index 000000000000..0d4826b3ab69 --- /dev/null +++ b/cmake/install/userver-etcd-config.cmake @@ -0,0 +1,11 @@ +include_guard(GLOBAL) + +if(userver_etcd_FOUND) + return() +endif() + +find_package(userver REQUIRED COMPONENTS + core +) + +set(userver_etcd_FOUND TRUE) diff --git a/etcd/CMakeLists.txt b/etcd/CMakeLists.txt new file mode 100644 index 000000000000..b94a0168d9b9 --- /dev/null +++ b/etcd/CMakeLists.txt @@ -0,0 +1,24 @@ +project(userver-etcd CXX) + +file(GLOB_RECURSE SCHEMAS ${CMAKE_CURRENT_SOURCE_DIR}/schemas/*.yaml) +userver_target_generate_chaotic(${PROJECT_NAME}-chgen + GENERATE_SERIALIZERS + LAYOUT + "/components/schemas/([^/]*)/=etcd_schemas::{0}" + OUTPUT_DIR + ${CMAKE_CURRENT_BINARY_DIR}/src + SCHEMAS + ${SCHEMAS} + RELATIVE_TO + ${CMAKE_CURRENT_SOURCE_DIR} +) + +userver_module(etcd + SOURCE_DIR "${CMAKE_CURRENT_SOURCE_DIR}" + UTEST_SOURCES "${CMAKE_CURRENT_SOURCE_DIR}/src/*_test.cpp" + LINK_LIBRARIES ${PROJECT_NAME}-chgen +) + +if (USERVER_BUILD_TESTS) + add_subdirectory(functional_tests) +endif() diff --git a/etcd/functional_tests/CMakeLists.txt b/etcd/functional_tests/CMakeLists.txt new file mode 100644 index 000000000000..4f24d80d6503 --- /dev/null +++ b/etcd/functional_tests/CMakeLists.txt @@ -0,0 +1,6 @@ +project(userver-etcd-tests CXX) + +add_custom_target(${PROJECT_NAME}) + +add_subdirectory(client) +add_dependencies(${PROJECT_NAME} ${PROJECT_NAME}-client) diff --git a/etcd/functional_tests/client/CMakeLists.txt b/etcd/functional_tests/client/CMakeLists.txt new file mode 100644 index 000000000000..76b956f0c704 --- /dev/null +++ b/etcd/functional_tests/client/CMakeLists.txt @@ -0,0 +1,6 @@ +project(userver-etcd-tests-client CXX) + +add_executable(${PROJECT_NAME} "etcd_service.cpp") +target_link_libraries(${PROJECT_NAME} userver-core userver-etcd) + +userver_chaos_testsuite_add() diff --git a/etcd/functional_tests/client/etcd_service.cpp b/etcd/functional_tests/client/etcd_service.cpp new file mode 100644 index 000000000000..ed402b818da4 --- /dev/null +++ b/etcd/functional_tests/client/etcd_service.cpp @@ -0,0 +1,92 @@ +#include + +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace { + +class HandlerV1Get final : public server::handlers::HttpHandlerBase { +public: + static constexpr std::string_view kName = "handler-v1-get"; + + HandlerV1Get(const components::ComponentConfig& config, const components::ComponentContext& component_context) + : HttpHandlerBase(config, component_context), + etcd_client_ptr_(component_context.FindComponent("etcd-client").GetClient()) {} + + std::string HandleRequestThrow(const server::http::HttpRequest& request, server::request::RequestContext&) + const override { + const auto maybe_value = etcd_client_ptr_->Get(request.GetArg("key")); + return maybe_value.value_or("No value"); + } + +private: + etcd::ClientPtr etcd_client_ptr_; +}; + +class HandlerV1Put final : public server::handlers::HttpHandlerBase { +public: + static constexpr std::string_view kName = "handler-v1-put"; + + HandlerV1Put(const components::ComponentConfig& config, const components::ComponentContext& component_context) + : HttpHandlerBase(config, component_context), + etcd_client_ptr_(component_context.FindComponent("etcd-client").GetClient()) {} + + std::string HandleRequestThrow(const server::http::HttpRequest& request, server::request::RequestContext&) + const override { + etcd_client_ptr_->Put(request.GetArg("key"), request.GetArg("value")); + + return std::string(); + } + +private: + etcd::ClientPtr etcd_client_ptr_; +}; + +class HandlerV1Watch final : public server::handlers::HttpHandlerBase { +public: + static constexpr std::string_view kName = "handler-v1-watch"; + + HandlerV1Watch(const components::ComponentConfig& config, const components::ComponentContext& component_context) + : HttpHandlerBase(config, component_context), + etcd_client_ptr_(component_context.FindComponent("etcd-client").GetClient()) {} + + std::string HandleRequestThrow(const server::http::HttpRequest& request, server::request::RequestContext&) + const override { + const auto key = request.GetArg("key"); + const auto maybe_original_value = etcd_client_ptr_->Get(key); + auto watch_listener = etcd_client_ptr_->StartWatch(key); + const auto watch_event = watch_listener.GetEvent(); + const auto new_value = watch_event.value; + return fmt::format("original value: {}, new value: {}", maybe_original_value.value_or("No value"), new_value); + } + +private: + etcd::ClientPtr etcd_client_ptr_; +}; + +} // namespace + +int main(int argc, char* argv[]) { + auto component_list = components::MinimalServerComponentList() + .Append() + .Append() + .Append() + .Append() + .Append() + .Append() + .Append() + .Append() + .Append(); + + return utils::DaemonMain(argc, argv, component_list); +} diff --git a/etcd/functional_tests/client/static_config.yaml b/etcd/functional_tests/client/static_config.yaml new file mode 100644 index 000000000000..6ddd00996d71 --- /dev/null +++ b/etcd/functional_tests/client/static_config.yaml @@ -0,0 +1,76 @@ +# yaml +components_manager: + task_processors: + main-task-processor: + worker_threads: 4 + + fs-task-processor: + worker_threads: 1 + + default_task_processor: main-task-processor + + components: + server: + listener: + port: 8080 + task_processor: main-task-processor + connection: + http-version: '2' + http2-session: + max_concurrent_streams: 100 + max_frame_size: 16384 + initial_window_size: 65536 + listener-monitor: + port: 8081 + task_processor: main-task-processor + logging: + fs-task-processor: fs-task-processor + loggers: + default: + file_path: '@stderr' + level: error + overflow_behavior: discard + + http-client: + fs-task-processor: fs-task-processor + user-agent: $server-name + user-agent#fallback: 'userver-based-service 1.0' + dns-client: + fs-task-processor: fs-task-processor + testsuite-support: + tests-control: + path: /tests/{action} + method: POST + task_processor: main-task-processor + testpoint-timeout: 10s + testpoint-url: $mockserver/testpoint + throttling_enabled: false + + handler-ping: + path: /ping + method: GET + task_processor: main-task-processor + throttling_enabled: false + url_trailing_slash: strict-match + + etcd-client: + endpoints: + - http://localhost:2379 + attempts: 2 + request_timeout_ms: 500 + watch_timeout_ms: 1000000 + + handler-v1-get: + path: /v1/get + method: POST + task_processor: main-task-processor + + handler-v1-put: + path: /v1/put + method: PUT + task_processor: main-task-processor + + handler-v1-watch: + path: /v1/watch + method: POST + task_processor: main-task-processor diff --git a/etcd/functional_tests/client/tests/conftest.py b/etcd/functional_tests/client/tests/conftest.py new file mode 100644 index 000000000000..0c8418fa9565 --- /dev/null +++ b/etcd/functional_tests/client/tests/conftest.py @@ -0,0 +1,102 @@ +import asyncio +import base64 +import json + +import aiohttp + +import pytest + +pytest_plugins = ['pytest_userver.plugins.core'] + + +@pytest.fixture(scope='session') +def userver_config_http_client( + mockserver_info, + mockserver_ssl_info, + allowed_url_prefixes_extra, +): + def patch_config(config, config_vars): + components: dict = config['components_manager']['components'] + + http_client = components['http-client'] or {} + http_client['testsuite-enabled'] = False + + etcd_client = components['etcd-client'] or {} + etcd_client['endpoints'] = [mockserver_info.base_url[:-1]] + + allowed_urls = [mockserver_info.base_url] + if mockserver_ssl_info: + allowed_urls.append(mockserver_ssl_info.base_url) + allowed_urls += allowed_url_prefixes_extra + http_client['testsuite-allowed-url-prefixes'] = allowed_urls + + return patch_config + + +@pytest.fixture(name='etcd_mock') +def etcd_mock(mockserver): + etcd_storage = {} + + @mockserver.json_handler('/v3/kv/put') + async def mock(request): + key = base64.b64decode(request.json['key']) + value = base64.b64decode(request.json['value']) + etcd_storage[key] = value + return mockserver.make_response('OK!') + + @mockserver.json_handler('/v3/kv/range') + async def mock(request): + request_key = base64.b64decode(request.json['key']) + if 'range_end' in request.json: + request_range_end = base64.b64decode(request.json['range_end']) + elif request_key in etcd_storage: + return mockserver.make_response(json={ + 'kvs': [{ + 'key': base64.b64encode(request_key).decode('utf-8'), + 'value': base64.b64encode(etcd_storage[request_key]).decode('utf-8'), + 'version': '2', + }] + }) + else: + return mockserver.make_response(json={'kvs': []}) + + values = [] + for key, value in etcd_storage.items(): + if key >= request_key and key < request_range_end: + values.append({ + 'key': base64.b64encode(key).decode('utf-8'), + 'value': base64.b64encode(value).decode('utf-8'), + 'version': '2', + }) + + return mockserver.make_response(json={ + 'kvs': values + }) + + @mockserver.handler('/v3/watch') + async def mock(request): + key = base64.b64decode(request.json['create_request']['key']) + response = aiohttp.web.StreamResponse( + status=200, + headers={ + 'Content-Type': 'application/json', + } + ) + await response.prepare(request._request) + data = json.dumps({ + 'result': { + 'events': [ + { + 'kv': { + 'key': base64.b64encode(key).decode(), + 'value': base64.b64encode(b'new_value').decode(), + 'version': '2', + } + } + ] + } + }) + await response.write(data.encode()) + await asyncio.sleep(0.01) + + return response diff --git a/etcd/functional_tests/client/tests/test_etcd_client.py b/etcd/functional_tests/client/tests/test_etcd_client.py new file mode 100644 index 000000000000..7105f415b351 --- /dev/null +++ b/etcd/functional_tests/client/tests/test_etcd_client.py @@ -0,0 +1,35 @@ +async def test_etcd_put_get(service_client, etcd_mock): + response = await service_client.post( + '/v1/get', + params={'key': 'some_key'} + ) + assert response.status == 200 + assert response.content == b'No value' + + response = await service_client.put( + '/v1/put', + params={'key': 'some_key', 'value': 'some_value'} + ) + assert response.status == 200 + + response = await service_client.post( + '/v1/get', + params={'key': 'some_key'} + ) + assert response.status == 200 + assert response.content == b'some_value' + + +async def test_etcd_watch(service_client, etcd_mock): + response = await service_client.put( + '/v1/put', + params={'key': 'some_key', 'value': 'original_value'} + ) + assert response.status == 200 + + response = await service_client.post( + '/v1/watch', + params={'key': 'some_key'} + ) + assert response.status == 200 + assert response.content == b'original value: original_value, new value: new_value' diff --git a/etcd/include/userver/etcd/client.hpp b/etcd/include/userver/etcd/client.hpp new file mode 100644 index 000000000000..aa5a1a743c32 --- /dev/null +++ b/etcd/include/userver/etcd/client.hpp @@ -0,0 +1,62 @@ +#pragma once + +/// @file userver/etcd/client.hpp +/// @brief @copybrief etcd::Client + +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include + +#include + +USERVER_NAMESPACE_BEGIN + +namespace etcd { + +/// @brief Etcd client implemented using http client +class Client { +public: + virtual ~Client() = default; + + /// @brief Puts a key value pair into etcd cluster. + /// The pair should be retrieve only with current client, + /// because Put can transform the key. + /// + virtual void Put(std::string_view key, std::string_view value) = 0; + + /// @brief Gets a value from etcd cluster by a key. + /// If there is no value with such key, returns std::nullopt + /// + [[nodiscard]] virtual std::optional Get(std::string_view key) = 0; + + /// @brief Retrieves key values pairs from the etcd cluster, + /// the keys of which match the passed prefix. + /// If there is no value with such key, returns empty vector + /// + [[nodiscard]] virtual std::vector Range(std::string_view key_prefix) = 0; + + /// @brief Delete a key value pair with the passed key + /// from the etcd cluster + /// + virtual void Delete(std::string_view key) = 0; + + /// @brief Start task that produces events when key value pair changes + /// + virtual WatchListener StartWatch(std::string_view key) = 0; +}; + +using ClientPtr = std::shared_ptr; + +} // namespace etcd + +USERVER_NAMESPACE_END diff --git a/etcd/include/userver/etcd/component.hpp b/etcd/include/userver/etcd/component.hpp new file mode 100644 index 000000000000..3ea243235031 --- /dev/null +++ b/etcd/include/userver/etcd/component.hpp @@ -0,0 +1,48 @@ +#pragma once + +/// @file userver/etcd/component.hpp +/// @brief @copybrief etcd::Component + +#include +#include +#include +#include + +USERVER_NAMESPACE_BEGIN + +namespace etcd { + +// clang-format off +/// @ingroup userver_components +/// +/// @brief Etcd client component +/// +/// Provides access to a etcd cluster. +/// +/// ## Static options: +/// Name | Description | Default value +/// ---------------------------------- | ---------------------------------------------------------- | --------------- +/// endpoints | Etcd endpoints to which client make HTTP requests | - +/// attempts | Number of attempts to each endpoint, on failed attempts client randomly moves to another endpoint | 3 +/// request_timeout_ms | Timeout for all HTTP requests to etcd except watch request | 1000 +/// watch_timeout_ms | Timeout for watch HTTP request. It's a stremed request, so it is used also as a connection timeout, so it should not be too short | 1000000 +/// +// clang-format on + +class Component final : public components::ComponentBase { +public: + static constexpr std::string_view kName = "etcd-client"; + + Component(const components::ComponentConfig&, const components::ComponentContext&); + + static yaml_config::Schema GetStaticConfigSchema(); + + ClientPtr GetClient(); + +private: + const ClientPtr etcd_client_ptr_; +}; + +} // namespace etcd + +USERVER_NAMESPACE_END diff --git a/etcd/include/userver/etcd/exceptions.hpp b/etcd/include/userver/etcd/exceptions.hpp new file mode 100644 index 000000000000..c2fab3a76287 --- /dev/null +++ b/etcd/include/userver/etcd/exceptions.hpp @@ -0,0 +1,34 @@ +#pragma once + +/// @file userver/etcd/exceptions.hpp +/// @brief Exceptions thrown by etcd client + +#include + +#include + +USERVER_NAMESPACE_BEGIN + +namespace etcd { + +/// @brief Base class for all etcd client exceptions +class EtcdError : public std::runtime_error { +public: + using std::runtime_error::runtime_error; +}; + +/// @brief Error during a request to etcd +class EtcdRequestError : public EtcdError { +public: + using EtcdError::EtcdError; +}; + +/// @brief Error during parsing of etcd watch response +class EtcdWatchResponseParseError : public EtcdError { +public: + using EtcdError::EtcdError; +}; + +} // namespace etcd + +USERVER_NAMESPACE_END diff --git a/etcd/include/userver/etcd/key_value_state.hpp b/etcd/include/userver/etcd/key_value_state.hpp new file mode 100644 index 000000000000..4a9a5fc7d12f --- /dev/null +++ b/etcd/include/userver/etcd/key_value_state.hpp @@ -0,0 +1,21 @@ +#pragma once + +/// @file userver/etcd/key_value_state.hpp +/// @brief @copybrief etcd::KeyValueState + +#include + +USERVER_NAMESPACE_BEGIN + +namespace etcd { + +/// @brief Struct with key value pair from etcd. It represents current status of key value pair. +struct KeyValueState final { + std::string key; + std::string value; + std::int32_t version; +}; + +} // namespace etcd + +USERVER_NAMESPACE_END diff --git a/etcd/include/userver/etcd/settings.hpp b/etcd/include/userver/etcd/settings.hpp new file mode 100644 index 000000000000..7a8b147a2674 --- /dev/null +++ b/etcd/include/userver/etcd/settings.hpp @@ -0,0 +1,37 @@ +#pragma once + +/// @file userver/etcd/settings.hpp +/// @brief etcd client settings + +#include +#include +#include + +#include + +USERVER_NAMESPACE_BEGIN + +namespace etcd { + +/// @brief Etcd client settigs struct +struct ClientSettings final { + // Etcd endpoints to which client make HTTP requests + std::vector endpoints; + // Number of attempts to each endpoint, on failed attempts client randomly moves to another endpoint + std::uint32_t attempts; + // Timeout for all HTTP requests to etcd except watch request + std::chrono::microseconds request_timeout_ms; + // Timeout for watch HTTP request. It's a stremed request, so it is used also as a connection timeout, so it should + // not be too short + std::chrono::microseconds watch_timeout_ms; +}; + +} // namespace etcd + +namespace formats::parse { + +etcd::ClientSettings Parse(const yaml_config::YamlConfig& value, To); + +} + +USERVER_NAMESPACE_END diff --git a/etcd/include/userver/etcd/watch_listener.hpp b/etcd/include/userver/etcd/watch_listener.hpp new file mode 100644 index 000000000000..d0ea1d59ee92 --- /dev/null +++ b/etcd/include/userver/etcd/watch_listener.hpp @@ -0,0 +1,31 @@ +#pragma once + +/// @file userver/etcd/watch_listener.hpp +/// @brief Queue with value change events in etcd + +#include + +#include +#include +#include + +USERVER_NAMESPACE_BEGIN + +namespace etcd { + +/// @brief Struct that return value change events in etcd +class WatchListener final { +public: + WatchListener(concurrent::SpscQueue::Consumer&& consumer); + + /// @brief Get an event from etcd if there was one, otherwise waits asynchronously until a next event occurs. + /// @throws EtcdError if event producing coroutine finished or failed + KeyValueState GetEvent(); + +private: + concurrent::SpscQueue::Consumer consumer_; +}; + +} // namespace etcd + +USERVER_NAMESPACE_END diff --git a/etcd/library.yaml b/etcd/library.yaml new file mode 100644 index 000000000000..0c3a52686ba2 --- /dev/null +++ b/etcd/library.yaml @@ -0,0 +1,9 @@ +project-name: userver-etcd +project-alt-names: + - yandex-userver-etcd +maintainers: + - Common components +description: Etcd client + +libraries: + - userver-core diff --git a/etcd/schemas/etcd.yaml b/etcd/schemas/etcd.yaml new file mode 100644 index 000000000000..5a04432a1ec4 --- /dev/null +++ b/etcd/schemas/etcd.yaml @@ -0,0 +1,10 @@ +components: + schemas: + EtcdRangeResponse: + type: object + additionalProperties: true + properties: + kvs: + type: array + items: + $ref: 'types.yaml#/components/schemas/RawKeyValueState' diff --git a/etcd/schemas/types.yaml b/etcd/schemas/types.yaml new file mode 100644 index 000000000000..c69dff479c69 --- /dev/null +++ b/etcd/schemas/types.yaml @@ -0,0 +1,18 @@ +components: + schemas: + RawKeyValueState: + type: object + additionalProperties: true + required: + - key + - value + - version + properties: + key: + type: string + format: byte + value: + type: string + format: byte + version: + type: string diff --git a/etcd/src/etcd/client_impl.cpp b/etcd/src/etcd/client_impl.cpp new file mode 100644 index 000000000000..eb0c3931bfa2 --- /dev/null +++ b/etcd/src/etcd/client_impl.cpp @@ -0,0 +1,254 @@ +#include + +#include +#include +#include + +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#include + +USERVER_NAMESPACE_BEGIN + +namespace etcd { + +namespace { + +const std::uint32_t kMinRetryStatusCode = 500; +const std::uint32_t kMaxRetryStatusCode = 599; + +const std::uint32_t kMinGoodStatusCode = 200; +const std::uint32_t kMaxGoodStatusCode = 299; + +const std::string kKeyPrefix = "/etcd/"; + +KeyValueState ConvertRawKeyValueState(const etcd_schemas::RawKeyValueState& raw_key_value_state) { + KeyValueState key_value_state; + key_value_state.key = raw_key_value_state.key.GetUnderlying().substr(kKeyPrefix.size()); + key_value_state.value = raw_key_value_state.value.GetUnderlying(); + key_value_state.version = std::stoi(raw_key_value_state.version); + return key_value_state; +} + +std::string BuildPutUrl(std::string_view service_url) { return fmt::format("{}/v3/kv/put", service_url); } + +std::string BuildPutData(std::string_view key, std::string_view value) { + const auto etcd_key = fmt::format("{}{}", kKeyPrefix, key); + return formats::json::ToString(formats::json::MakeObject( + "key", crypto::base64::Base64Encode(etcd_key), "value", crypto::base64::Base64Encode(value) + )); +} + +std::string BuildRangeUrl(std::string_view service_url) { return fmt::format("{}/v3/kv/range", service_url); } + +std::string BuildRangeData(std::string_view key, const std::optional maybe_range_end = std::nullopt) { + const auto etcd_key = fmt::format("{}{}", kKeyPrefix, key); + if (!maybe_range_end.has_value()) { + return formats::json::ToString(formats::json::MakeObject("key", crypto::base64::Base64Encode(etcd_key))); + } + const auto etcd_range_end = fmt::format("{}{}", kKeyPrefix, maybe_range_end.value()); + return formats::json::ToString(formats::json::MakeObject( + "key", crypto::base64::Base64Encode(etcd_key), "range_end", crypto::base64::Base64Encode(etcd_range_end) + )); +} + +std::string BuildDeleteUrl(std::string_view service_url) { return fmt::format("{}/v3/kv/deleterange", service_url); } + +std::string BuildDeleteData(std::string_view key) { + const auto etcd_key = fmt::format("{}{}", kKeyPrefix, key); + return formats::json::ToString(formats::json::MakeObject("key", crypto::base64::Base64Encode(etcd_key))); +} + +std::string BuildWatchUrl(std::string_view service_url) { return fmt::format("{}/v3/watch", service_url); } + +std::string BuildWatchData(std::string_view key) { + const auto etcd_key = fmt::format("{}{}", kKeyPrefix, key); + return formats::json::ToString(formats::json::MakeObject( + "create_request", formats::json::MakeObject("key", crypto::base64::Base64Encode(etcd_key)) + )); +} + +bool ShouldRetry(const http::StatusCode status_code) { + return kMinRetryStatusCode <= status_code && status_code <= kMaxRetryStatusCode; +} + +void CheckResponseStatusCode(const http::StatusCode status_code, std::string_view body) { + if (status_code < kMinGoodStatusCode || kMaxGoodStatusCode < status_code) { + throw EtcdRequestError(fmt::format("Got bad status code from etcd: {}, body: {}", status_code, body)); + } +} + +} // namespace + +namespace impl { + +ClientImpl::ClientImpl(clients::http::Client& http_client, ClientSettings settings) + : http_client_(http_client), settings_(settings) {} + +void ClientImpl::Put(std::string_view key, std::string_view value) { + PerformEtcdRequest(BuildPutUrl, BuildPutData(key, value)); +} + +void ClientImpl::Delete(std::string_view key) { PerformEtcdRequest(BuildDeleteUrl, BuildDeleteData(key)); } + +std::optional ClientImpl::Get(std::string_view key) { + auto response = PerformEtcdRequest(BuildRangeUrl, BuildRangeData(key)); + + const auto maybe_range_response = formats::json::FromString(response.body()).As(); + if (!maybe_range_response.kvs.has_value()) { + return std::nullopt; + } + for (const auto& raw_key_value_state : maybe_range_response.kvs.value()) { + const auto key_value_state = ConvertRawKeyValueState(raw_key_value_state); + if (key_value_state.key == key) { + return key_value_state.value; + } + } + return std::nullopt; +} + +std::vector ClientImpl::Range(std::string_view key_prefix) { + const auto response = + PerformEtcdRequest(BuildRangeUrl, BuildRangeData(key_prefix, fmt::format("{}\xFF", key_prefix))); + + const auto maybe_range_response = formats::json::FromString(response.body()).As(); + if (!maybe_range_response.kvs.has_value()) { + return {}; + } + + std::vector range_result; + range_result.reserve(maybe_range_response.kvs.value().size()); + for (const auto& raw_key_value_state : maybe_range_response.kvs.value()) { + const auto key_value_state = ConvertRawKeyValueState(raw_key_value_state); + range_result.push_back(key_value_state); + } + return range_result; +} + +WatchListener ClientImpl::StartWatch(std::string_view key) { + auto queue = concurrent::SpscQueue::Create(); + + auto watch_queues_ptr = watch_queues_.Lock(); + watch_queues_ptr->push_back(queue); + + bts_.AsyncDetach("watch task", [string_key = std::string(key), producer = queue->GetProducer(), this]() mutable { + this->WatchKeyChanges(string_key, std::move(producer)); + }); + + return WatchListener{queue->GetConsumer()}; +} + +clients::http::Response +ClientImpl::PerformEtcdRequest(const std::function& url_builder, std::string_view data) { + auto endpoints = settings_.endpoints; + utils::Shuffle(endpoints); + + std::optional maybe_response; + for (const auto& endpoint : endpoints) { + const auto response_ptr = http_client_.CreateRequest() + .post(url_builder(endpoint), std::string{data}) + .retry(settings_.attempts) + .timeout(settings_.request_timeout_ms.count()) + .perform(); + if (response_ptr == nullptr) { + LOG_WARNING() << "Perform request returns nullptr"; + continue; + } + maybe_response = *(response_ptr); + const auto& response = maybe_response.value(); + if (!ShouldRetry(response.status_code())) { + CheckResponseStatusCode(response.status_code(), response.body()); + return response; + } + } + LOG_ERROR() << "Request was not successful"; + if (maybe_response.has_value()) { + throw EtcdRequestError( + fmt::format("Failed to get Ok response from etcd with error: {}", maybe_response.value().body()) + ); + } else { + throw EtcdRequestError( + fmt::format("Failed to get streamed response, number of etcd endpoints: {}", endpoints.size()) + ); + } +} + +clients::http::StreamedResponse ClientImpl::PerformStreamedEtcdRequest( + const std::function& url_builder, + std::string_view data +) { + auto endpoints = settings_.endpoints; + utils::Shuffle(endpoints); + + std::optional maybe_streamed_response; + for (const auto& endpoint : endpoints) { + const auto queue = concurrent::StringStreamQueue::Create(); + maybe_streamed_response = http_client_.CreateRequest() + .post(url_builder(endpoint), std::string{data}) + .retry(settings_.attempts) + .timeout(settings_.watch_timeout_ms.count()) + .async_perform_stream_body(queue); + auto& streamed_response = maybe_streamed_response.value(); + if (!ShouldRetry(streamed_response.StatusCode())) { + CheckResponseStatusCode(streamed_response.StatusCode(), "There is no body in stream responses"); + return std::move(streamed_response); + } + } + if (maybe_streamed_response.has_value()) { + throw EtcdError(fmt::format( + "Failed to get Ok response from etcd with status code: {}", maybe_streamed_response.value().StatusCode() + )); + } else { + throw EtcdError(fmt::format("Failed to get streamed response, number of etcd endpoints: {}", endpoints.size())); + } +} + +void ClientImpl::WatchKeyChanges(const std::string key, concurrent::SpscQueue::Producer producer) { + const auto watch_data = BuildWatchData(key); + + while (true) { + LOG_DEBUG() << "Start whatching key changes"; + auto stream_response = PerformStreamedEtcdRequest(BuildWatchUrl, watch_data); + std::string body_part; + while (stream_response.ReadChunk(body_part, engine::Deadline())) { + EtcdWatchResponse etcd_watch_response; + try { + etcd_watch_response = formats::json::FromString(body_part).As(); + } catch (const EtcdWatchResponseParseError& error) { + LOG_DEBUG() << "Couldnot parse etcd response: " << error; + continue; + } + for (const auto& raw_key_value_state : etcd_watch_response.raw_key_value_states) { + auto key_value_state = ConvertRawKeyValueState(raw_key_value_state); + if (!producer.Push(std::move(key_value_state))) { + LOG_ERROR() << "Could not push to queue, aborting task"; + return; + }; + } + } + LOG_ERROR() << "Could not read chunk from stream response"; + } +} + +} // namespace impl + +} // namespace etcd + +USERVER_NAMESPACE_END diff --git a/etcd/src/etcd/client_impl.hpp b/etcd/src/etcd/client_impl.hpp new file mode 100644 index 000000000000..d54b07ba6d86 --- /dev/null +++ b/etcd/src/etcd/client_impl.hpp @@ -0,0 +1,50 @@ +#pragma once + +#include +#include +#include +#include +#include +#include + +USERVER_NAMESPACE_BEGIN + +namespace etcd { + +namespace impl { + +class ClientImpl : public Client { +public: + ClientImpl(clients::http::Client& http_client, ClientSettings settings); + + void Put(std::string_view key, std::string_view value) override; + + [[nodiscard]] std::optional Get(std::string_view key) override; + + [[nodiscard]] std::vector Range(std::string_view key_prefix) override; + + void Delete(std::string_view key) override; + + WatchListener StartWatch(std::string_view key) override; + +private: + clients::http::Response + PerformEtcdRequest(const std::function& url_builder, std::string_view data); + + [[nodiscard]] clients::http::StreamedResponse + PerformStreamedEtcdRequest(const std::function& url_builder, std::string_view data); + + void WatchKeyChanges(const std::string key, concurrent::SpscQueue::Producer producer); + + using WatchQueuePtr = std::shared_ptr>; + clients::http::Client& http_client_; + concurrent::Variable> watch_queues_; + concurrent::BackgroundTaskStorage bts_; + const ClientSettings settings_; +}; + +} // namespace impl + +} // namespace etcd + +USERVER_NAMESPACE_END diff --git a/etcd/src/etcd/component.cpp b/etcd/src/etcd/component.cpp new file mode 100644 index 000000000000..dae36c54c6cb --- /dev/null +++ b/etcd/src/etcd/component.cpp @@ -0,0 +1,52 @@ +#include + +#include +#include +#include +#include + +USERVER_NAMESPACE_BEGIN + +namespace etcd { + +Component::Component(const components::ComponentConfig& config, const components::ComponentContext& context) + : ComponentBase(config, context), + etcd_client_ptr_(std::make_shared( + context.FindComponent().GetHttpClient(), + config.As() + )) {} + +yaml_config::Schema Component::GetStaticConfigSchema() { + return yaml_config::MergeSchemas(R"( +type: object +description: Etcd cluster component +additionalProperties: false +properties: + endpoints: + type: array + description: Etcd endpoints to which client make HTTP requests + items: + type: string + description: Etcd endpoint, e.g. http://localhost:2379 + attempts: + type: integer + description: > + Number of attempts to each endpoint, on failed attempts client randomly moves to another endpoint + minimum: 1 + request_timeout_ms: + type: integer + description: Timeout for all HTTP requests to etcd except watch request + minimum: 1 + watch_timeout_ms: + type: integer + description: > + Timeout for watch HTTP request. It's a stremed request, so it is used also as a connection timeout, so it should not be too short + minimum: 1 +)"); +} + +ClientPtr Component::GetClient() { return etcd_client_ptr_; } + +} // namespace etcd + +USERVER_NAMESPACE_END diff --git a/etcd/src/etcd/etcd_client_test.cpp b/etcd/src/etcd/etcd_client_test.cpp new file mode 100644 index 000000000000..779312d63bc6 --- /dev/null +++ b/etcd/src/etcd/etcd_client_test.cpp @@ -0,0 +1,146 @@ +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +USERVER_NAMESPACE_BEGIN + +namespace { + +utest::HttpServerMock::HttpResponse EtcdRequestProcessor(const utest::HttpServerMock::HttpRequest& request) { + static std::map storage; + + EXPECT_EQ(request.method, clients::http::HttpMethod::kPost); + const auto request_body = formats::json::FromString(request.body); + formats::json::ValueBuilder response_body_value_builder; + const auto key = crypto::base64::Base64Decode(request_body["key"].As()); + + if (request.path == "/v3/kv/put") { + const auto value = crypto::base64::Base64Decode(request_body["value"].As()); + int32_t new_version = 1; + const auto key_value_iterator = storage.find(key); + if (key_value_iterator != storage.end()) { + new_version = (key_value_iterator->second).version + 1; + } + etcd::KeyValueState key_value_state; + key_value_state.key = key; + key_value_state.value = value; + key_value_state.version = new_version; + storage[key] = key_value_state; + + } else if (request.path == "/v3/kv/range" && !request_body.HasMember("range_end")) { + const auto value_iterator = storage.find(key); + response_body_value_builder["kvs"] = formats::json::MakeArray(); + if (value_iterator != storage.end()) { + response_body_value_builder["kvs"].PushBack(formats::json::MakeObject( + "key", + crypto::base64::Base64Encode((value_iterator->second).key), + "value", + crypto::base64::Base64Encode((value_iterator->second).value), + "version", + std::to_string((value_iterator->second).version) + )); + } + } else if (request.path == "/v3/kv/range") { + const auto range_end = crypto::base64::Base64Decode(request_body["range_end"].As()); + auto first_key = storage.lower_bound(key); + const auto last_key = storage.upper_bound(range_end); + + response_body_value_builder["kvs"] = formats::json::MakeArray(); + while (first_key != last_key) { + response_body_value_builder["kvs"].PushBack(formats::json::MakeObject( + "key", + crypto::base64::Base64Encode((first_key->second).key), + "value", + crypto::base64::Base64Encode((first_key->second).value), + "version", + std::to_string((first_key->second).version) + )); + ++first_key; + } + } else if (request.path == "/v3/kv/deleterange") { + storage.erase(key); + } + + return utest::HttpServerMock::HttpResponse{ + 200, clients::http::Headers{}, formats::json::ToString(response_body_value_builder.ExtractValue())}; +} + +} // namespace + +UTEST(Etcd, TestKeyValueStorage) { + utest::HttpServerMock mock_server(&EtcdRequestProcessor); + auto http_client_ptr = utest::CreateHttpClient(); + auto etcd_client_ptr = std::make_shared( + *http_client_ptr, + etcd::ClientSettings{ + {mock_server.GetBaseUrl()}, + 2, + std::chrono::milliseconds{500}, + std::chrono::milliseconds{100'000}, + } + ); + + const auto empty_value = etcd_client_ptr->Get("key_with_empty_value"); + EXPECT_EQ(empty_value, std::nullopt); + + etcd_client_ptr->Put("some_key", "some_value"); + const auto some_value = etcd_client_ptr->Get("some_key"); + EXPECT_EQ(some_value, "some_value"); + + etcd_client_ptr->Put("some_key", "some_new_value"); + const auto some_new_value = etcd_client_ptr->Get("some_key"); + EXPECT_EQ(some_new_value, "some_new_value"); + + etcd_client_ptr->Delete("some_key"); + const auto deleted_value = etcd_client_ptr->Get("some_key"); + EXPECT_EQ(deleted_value, std::nullopt); +} + +UTEST(Etcd, TestRange) { + utest::HttpServerMock mock_server(&EtcdRequestProcessor); + auto http_client_ptr = utest::CreateHttpClient(); + auto etcd_client_ptr = std::make_shared( + *http_client_ptr, + etcd::ClientSettings{ + {mock_server.GetBaseUrl()}, + 2, + std::chrono::milliseconds{500}, + std::chrono::milliseconds{100'000}, + } + ); + const uint32_t range_size = 3; + + EXPECT_TRUE(etcd_client_ptr->Range("some_key").empty()); + + for (uint32_t i = 1; i <= range_size; ++i) { + etcd_client_ptr->Put(fmt::format("some_key_{}", i), fmt::format("some_value_{}", i)); + } + + auto range_result = etcd_client_ptr->Range("some_key"); + EXPECT_EQ(range_result.size(), range_size); + std::sort(range_result.begin(), range_result.end(), [](const etcd::KeyValueState& l, const etcd::KeyValueState& r) { + return l.value < r.value; + }); + + for (uint32_t i = 1; i <= range_size; ++i) { + EXPECT_EQ(range_result[i - 1].value, fmt::format("some_value_{}", i)); + } + + for (uint32_t i = 1; i <= range_size; ++i) { + etcd_client_ptr->Delete(fmt::format("some_key_{}", i)); + } + + EXPECT_TRUE(etcd_client_ptr->Range("some_key").empty()); +} + +USERVER_NAMESPACE_END diff --git a/etcd/src/etcd/etcd_responses.cpp b/etcd/src/etcd/etcd_responses.cpp new file mode 100644 index 000000000000..04f2d13b265b --- /dev/null +++ b/etcd/src/etcd/etcd_responses.cpp @@ -0,0 +1,35 @@ +#include + +#include +#include +#include + +USERVER_NAMESPACE_BEGIN + +namespace formats::parse { + +etcd::EtcdWatchResponse Parse(const formats::json::Value& value, To) { + if (!value.HasMember("result")) { + throw etcd::EtcdWatchResponseParseError( + fmt::format("No result in watch response: {}", formats::json::ToString(value)) + ); + } + if (!value["result"].HasMember("events")) { + throw etcd::EtcdWatchResponseParseError( + fmt::format("No events in watch response: {}", formats::json::ToString(value)) + ); + } + etcd::EtcdWatchResponse etcd_watch_response; + for (const auto& event : value["result"]["events"]) { + if (!event.HasMember("kv")) { + LOG_DEBUG() << "Event is not key value change, skipping"; + continue; + } + etcd_watch_response.raw_key_value_states.push_back(event["kv"].As()); + } + return etcd_watch_response; +} + +} // namespace formats::parse + +USERVER_NAMESPACE_END diff --git a/etcd/src/etcd/etcd_responses.hpp b/etcd/src/etcd/etcd_responses.hpp new file mode 100644 index 000000000000..b0eae0df83ab --- /dev/null +++ b/etcd/src/etcd/etcd_responses.hpp @@ -0,0 +1,23 @@ +#pragma once + +#include + +#include + +USERVER_NAMESPACE_BEGIN + +namespace etcd { + +struct EtcdWatchResponse final { + std::vector raw_key_value_states; +}; + +} // namespace etcd + +namespace formats::parse { + +etcd::EtcdWatchResponse Parse(const formats::json::Value& value, To); + +} // namespace formats::parse + +USERVER_NAMESPACE_END diff --git a/etcd/src/etcd/settings.cpp b/etcd/src/etcd/settings.cpp new file mode 100644 index 000000000000..48943493831d --- /dev/null +++ b/etcd/src/etcd/settings.cpp @@ -0,0 +1,35 @@ +#include + +#include +#include + +USERVER_NAMESPACE_BEGIN + +namespace etcd { + +namespace { + +constexpr std::uint32_t kDefaultAttempts{3}; +constexpr std::chrono::milliseconds kDefaultRequestTimeout{1'000}; +constexpr std::chrono::milliseconds kDefaultWatchTimeout{1'000'000}; + +} // namespace + +} // namespace etcd + +namespace formats::parse { + +etcd::ClientSettings Parse(const yaml_config::YamlConfig& config, To) { + etcd::ClientSettings client_settings; + client_settings.endpoints = config["endpoints"].As>(); + client_settings.attempts = config["attempts"].As(etcd::kDefaultAttempts); + client_settings.request_timeout_ms = + config["request_timeout_ms"].As(etcd::kDefaultRequestTimeout); + client_settings.watch_timeout_ms = + config["watch_timeout_ms"].As(etcd::kDefaultWatchTimeout); + return client_settings; +} + +} // namespace formats::parse + +USERVER_NAMESPACE_END diff --git a/etcd/src/etcd/watch_listener.cpp b/etcd/src/etcd/watch_listener.cpp new file mode 100644 index 000000000000..926d2e6d9dc1 --- /dev/null +++ b/etcd/src/etcd/watch_listener.cpp @@ -0,0 +1,23 @@ +#include + +#include +#include + +USERVER_NAMESPACE_BEGIN + +namespace etcd { + +WatchListener::WatchListener(concurrent::SpscQueue::Consumer&& consumer) + : consumer_(std::move(consumer)) {} + +KeyValueState WatchListener::GetEvent() { + KeyValueState event; + if (!consumer_.Pop(event)) { + throw EtcdError("Consumer pop failed while trying to get etcd key-value event"); + } + return event; +} + +} // namespace etcd + +USERVER_NAMESPACE_END