feat redis: support username authorization

Author: rsaddatimov

.gitignore

@@ -3,6 +3,7 @@
 .DS_Store
 .mypy_cache
 .pytest_cache
+.python-version
 *.orig
 /Makefile.local
 *.yawiki
@@ -12,6 +13,7 @@
 /build*/
 /cmake-build-*/
 CMakeLists.txt.user
+CMakeUserPresets.json
 compile_commands.json
 tags
 static-analyzer-report

redis/include/userver/storages/redis/base.hpp

@@ -14,13 +14,15 @@ USERVER_NAMESPACE_BEGIN
 
 namespace storages::redis {
 
+using Username = utils::StrongTypedef<class UsernameTag, std::string>;
 using Password = utils::NonLoggable<class PasswordTag, std::string>;
 
 enum class ConnectionSecurity { kNone, kTLS };
 
 struct ConnectionInfo {
     std::string host = "localhost";
     int port = 26379;
+    Username username;
     Password password;
     bool read_only = false;
     ConnectionSecurity connection_security = ConnectionSecurity::kNone;
@@ -31,18 +33,19 @@ struct ConnectionInfo {
     ConnectionInfo(
         std::string host,
         int port,
+        Username username,
         Password password,
         bool read_only = false,
         ConnectionSecurity security = ConnectionSecurity::kNone,
         std::size_t db_index = 0
     )
         : host{std::move(host)},
           port{port},
+          username{std::move(username)},
           password{std::move(password)},
           read_only{read_only},
           connection_security(security),
-          database_index(db_index)
-    {}
+          database_index(db_index) {}
 };
 
 struct Stat {
@@ -81,9 +84,7 @@ struct MetricsSettings {
 
     DynamicSettings dynamic_settings;
 
-    MetricsSettings(const DynamicSettings& dynamic_settings)
-        : dynamic_settings(dynamic_settings)
-    {}
+    MetricsSettings(const DynamicSettings& dynamic_settings) : dynamic_settings(dynamic_settings) {}
     MetricsSettings() = default;
     MetricsSettings(const MetricsSettings&) = default;
     MetricsSettings(MetricsSettings&&) = default;

redis/src/storages/redis/component.cpp

@@ -115,9 +115,7 @@ RedisPools Parse(const yaml_config::YamlConfig& value, formats::parse::To<RedisP
 }
 
 Redis::Redis(const ComponentConfig& config, const ComponentContext& component_context)
-    : ComponentBase(config, component_context),
-      config_(component_context.FindComponent<DynamicConfig>().GetSource())
-{
+    : ComponentBase(config, component_context), config_(component_context.FindComponent<DynamicConfig>().GetSource()) {
     const auto&
         testsuite_redis_control = component_context.FindComponent<components::TestsuiteSupport>().GetRedisControl();
     Connect(config, component_context, testsuite_redis_control);
@@ -145,11 +143,13 @@ std::shared_ptr<storages::redis::Client> Redis::GetClient(
 ) const {
     auto it = clients_.find(name);
     if (it == clients_.end()) {
-        throw std::runtime_error(fmt::format(
-            "{} redis client not found. Available clients: [{}]",
-            name,
-            fmt::join(clients_ | boost::adaptors::map_keys, ", ")
-        ));
+        throw std::runtime_error(
+            fmt::format(
+                "{} redis client not found. Available clients: [{}]",
+                name,
+                fmt::join(clients_ | boost::adaptors::map_keys, ", ")
+            )
+        );
     }
     it->second->WaitConnectedOnce(wait_connected);
     return it->second;
@@ -158,11 +158,13 @@ std::shared_ptr<storages::redis::Client> Redis::GetClient(
 std::shared_ptr<storages::redis::impl::Sentinel> Redis::Client(const std::string& name) const {
     auto it = sentinels_.find(name);
     if (it == sentinels_.end()) {
-        throw std::runtime_error(fmt::format(
-            "{} redis client not found. Available clients: [{}]",
-            name,
-            fmt::join(clients_ | boost::adaptors::map_keys, ", ")
-        ));
+        throw std::runtime_error(
+            fmt::format(
+                "{} redis client not found. Available clients: [{}]",
+                name,
+                fmt::join(clients_ | boost::adaptors::map_keys, ", ")
+            )
+        );
     }
     return it->second;
 }
@@ -173,12 +175,14 @@ std::shared_ptr<storages::redis::SubscribeClient> Redis::GetSubscribeClient(
 ) const {
     auto it = subscribe_clients_.find(name);
     if (it == subscribe_clients_.end()) {
-        throw std::runtime_error(fmt::format(
-            "{} redis subscribe-client not found. Available subscribe-clients: "
-            "[{}]",
-            name,
-            fmt::join(subscribe_clients_ | boost::adaptors::map_keys, ", ")
-        ));
+        throw std::runtime_error(
+            fmt::format(
+                "{} redis subscribe-client not found. Available subscribe-clients: "
+                "[{}]",
+                name,
+                fmt::join(subscribe_clients_ | boost::adaptors::map_keys, ", ")
+            )
+        );
     }
     it->second->WaitConnectedOnce(wait_connected);
     return std::static_pointer_cast<storages::redis::SubscribeClient>(it->second);
@@ -331,11 +335,13 @@ void Redis::OnSecdistUpdate(const storages::secdist::SecdistConfig& cfg) {
 
         std::vector<storages::redis::ConnectionInfo> cii;
         for (const auto& host_port : settings.sentinels) {
-            const storages::redis::ConnectionInfo ci(host_port.host, host_port.port, settings.password);
+            const storages::redis::ConnectionInfo
+                ci(host_port.host, host_port.port, settings.username, settings.password);
             cii.push_back(ci);
         }
 
         sentinel->SetConnectionInfo(cii);
+        sentinel->UpdateUsername(settings.username);
         sentinel->UpdatePassword(settings.password);
     }
 }

redis/src/storages/redis/impl/cluster_topology_holder.cpp

@@ -128,6 +128,7 @@ ClusterTopologyHolder::ClusterTopologyHolder(
     const engine::ev::ThreadControl& sentinel_thread_control,
     const std::shared_ptr<engine::ev::ThreadPool>& redis_thread_pool,
     std::string shard_group_name,
+    Username username,
     Password password,
     const std::vector<std::string>& /*shards*/,
     const std::vector<ConnectionInfo>& conns,
@@ -136,11 +137,16 @@ ClusterTopologyHolder::ClusterTopologyHolder(
     : ev_thread_(sentinel_thread_control),
       redis_thread_pool_(redis_thread_pool),
       shard_group_name_(std::move(shard_group_name)),
+      username_(std::move(username)),
       password_(std::move(password)),
       shards_names_(MakeShardNames()),
       conns_(conns),
       statistics_holder_(),
-      update_topology_timer_(ev_thread_, [this] { UpdateClusterTopology(); }, kSentinelGetHostsCheckInterval),
+      update_topology_timer_(
+          ev_thread_,
+          [this] { UpdateClusterTopology(); },
+          kSentinelGetHostsCheckInterval
+      ),
       update_topology_watch_(
           ev_thread_,
           [this] {
@@ -155,15 +161,23 @@ ClusterTopologyHolder::ClusterTopologyHolder(
               explore_nodes_watch_.Start();
           }
       ),
-      explore_nodes_timer_(ev_thread_, [this] { ExploreNodes(); }, kSentinelGetHostsCheckInterval),
+      explore_nodes_timer_(
+          ev_thread_,
+          [this] { ExploreNodes(); },
+          kSentinelGetHostsCheckInterval
+      ),
       create_nodes_watch_(
           ev_thread_,
           [this] {
               CreateNodes();
               create_nodes_watch_.Start();
           }
       ),
-      delete_expired_nodes_timer_(ev_thread_, [this] { DeleteNodes(); }, kDeleteNodesCheckInterval),
+      delete_expired_nodes_timer_(
+          ev_thread_,
+          [this] { DeleteNodes(); },
+          kDeleteNodesCheckInterval
+      ),
       sentinels_process_creation_timer_(
           ev_thread_,
           [this] {
@@ -194,8 +208,7 @@ ClusterTopologyHolder::ClusterTopologyHolder(
       ),
       is_topology_received_(false),
       update_cluster_slots_flag_(false),
-      connection_security_(connection_security)
-{
+      connection_security_(connection_security) {
     log_extra_.Extend("shard_group_name", shard_group_name_);
     LOG_DEBUG() << log_extra_ << "Created ClusterTopologyHolder";
 }
@@ -339,11 +352,21 @@ boost::signals2::signal<void(size_t)>& ClusterTopologyHolder::GetSignalTopologyC
     return signal_topology_changed_;
 }
 
+void ClusterTopologyHolder::UpdateUsername(const Username& username) {
+    auto lock = username_.UniqueLock();
+    *lock = username;
+}
+
 void ClusterTopologyHolder::UpdatePassword(const Password& password) {
     auto lock = password_.UniqueLock();
     *lock = password;
 }
 
+Username ClusterTopologyHolder::GetUsername() {
+    const auto lock = username_.Lock();
+    return *lock;
+}
+
 Password ClusterTopologyHolder::GetPassword() {
     const auto lock = password_.Lock();
     return *lock;
@@ -471,6 +494,7 @@ std::shared_ptr<RedisConnectionHolder> ClusterTopologyHolder::CreateRedisInstanc
         shard_group_name_,
         host,
         port,
+        GetUsername(),
         GetPassword(),
         kClusterDatabaseIndex,
         buffering_settings_ptr->value_or(CommandsBufferingSettings{}),
@@ -497,11 +521,14 @@ void ClusterTopologyHolder::UpdateClusterTopology() {
     /// ...
     ProcessGetClusterHostsRequest(
         shards_names_,
-        GetClusterHostsRequest(*sentinels_, GetPassword(), shard_group_name_),
+        GetClusterHostsRequest(*sentinels_, GetUsername(), GetPassword(), shard_group_name_),
         [this,
-         reset{std::move(reset_update_cluster_slots)
-         }](ClusterShardHostInfos shard_infos, size_t requests_sent, size_t responses_parsed, bool is_non_cluster_error
-        ) {
+         reset{
+             std::move(reset_update_cluster_slots)
+         }](ClusterShardHostInfos shard_infos,
+            size_t requests_sent,
+            size_t responses_parsed,
+            bool is_non_cluster_error) {
             LOG_DEBUG()
                 << log_extra_ << "Parsing response from cluster slots: shard_infos.size(): " << shard_infos.size()
                 << ", requests_sent=" << requests_sent << ", responses_parsed=" << responses_parsed;

redis/src/storages/redis/impl/cluster_topology_holder.hpp

@@ -25,6 +25,7 @@ class ClusterTopologyHolder final : public TopologyHolderBase {
         const engine::ev::ThreadControl& sentinel_thread_control,
         const std::shared_ptr<engine::ev::ThreadPool>& redis_thread_pool,
         std::string shard_group_name,
+        Username username,
         Password password,
         const std::vector<std::string>& /*shards*/,
         const std::vector<ConnectionInfo>& conns,
@@ -47,8 +48,13 @@ class ClusterTopologyHolder final : public TopologyHolderBase {
     void SetConnectionInfo(const std::vector<ConnectionInfoInt>& info_array) override;
     boost::signals2::signal<void(HostPort, Redis::State)>& GetSignalNodeStateChanged() override;
     boost::signals2::signal<void(size_t)>& GetSignalTopologyChanged() override;
+
+    void UpdateUsername(const Username& username) override;
     void UpdatePassword(const Password& password) override;
+
+    Username GetUsername() override;
     Password GetPassword() override;
+
     std::string GetReadinessInfo() const override;
 
     static size_t GetClusterSlotsCalledCounter() { return cluster_slots_call_counter.load(std::memory_order_relaxed); }
@@ -61,6 +67,7 @@ class ClusterTopologyHolder final : public TopologyHolderBase {
 
     const std::string shard_group_name_;
     logging::LogExtra log_extra_;
+    concurrent::Variable<Username, std::mutex> username_;
     concurrent::Variable<Password, std::mutex> password_;
     std::shared_ptr<const std::vector<std::string>> shards_names_;
     const std::vector<ConnectionInfo> conns_;