feat etcd: Init etcd client

CMakeLists.txt

@@ -143,6 +143,7 @@ option(USERVER_FEATURE_MYSQL "Provide asynchronous driver for MariaDB/MySQL" "${
 option(USERVER_FEATURE_ROCKS "Provide asynchronous driver for Rocks" "${USERVER_LIB_ENABLED_DEFAULT}")
 option(USERVER_FEATURE_YDB "Provide asynchronous driver for YDB" "${USERVER_YDB_DEFAULT}")
 option(USERVER_FEATURE_OTLP "Provide asynchronous OTLP exporters" "${USERVER_LIB_ENABLED_DEFAULT}")
+option(USERVER_FEATURE_ETCD "Provide asynchronous driver for etcd" "${USERVER_LIB_ENABLED_DEFAULT}")
 
 set(CMAKE_DEBUG_POSTFIX d)
 
@@ -277,6 +278,11 @@ if (USERVER_FEATURE_YDB)
   list(APPEND USERVER_AVAILABLE_COMPONENTS ydb)
 endif()
 
+if (USERVER_FEATURE_ETCD)
+  _require_userver_core("USERVER_FEATURE_ETCD")
+  add_subdirectory(etcd)
+endif()
+
 add_subdirectory(libraries)
 
 if (USERVER_BUILD_TESTS)

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)

etcd/CMakeLists.txt

@@ -0,0 +1,8 @@
+project(userver-etcd CXX)
+
+# TODO: Add etcd setup
+
+userver_module(etcd
+    SOURCE_DIR "${CMAKE_CURRENT_SOURCE_DIR}"
+    UTEST_SOURCES "${CMAKE_CURRENT_SOURCE_DIR}/src/*_test.cpp"
+)

etcd/include/userver/storages/etcd/client.hpp

@@ -0,0 +1,36 @@
+#pragma once
+
+#include <chrono>
+#include <cstdint>
+#include <memory>
+#include <vector>
+
+#include <userver/clients/http/component.hpp>
+#include <userver/engine/shared_mutex.hpp>
+#include <userver/storages/etcd/settings.hpp>
+#include <userver/yaml_config/fwd.hpp>
+
+USERVER_NAMESPACE_BEGIN
+
+namespace storages::etcd {
+
+class Client final {
+public:
+    Client(clients::http::Client& http_client, ClientSettings settings);
+    void Put(const std::string& key, const std::string& value);
+    [[nodiscard]] std::vector<std::string> Range(const std::string& key);
+    void DeleteRange(const std::string& key);
+private:
+    [[nodiscard]] std::shared_ptr<clients::http::Response>
+     PerformEtcdRequest(const std::function<std::string(const std::string&)>& url_builder, const std::string& data);
+
+    clients::http::Client& http_client_;
+    engine::SharedMutex endpoints_shared_mutex_;
+    ClientSettings settings_;
+};
+
+using ClientPtr = std::shared_ptr<Client>;
+
+}
+
+USERVER_NAMESPACE_END

etcd/include/userver/storages/etcd/component.hpp

@@ -0,0 +1,30 @@
+#pragma once
+
+#include <userver/components/component_base.hpp>
+#include <userver/components/component_config.hpp>
+#include <userver/components/component_context.hpp>
+#include <userver/storages/etcd/client.hpp>
+
+USERVER_NAMESPACE_BEGIN
+
+namespace storages::etcd {
+
+class Component final : public components::ComponentBase {
+public:
+    static constexpr std::string_view kName = "etcd";
+
+    Component(const components::ComponentConfig&, const components::ComponentContext&);
+
+    ~Component() = default;
+
+    static yaml_config::Schema GetStaticConfigSchema();
+
+    ClientPtr GetClient();
+
+private:
+    const ClientPtr etcd_client_ptr_;
+};
+
+}
+
+USERVER_NAMESPACE_END

etcd/include/userver/storages/etcd/settings.hpp

@@ -0,0 +1,27 @@
+#pragma once
+
+#include <chrono>
+#include <string>
+#include <vector>
+
+#include <userver/yaml_config/yaml_config.hpp>
+
+USERVER_NAMESPACE_BEGIN
+
+namespace storages::etcd {
+
+struct ClientSettings final {
+    std::vector<std::string> endpoints;
+    std::uint32_t retries;
+    std::chrono::microseconds request_timeout_ms;
+};
+
+}
+
+namespace formats::parse {
+
+storages::etcd::ClientSettings Parse(const yaml_config::YamlConfig& value, To<storages::etcd::ClientSettings>);
+
+}
+
+USERVER_NAMESPACE_END

etcd/library.yaml

@@ -0,0 +1,9 @@
+project-name: userver-etcd
+project-alt-names:
+  - yandex-userver-etcd
+maintainers:
+  - Common components
+description: Etcd driver
+
+libraries:
+  - userver-core

etcd/src/storages/etcd/client.cpp

@@ -0,0 +1,125 @@
+#include <userver/storages/etcd/client.hpp>
+
+#include <iostream>
+#include <string>
+
+#include <fmt/format.h>
+
+#include <userver/crypto/base64.hpp>
+#include <userver/dynamic_config/value.hpp>
+#include <userver/formats/json/string_builder.hpp>
+#include <userver/formats/parse/common_containers.hpp>
+#include <userver/http/common_headers.hpp>
+#include <userver/logging/log.hpp>
+#include <userver/utils/rand.hpp>
+#include <userver/yaml_config/yaml_config.hpp>
+
+USERVER_NAMESPACE_BEGIN
+
+namespace storages::etcd {
+
+namespace {
+
+std::string BuildPutUrl(const std::string& service_url) {
+    return fmt::format("{}/v3/kv/put", service_url);
+}
+
+std::string BuildPutData(const std::string& key, const std::string& value) {
+    formats::json::StringBuilder sb;
+    {
+        formats::json::StringBuilder::ObjectGuard guard{sb};
+        sb.Key("key");
+        sb.WriteString(crypto::base64::Base64Encode(key));
+        sb.Key("value");
+        sb.WriteString(crypto::base64::Base64Encode(value));
+    }
+    return sb.GetString();
+}
+
+std::string BuildRangeUrl(const std::string& service_url) {
+    return fmt::format("{}/v3/kv/range", service_url);
+}
+
+std::string BuildRangeData(const std::string& key) {
+    formats::json::StringBuilder sb;
+    {
+        formats::json::StringBuilder::ObjectGuard guard{sb};
+        sb.Key("key");
+        sb.WriteString(crypto::base64::Base64Encode(key));
+    }
+    return sb.GetString();
+}
+
+std::string BuildDeleteRangeUrl(const std::string& service_url) {
+    return fmt::format("{}/v3/kv/deleterange", service_url);
+}
+
+std::string BuildDeleteRangeData(const std::string& key) {
+    formats::json::StringBuilder sb;
+    {
+        formats::json::StringBuilder::ObjectGuard guard{sb};
+        sb.Key("key");
+        sb.WriteString(crypto::base64::Base64Encode(key));
+    }
+    return sb.GetString();
+}
+
+bool ShouldRetry(std::shared_ptr<clients::http::Response> response) {
+    return false;
+}
+
+}
+
+Client::Client(clients::http::Client& http_client, ClientSettings settings)
+    : http_client_(http_client),
+    settings_(settings) {
+}
+
+void Client::Put(const std::string& key, const std::string& value) {
+    auto response = PerformEtcdRequest(BuildPutUrl, BuildPutData(key, value));
+}
+
+std::vector<std::string> Client::Range(const std::string& key) {
+    auto response = PerformEtcdRequest(BuildRangeUrl, BuildRangeData(key));
+
+    const auto json_body = formats::json::FromString(response->body());
+    const auto& key_value_list = json_body["kvs"];
+    std::vector<std::string> values;
+    values.reserve(key_value_list.GetSize());
+    for (const auto& key_value : key_value_list) {
+        values.push_back(
+            crypto::base64::Base64Decode(key_value["value"].As<std::string>())
+        );
+    }
+    return values;
+}
+
+void Client::DeleteRange(const std::string& key) {
+    auto response = PerformEtcdRequest(BuildDeleteRangeUrl, BuildDeleteRangeData(key));
+}
+
+std::shared_ptr<clients::http::Response> Client::PerformEtcdRequest(
+    const std::function<std::string(const std::string&)>& url_builder, const std::string& data
+) {
+    endpoints_shared_mutex_.lock_shared();
+    auto endpoints = settings_.endpoints;
+    endpoints_shared_mutex_.unlock_shared();
+    utils::Shuffle(endpoints);
+
+    for (const auto& endpoint : endpoints) {
+        auto response = http_client_
+            .CreateRequest()
+            .post(url_builder(endpoint), data)
+            .retry(settings_.retries)
+            .timeout(settings_.request_timeout_ms.count())
+            .perform();
+        LOG_DEBUG() << "Response: " << formats::json::FromString(response->body());
+        if (!ShouldRetry(response)) {
+            return response;
+        }
+    }
+}
+
+}  // namespace storages::etcd
+
+USERVER_NAMESPACE_END

etcd/src/storages/etcd/component.cpp

@@ -0,0 +1,50 @@
+#include <userver/storages/etcd/component.hpp>
+
+#include <userver/clients/http/component.hpp>
+#include <userver/storages/etcd/settings.hpp>
+#include <userver/yaml_config/merge_schemas.hpp>
+
+
+USERVER_NAMESPACE_BEGIN
+
+namespace storages::etcd {
+
+Component::Component(const components::ComponentConfig& config, const components::ComponentContext& context)
+    :
+    ComponentBase(config, context),
+    etcd_client_ptr_(std::make_shared<Client>(
+        context.FindComponent<components::HttpClient>().GetHttpClient(),
+        config.As<ClientSettings>()
+    )) {}
+
+yaml_config::Schema Component::GetStaticConfigSchema() {
+    return yaml_config::MergeSchemas<ComponentBase>(R"(
+type: object
+description: Etcd cluster component
+additionalProperties: false
+properties:
+    endpoints:
+        type: array
+        description: Etcd endpoints
+        items:
+            type: string
+            description: host
+    retries:
+        type: integer
+        description: >
+            Number of retries per one endpoints, total number of retries is number of endpoints times retries
+        minimum: 1
+    request_timeout_ms:
+        type: integer
+        description: Number of miliseconds to timeout request
+        minimum: 1
+)");
+}
+
+ClientPtr Component::GetClient() {
+    return etcd_client_ptr_;
+}
+
+}  // namespace storages::etcd
+
+USERVER_NAMESPACE_END

etcd/src/storages/etcd/settings.cpp

@@ -0,0 +1,32 @@
+#include <userver/storages/etcd/settings.hpp>
+
+#include <userver/dynamic_config/value.hpp>
+#include <userver/formats/parse/common_containers.hpp>
+
+USERVER_NAMESPACE_BEGIN
+
+
+namespace storages::etcd {
+
+namespace {
+
+constexpr std::uint32_t kDefaultRetries{3};
+constexpr std::chrono::milliseconds kDefaultRequestTimeout{1'000};
+
+}
+
+}
+
+namespace formats::parse {
+
+storages::etcd::ClientSettings Parse(const yaml_config::YamlConfig& cofig, To<storages::etcd::ClientSettings>) {
+    storages::etcd::ClientSettings result;
+    result.endpoints = cofig["endpoints"].As<std::vector<std::string>>(result.endpoints);
+    result.retries = cofig["retries"].As<std::uint32_t>(storages::etcd::kDefaultRetries);
+    result.request_timeout_ms = cofig["request_timeout_ms"].As<std::chrono::milliseconds>(storages::etcd::kDefaultRequestTimeout);
+    return result;
+}
+
+}  // namespace formats::parse
+
+USERVER_NAMESPACE_END

testsuite/pytest_plugins/pytest_userver/plugins/config.py

@@ -406,7 +406,7 @@ def allowed_url_prefixes_extra() -> typing.List[str]:
 
     @ingroup userver_testsuite_fixtures
     """
-    return []
+    return ["http://localhost:2379"]
 
 
 @pytest.fixture(scope='session')