feat redis: add CLUSTER SHARDS option

Adding support for CLUSTER SHARDS to determine topology instead of CLUSTER SLOTS.
Unlike cluster slots, it includes unavailable hosts (marked with "health": "failed") in the list of shards. This is important for determining the source of the problem: whether we cannot connect to the host, or the host is unavailable and the cluster itself is aware of it.
commit_hash:2fccb9fbebd76ec9093937b77b4f89c1635ead20

Author: vaninvv

.mapping.json

@@ -4016,6 +4016,11 @@
   "redis/functional_tests/cluster_auto_topology_pubsub/static_config.yaml":"taxi/uservices/userver/redis/functional_tests/cluster_auto_topology_pubsub/static_config.yaml",
   "redis/functional_tests/cluster_auto_topology_pubsub/tests/conftest.py":"taxi/uservices/userver/redis/functional_tests/cluster_auto_topology_pubsub/tests/conftest.py",
   "redis/functional_tests/cluster_auto_topology_pubsub/tests/test_redis_pubsub.py":"taxi/uservices/userver/redis/functional_tests/cluster_auto_topology_pubsub/tests/test_redis_pubsub.py",
+  "redis/functional_tests/cluster_shards_auto_topology/CMakeLists.txt":"taxi/uservices/userver/redis/functional_tests/cluster_shards_auto_topology/CMakeLists.txt",
+  "redis/functional_tests/cluster_shards_auto_topology/redis_service.cpp":"taxi/uservices/userver/redis/functional_tests/cluster_shards_auto_topology/redis_service.cpp",
+  "redis/functional_tests/cluster_shards_auto_topology/static_config.yaml":"taxi/uservices/userver/redis/functional_tests/cluster_shards_auto_topology/static_config.yaml",
+  "redis/functional_tests/cluster_shards_auto_topology/tests/conftest.py":"taxi/uservices/userver/redis/functional_tests/cluster_shards_auto_topology/tests/conftest.py",
+  "redis/functional_tests/cluster_shards_auto_topology/tests/test_redis_cluster.py":"taxi/uservices/userver/redis/functional_tests/cluster_shards_auto_topology/tests/test_redis_cluster.py",
   "redis/functional_tests/database_index/CMakeLists.txt":"taxi/uservices/userver/redis/functional_tests/database_index/CMakeLists.txt",
   "redis/functional_tests/database_index/redis_service.cpp":"taxi/uservices/userver/redis/functional_tests/database_index/redis_service.cpp",
   "redis/functional_tests/database_index/static_config.yaml":"taxi/uservices/userver/redis/functional_tests/database_index/static_config.yaml",
@@ -4078,6 +4083,7 @@
   "redis/include/userver/storages/redis/sharding_strategies.hpp":"taxi/uservices/userver/redis/include/userver/storages/redis/sharding_strategies.hpp",
   "redis/include/userver/storages/redis/subscribe_client.hpp":"taxi/uservices/userver/redis/include/userver/storages/redis/subscribe_client.hpp",
   "redis/include/userver/storages/redis/subscription_token.hpp":"taxi/uservices/userver/redis/include/userver/storages/redis/subscription_token.hpp",
+  "redis/include/userver/storages/redis/topology_update_method.hpp":"taxi/uservices/userver/redis/include/userver/storages/redis/topology_update_method.hpp",
   "redis/include/userver/storages/redis/transaction.hpp":"taxi/uservices/userver/redis/include/userver/storages/redis/transaction.hpp",
   "redis/include/userver/storages/redis/ttl_reply.hpp":"taxi/uservices/userver/redis/include/userver/storages/redis/ttl_reply.hpp",
   "redis/include/userver/storages/redis/utest/redis_fixture.hpp":"taxi/uservices/userver/redis/include/userver/storages/redis/utest/redis_fixture.hpp",
@@ -4107,6 +4113,12 @@
   "redis/src/storages/redis/impl/cluster_shard.cpp":"taxi/uservices/userver/redis/src/storages/redis/impl/cluster_shard.cpp",
   "redis/src/storages/redis/impl/cluster_shard.hpp":"taxi/uservices/userver/redis/src/storages/redis/impl/cluster_shard.hpp",
   "redis/src/storages/redis/impl/cluster_shard_test.cpp":"taxi/uservices/userver/redis/src/storages/redis/impl/cluster_shard_test.cpp",
+  "redis/src/storages/redis/impl/cluster_shards_query.cpp":"taxi/uservices/userver/redis/src/storages/redis/impl/cluster_shards_query.cpp",
+  "redis/src/storages/redis/impl/cluster_shards_query.hpp":"taxi/uservices/userver/redis/src/storages/redis/impl/cluster_shards_query.hpp",
+  "redis/src/storages/redis/impl/cluster_shards_query_test.cpp":"taxi/uservices/userver/redis/src/storages/redis/impl/cluster_shards_query_test.cpp",
+  "redis/src/storages/redis/impl/cluster_slots_query.cpp":"taxi/uservices/userver/redis/src/storages/redis/impl/cluster_slots_query.cpp",
+  "redis/src/storages/redis/impl/cluster_slots_query.hpp":"taxi/uservices/userver/redis/src/storages/redis/impl/cluster_slots_query.hpp",
+  "redis/src/storages/redis/impl/cluster_slots_query_test.cpp":"taxi/uservices/userver/redis/src/storages/redis/impl/cluster_slots_query_test.cpp",
   "redis/src/storages/redis/impl/cluster_subscription_storage.cpp":"taxi/uservices/userver/redis/src/storages/redis/impl/cluster_subscription_storage.cpp",
   "redis/src/storages/redis/impl/cluster_subscription_storage.hpp":"taxi/uservices/userver/redis/src/storages/redis/impl/cluster_subscription_storage.hpp",
   "redis/src/storages/redis/impl/cluster_subscription_storage_test.cpp":"taxi/uservices/userver/redis/src/storages/redis/impl/cluster_subscription_storage_test.cpp",
@@ -4212,6 +4224,7 @@
   "redis/src/storages/redis/test/mock_client_google_test.cpp":"taxi/uservices/userver/redis/src/storages/redis/test/mock_client_google_test.cpp",
   "redis/src/storages/redis/test/mock_publish_waiter_test.cpp":"taxi/uservices/userver/redis/src/storages/redis/test/mock_publish_waiter_test.cpp",
   "redis/src/storages/redis/test/subscribe_client_mock_test.cpp":"taxi/uservices/userver/redis/src/storages/redis/test/subscribe_client_mock_test.cpp",
+  "redis/src/storages/redis/topology_update_method.cpp":"taxi/uservices/userver/redis/src/storages/redis/topology_update_method.cpp",
   "redis/src/storages/redis/transaction_impl.cpp":"taxi/uservices/userver/redis/src/storages/redis/transaction_impl.cpp",
   "redis/src/storages/redis/transaction_impl.hpp":"taxi/uservices/userver/redis/src/storages/redis/transaction_impl.hpp",
   "redis/src/storages/redis/transaction_redistest.cpp":"taxi/uservices/userver/redis/src/storages/redis/transaction_redistest.cpp",

redis/functional_tests/cluster_shards_auto_topology/CMakeLists.txt

@@ -0,0 +1,6 @@
+project(userver-redis-tests-cluster-auto-topology CXX)
+
+add_executable(${PROJECT_NAME} "redis_service.cpp")
+target_link_libraries(${PROJECT_NAME} userver::redis)
+
+userver_chaos_testsuite_add_redis()

redis/functional_tests/cluster_shards_auto_topology/redis_service.cpp

@@ -0,0 +1,140 @@
+#include <userver/testsuite/testsuite_support.hpp>
+#include <userver/utest/using_namespace_userver.hpp>  // IWYU pragma: keep
+
+#include <iostream>
+#include <string>
+#include <string_view>
+
+#include <fmt/format.h>
+
+#include <userver/clients/dns/component.hpp>
+#include <userver/clients/http/component_list.hpp>
+#include <userver/components/component.hpp>
+#include <userver/components/minimal_server_component_list.hpp>
+#include <userver/dynamic_config/updater/component_list.hpp>
+#include <userver/server/handlers/http_handler_base.hpp>
+#include <userver/server/handlers/tests_control.hpp>
+#include <userver/storages/redis/client.hpp>
+#include <userver/storages/redis/component.hpp>
+#include <userver/storages/secdist/component.hpp>
+#include <userver/storages/secdist/provider_component.hpp>
+#include <userver/utils/daemon_run.hpp>
+#include <userver/yaml_config/merge_schemas.hpp>
+#include "userver/storages/redis/exception.hpp"
+
+namespace chaos {
+
+class KeyValue final : public server::handlers::HttpHandlerBase {
+public:
+    static constexpr std::string_view kName = "handler-redis";
+
+    KeyValue(const components::ComponentConfig& config, const components::ComponentContext& context);
+
+    std::string HandleRequestThrow(const server::http::HttpRequest& request, server::request::RequestContext&)
+        const override;
+
+    static yaml_config::Schema GetStaticConfigSchema();
+
+private:
+    std::string GetValue(std::string_view key, const server::http::HttpRequest& request) const;
+    std::string PostValue(std::string_view key, const server::http::HttpRequest& request) const;
+    std::string DeleteValue(std::string_view key) const;
+
+    storages::redis::ClientPtr redis_client_;
+    storages::redis::CommandControl redis_cc_;
+};
+
+KeyValue::KeyValue(const components::ComponentConfig& config, const components::ComponentContext& context)
+    : server::handlers::HttpHandlerBase(config, context),
+      redis_client_{
+          context.FindComponent<components::Redis>("key-value-database").GetClient(config["db"].As<std::string>())
+      }
+{}
+
+std::string KeyValue::HandleRequestThrow(
+    const server::http::HttpRequest& request,
+    server::request::RequestContext& /*context*/
+) const {
+    const auto& key = request.GetArg("key");
+    if (key.empty()) {
+        throw server::handlers::ClientError(server::handlers::ExternalBody{"No 'key' query argument"});
+    }
+
+    switch (request.GetMethod()) {
+        case server::http::HttpMethod::kGet:
+            return GetValue(key, request);
+        case server::http::HttpMethod::kPost:
+            return PostValue(key, request);
+        case server::http::HttpMethod::kDelete:
+            return DeleteValue(key);
+        default:
+            throw server::handlers::ClientError(server::handlers::ExternalBody{
+                fmt::format("Unsupported method {}", request.GetMethod())
+            });
+    }
+}
+
+yaml_config::Schema KeyValue::GetStaticConfigSchema() {
+    return yaml_config::MergeSchemas<HandlerBase>(R"(
+type: object
+description: KeyValue handler schema
+additionalProperties: false
+properties:
+    db:
+        type: string
+        description: redis database name
+)");
+}
+
+std::string KeyValue::GetValue(std::string_view key, const server::http::HttpRequest& request) const {
+    try {
+        const auto result = redis_client_->Get(std::string{key}, redis_cc_).Get();
+        if (!result) {
+            request.SetResponseStatus(server::http::HttpStatus::kNotFound);
+            return {};
+        }
+        return *result;
+    } catch (const storages::redis::RequestFailedException& e) {
+        LOG_WARNING() << "EXCEPTION GET: " << e.what() << ", " << e.GetStatusString() << "\n";
+        throw;
+    }
+}
+
+std::string KeyValue::PostValue(std::string_view key, const server::http::HttpRequest& request) const {
+    try {
+        const auto& value = request.GetArg("value");
+        const auto result = redis_client_->SetIfNotExist(std::string{key}, value, redis_cc_).Get();
+        if (!result) {
+            request.SetResponseStatus(server::http::HttpStatus::kConflict);
+            return {};
+        }
+
+        request.SetResponseStatus(server::http::HttpStatus::kCreated);
+        return std::string{value};
+    } catch (const storages::redis::RequestFailedException& e) {
+        LOG_WARNING() << "EXCEPTION POST: " << e.what() << ", " << e.GetStatusString() << "\n";
+        throw;
+    }
+}
+
+std::string KeyValue::DeleteValue(std::string_view key) const {
+    const auto result = redis_client_->Del(std::string{key}, redis_cc_).Get();
+    return std::to_string(result);
+}
+
+}  // namespace chaos
+
+int main(int argc, char* argv[]) {
+    const auto component_list =
+        components::MinimalServerComponentList()
+            .AppendComponentList(USERVER_NAMESPACE::dynamic_config::updater::ComponentList())
+            .Append<chaos::KeyValue>("handler-cluster")
+            .AppendComponentList(clients::http::ComponentList())
+            .Append<components::Secdist>()
+            .Append<components::DefaultSecdistProvider>()
+            .Append<components::Redis>("key-value-database")
+            .Append<components::TestsuiteSupport>()
+            .Append<server::handlers::TestsControl>()
+            .Append<clients::dns::Component>();
+    return utils::DaemonMain(argc, argv, component_list);
+}

redis/functional_tests/cluster_shards_auto_topology/static_config.yaml

@@ -0,0 +1,79 @@
+# yaml
+components_manager:
+    components:
+        handler-cluster:
+            db: redis-cluster
+            path: /redis-cluster
+            task_processor: main-task-processor
+            method: GET,DELETE,POST
+
+        key-value-database:
+            groups:
+              - config_name: redis-cluster
+                db: redis-cluster
+                sharding_strategy: RedisCluster
+                topology_update_method: cluster_shards
+              - config_name: redis-cluster2
+                db: redis-cluster2
+                sharding_strategy: RedisCluster
+                topology_update_method: cluster_shards
+            subscribe_groups: []
+            thread_pools:
+                redis_thread_pool_size: 2
+                sentinel_thread_pool_size: 1
+
+        testsuite-support:
+
+        http-client:
+        http-client-core:
+            fs-task-processor: main-task-processor
+
+        tests-control:
+            method: POST
+            path: /tests/{action}
+            skip-unregistered-testpoints: true
+            task_processor: main-task-processor
+            testpoint-timeout: 10s
+            testpoint-url: $mockserver/testpoint
+            throttling_enabled: false
+
+        secdist: {}
+        default-secdist-provider:
+            config: /etc/redis_service/secure_data.json
+            missing-ok: true
+            environment-secrets-key: SECDIST_CONFIG
+
+        server:
+            listener:
+                port: 8187
+                task_processor: main-task-processor
+        logging:
+            fs-task-processor: fs-task-processor
+            loggers:
+                default:
+                    file_path: '@stderr'
+                    level: debug
+                    overflow_behavior: discard
+
+        dynamic-config:
+            updates-enabled: true
+        dynamic-config-client:
+            config-url: $config-server-url
+            http-retries: 5
+            http-timeout: 20s
+            service-name: testsuite-support
+        dynamic-config-client-updater:
+            config-settings: false
+            first-update-fail-ok: true
+            full-update-interval: 1m
+            update-interval: 5s
+        dns-client:
+            fs-task-processor: fs-task-processor
+
+    task_processors:
+        main-task-processor:
+            worker_threads: 4
+        fs-task-processor:
+            worker_threads: 4
+
+    default_task_processor: main-task-processor

redis/functional_tests/cluster_shards_auto_topology/tests/conftest.py

@@ -0,0 +1,36 @@
+import json
+
+import pytest
+
+pytest_plugins = [
+    'pytest_userver.plugins.redis',
+    'pytest_userver.plugins.dynamic_config',
+    'taxi.uservices.userver.redis.functional_tests.pytest_redis_cluster_topology_plugin.pytest_plugin',
+]
+
+
+@pytest.fixture(scope='session')
+def service_env(redis_cluster_ports, redis_cluster_topology_session):
+    cluster_hosts = []
+    cluster_shards = []
+    for index, port in enumerate(redis_cluster_ports):
+        cluster_hosts.append({'host': '127.0.0.1', 'port': port})
+    for index in range(3):
+        cluster_shards.append({'name': f'shard{index}'})
+
+    secdist_config = {
+        'redis_settings': {
+            'redis-cluster': {
+                'password': '',
+                'sentinels': cluster_hosts,
+                'shards': cluster_shards,
+            },
+            'redis-cluster2': {
+                'password': '',
+                'sentinels': cluster_hosts,
+                'shards': cluster_shards,
+            },
+        },
+    }
+
+    return {'SECDIST_CONFIG': json.dumps(secdist_config)}

redis/functional_tests/cluster_shards_auto_topology/tests/test_redis_cluster.py

@@ -0,0 +1,91 @@
+import asyncio
+
+import pytest
+import pytest_userver.utils.sync as sync
+
+KEYS_SEQ_LEN = 20  # enough sequential keys to test all slots
+REDIS_PORT = 6379
+FAILOVER_DEADLINE_SEC = 30  # maximum time allowed to finish failover
+
+
+async def test_happy_path(service_client, redis_cluster_topology):
+    post_reqs = [
+        service_client.post(
+            '/redis-cluster',
+            params={'key': f'key{i}', 'value': 'abc'},
+        )
+        for i in range(KEYS_SEQ_LEN)
+    ]
+    assert all(res.status == 201 for res in await asyncio.gather(*post_reqs))
+
+    get_reqs = [service_client.get('/redis-cluster', params={'key': f'key{i}'}) for i in range(KEYS_SEQ_LEN)]
+    assert all(res.status == 200 and res.text == 'abc' for res in await asyncio.gather(*get_reqs))
+
+
+async def _check_write_all_slots(service_client, key_prefix, value):
+    post_reqs = [
+        service_client.post(
+            '/redis-cluster',
+            params={'key': f'{key_prefix}{i}', 'value': value},
+        )
+        for i in range(KEYS_SEQ_LEN)
+    ]
+    return all(res.status != 500 for res in await asyncio.gather(*post_reqs))
+
+
+async def _assert_read_all_slots(service_client, key_prefix, value):
+    get_reqs = [
+        service_client.get(
+            '/redis-cluster',
+            params={'key': f'{key_prefix}{i}'},
+        )
+        for i in range(KEYS_SEQ_LEN)
+    ]
+    assert all(res.status == 200 and res.text == value for res in await asyncio.gather(*get_reqs))
+
+
+async def test_hard_failover(service_client, redis_cluster_topology):
+    # Write enough different keys to have something in every slot
+    assert await _check_write_all_slots(service_client, 'hf_key1', 'abc')
+
+    # Start the failover
+    redis_cluster_topology.get_masters()[0].stop()
+
+    # wait until service detect that shard 0 is broken
+    # Failover starts in ~10 seconds
+    async def is_ready():
+        return await _check_write_all_slots(
+            service_client,
+            'hf_key2',
+            'cde',
+        )
+
+    await sync.wait(is_ready)
+
+    # Now that one of the replicas has become the master,
+    # check reading from the remaining replica
+    await _assert_read_all_slots(service_client, 'hf_key1', 'abc')
+    await _assert_read_all_slots(service_client, 'hf_key2', 'cde')
+
+
+@pytest.mark.skip(reason='Flaky TAXICOMMON-11677')
+async def test_add_shard(service_client, redis_cluster_topology):
+    # Write enough different keys to have something in every slot
+    assert await _check_write_all_slots(service_client, 'hf_key1', 'abc')
+
+    await redis_cluster_topology.add_shard()
+
+    # Failover starts in ~10 seconds
+    async def is_ready():
+        return await _check_write_all_slots(
+            service_client,
+            'hf_key2',
+            'cde',
+        )
+
+    await sync.wait(is_ready)
+
+    # Now that one of the replicas has become the master,
+    # check reading from the remaining replica
+    await _assert_read_all_slots(service_client, 'hf_key1', 'abc')
+    await _assert_read_all_slots(service_client, 'hf_key2', 'cde')