feat redis: reload secdist

commit_hash:761c62562a9bf66c6f973e00a5594d239b1b02db

Author: segoon

redis/include/userver/storages/redis/component.hpp

@@ -14,6 +14,7 @@
 #include <userver/storages/redis/base.hpp>
 #include <userver/storages/redis/fwd.hpp>
 #include <userver/storages/redis/wait_connected_mode.hpp>
+#include <userver/storages/secdist/secdist.hpp>
 #include <userver/testsuite/redis_control.hpp>
 #include <userver/utils/statistics/entry.hpp>
 
@@ -145,6 +146,7 @@ class Redis : public ComponentBase {
 
 private:
     void OnConfigUpdate(const dynamic_config::Snapshot& cfg);
+    void OnSecdistUpdate(const storages::secdist::SecdistConfig& cfg);
 
     void Connect(
         const ComponentConfig& config,
@@ -162,6 +164,7 @@ class Redis : public ComponentBase {
 
     dynamic_config::Source config_;
     concurrent::AsyncEventSubscriberScope config_subscription_;
+    concurrent::AsyncEventSubscriberScope secdist_subscription_;
 
     utils::statistics::Entry statistics_holder_;
     utils::statistics::Entry subscribe_statistics_holder_;

redis/src/storages/redis/component.cpp

@@ -128,6 +128,9 @@ Redis::Redis(const ComponentConfig& config, const ComponentContext& component_co
 
     config_subscription_ = config_.UpdateAndListen(this, "redis", &Redis::OnConfigUpdate);
 
+    auto& secdist = component_context.FindComponent<Secdist>();
+    secdist_subscription_ = secdist.GetStorage().UpdateAndListen(this, "redis", &Redis::OnSecdistUpdate);
+
     auto& statistics_storage = component_context.FindComponent<components::StatisticsStorage>().GetStorage();
 
     statistics_holder_ = statistics_storage.RegisterWriter(kStatisticsName, [this](utils::statistics::Writer& writer) {
@@ -323,6 +326,22 @@ void Redis::OnConfigUpdate(const dynamic_config::Snapshot& cfg) {
     }
 }
 
+void Redis::OnSecdistUpdate(const storages::secdist::SecdistConfig& cfg) {
+    for (auto& [db, sentinel] : sentinels_) {
+        const auto& config_name = sentinel->ShardGroupName();
+        const auto& settings = cfg.Get<storages::secdist::RedisMapSettings>().GetSettings(config_name);
+
+        std::vector<storages::redis::ConnectionInfo> cii;
+        for (const auto& host_port : settings.sentinels) {
+            storages::redis::ConnectionInfo ci(host_port.host, host_port.port, settings.password);
+
+            cii.push_back(ci);
+        }
+
+        sentinels_.at(db)->SetConnectionInfo(cii);
+    }
+}
+
 yaml_config::Schema Redis::GetStaticConfigSchema() {
     return yaml_config::MergeSchemas<ComponentBase>(R"(
 type: object

redis/src/storages/redis/impl/cluster_sentinel_impl.cpp

@@ -348,6 +348,10 @@ class ClusterTopologyHolder : public std::enable_shared_from_this<ClusterTopolog
         }
     }
 
+    void SetConnectionInfo(const std::vector<ConnectionInfoInt>& info_array) {
+        sentinels_->SetConnectionInfo(info_array);
+    }
+
     static size_t GetClusterSlotsCalledCounter() { return cluster_slots_call_counter_.load(std::memory_order_relaxed); }
 
     boost::signals2::signal<void(HostPort, Redis::State)>& GetSignalNodeStateChanged() {
@@ -1051,6 +1055,10 @@ size_t ClusterSentinelImpl::GetClusterSlotsCalledCounter() {
     return ClusterTopologyHolder::GetClusterSlotsCalledCounter();
 }
 
+void ClusterSentinelImpl::SetConnectionInfo(const std::vector<ConnectionInfoInt>& info_array) {
+    topology_holder_->SetConnectionInfo(info_array);
+}
+
 PublishSettings ClusterSentinelImpl::GetPublishSettings() {
     return PublishSettings{kUnknownShard, false, CommandControl::Strategy::kEveryDc};
 }

redis/src/storages/redis/impl/cluster_sentinel_impl.hpp

@@ -69,6 +69,8 @@ class ClusterSentinelImpl : public SentinelImplBase {
 
     static size_t GetClusterSlotsCalledCounter();
 
+    void SetConnectionInfo(const std::vector<ConnectionInfoInt>& info_array) override;
+
 private:
     void AsyncCommandFailed(const SentinelCommand& scommand);
     void EnqueueCommand(const SentinelCommand& command);

redis/src/storages/redis/impl/sentinel.cpp

@@ -73,7 +73,8 @@ Sentinel::Sentinel(
     const testsuite::RedisControl& testsuite_redis_control,
     ConnectionMode mode
 )
-    : thread_pools_(thread_pools),
+    : shard_group_name_(shard_group_name),
+      thread_pools_(thread_pools),
       secdist_default_command_control_(command_control),
       testsuite_redis_control_(testsuite_redis_control) {
     config_default_command_control_.Set(std::make_shared<CommandControl>(secdist_default_command_control_));
@@ -414,6 +415,16 @@ void Sentinel::SetConfigDefaultCommandControl(const std::shared_ptr<CommandContr
     config_default_command_control_.Set(cc);
 }
 
+const std::string& Sentinel::ShardGroupName() const { return shard_group_name_; }
+
+void Sentinel::SetConnectionInfo(std::vector<ConnectionInfo> info_array) {
+    std::vector<ConnectionInfoInt> cii;
+    cii.reserve(info_array.size());
+    for (const auto& ci : info_array) cii.emplace_back(ci);
+
+    impl_->SetConnectionInfo(cii);
+}
+
 std::vector<std::shared_ptr<const Shard>> Sentinel::GetMasterShards() const { return impl_->GetMasterShards(); }
 
 void Sentinel::CheckRenameParams(const std::string& key, const std::string& newkey) const {

redis/src/storages/redis/impl/sentinel.hpp

@@ -184,6 +184,9 @@ class Sentinel {
 
     virtual void SetConfigDefaultCommandControl(const std::shared_ptr<CommandControl>& cc);
 
+    void SetConnectionInfo(std::vector<ConnectionInfo> info_array);
+    const std::string& ShardGroupName() const;
+
     using UserMessageCallback = std::function<Outcome(const std::string& channel, const std::string& message)>;
     using UserPmessageCallback =
         std::function<Outcome(const std::string& pattern, const std::string& channel, const std::string& message)>;
@@ -228,6 +231,7 @@ class Sentinel {
 
     friend class Transaction;
 
+    const std::string shard_group_name_;
     std::shared_ptr<ThreadPools> thread_pools_;
     std::unique_ptr<engine::ev::ThreadControl> sentinel_thread_control_;
     CommandControl secdist_default_command_control_;

redis/src/storages/redis/impl/sentinel_impl.cpp

@@ -99,6 +99,13 @@ SentinelImpl::SentinelImpl(
 
 SentinelImpl::~SentinelImpl() { Stop(); }
 
+void SentinelImpl::SetSentinelConnectionInfo(const std::vector<ConnectionInfo>& sentinel_conns) {
+    std::vector<ConnectionInfoInt> cii;
+    cii.reserve(sentinel_conns.size());
+    for (const auto& conn : sentinel_conns) cii.emplace_back(ConnectionInfoInt{conn});
+    sentinels_->SetConnectionInfo(cii);
+}
+
 std::unordered_map<ServerId, size_t, ServerIdHasher>
 SentinelImpl::GetAvailableServersWeighted(size_t shard_idx, bool with_master, const CommandControl& cc) const {
     return master_shards_.at(shard_idx)->GetAvailableServersWeighted(with_master, cc);
@@ -1022,6 +1029,10 @@ bool SentinelImpl::ConnectedStatus::Wait(engine::Deadline deadline, const Pred&
     return cv_.WaitUntil(lock, deadline, pred);
 }
 
+void SentinelImpl::SetConnectionInfo(const std::vector<ConnectionInfoInt>& info_array) {
+    sentinels_->SetConnectionInfo(info_array);
+}
+
 }  // namespace storages::redis::impl
 
 USERVER_NAMESPACE_END

redis/src/storages/redis/impl/sentinel_impl.hpp

@@ -85,6 +85,7 @@ class SentinelImplBase {
     virtual void SetRetryBudgetSettings(const utils::RetryBudgetSettings& retry_budget_settings) = 0;
 
     virtual PublishSettings GetPublishSettings() = 0;
+    virtual void SetConnectionInfo(const std::vector<ConnectionInfoInt>& info_array) = 0;
 };
 
 bool AdjustDeadline(const SentinelImplBase::SentinelCommand& scommand, const dynamic_config::Snapshot& config);
@@ -110,6 +111,8 @@ class SentinelImpl : public SentinelImplBase {
     );
     ~SentinelImpl() override;
 
+    void SetSentinelConnectionInfo(const std::vector<ConnectionInfo>& sentinel_conns);
+
     std::unordered_map<ServerId, size_t, ServerIdHasher>
     GetAvailableServersWeighted(size_t shard_idx, bool with_master, const CommandControl& cc) const override;
 
@@ -139,6 +142,8 @@ class SentinelImpl : public SentinelImplBase {
     void SetRetryBudgetSettings(const utils::RetryBudgetSettings& retry_budget_settings) override;
     PublishSettings GetPublishSettings() override;
 
+    void SetConnectionInfo(const std::vector<ConnectionInfoInt>& info_array) override;
+
 private:
     static constexpr const std::chrono::milliseconds cluster_slots_timeout_ = std::chrono::milliseconds(4000);