feat userver: add dynamic configuration for postgres distlock

Dynamic config @ref POSTGRES_DISTLOCK_SETTINGS now allows to dynamically change some interval settings for components inherited from @ref storages::postgres::DistLockComponentBase. There is also `run_on_hosts` setting that allows to manually distribute distlocks to hosts, as well as to dynamically stop a failing distlock.

Tests: a test added to `uservice-sample`
commit_hash:2f0b3e1fdb6fb94005df1576c8498ba14957347a

Author: aleksey7sidor

.mapping.json

@@ -3285,6 +3285,7 @@
   "postgresql/dynamic_configs/POSTGRES_CONNLIMIT_MODE_AUTO_ENABLED.yaml":"taxi/uservices/userver/postgresql/dynamic_configs/POSTGRES_CONNLIMIT_MODE_AUTO_ENABLED.yaml",
   "postgresql/dynamic_configs/POSTGRES_DEADLINE_PROPAGATION_VERSION.yaml":"taxi/uservices/userver/postgresql/dynamic_configs/POSTGRES_DEADLINE_PROPAGATION_VERSION.yaml",
   "postgresql/dynamic_configs/POSTGRES_DEFAULT_COMMAND_CONTROL.yaml":"taxi/uservices/userver/postgresql/dynamic_configs/POSTGRES_DEFAULT_COMMAND_CONTROL.yaml",
+  "postgresql/dynamic_configs/POSTGRES_DISTLOCK_SETTINGS.yaml":"taxi/uservices/userver/postgresql/dynamic_configs/POSTGRES_DISTLOCK_SETTINGS.yaml",
   "postgresql/dynamic_configs/POSTGRES_HANDLERS_COMMAND_CONTROL.yaml":"taxi/uservices/userver/postgresql/dynamic_configs/POSTGRES_HANDLERS_COMMAND_CONTROL.yaml",
   "postgresql/dynamic_configs/POSTGRES_OMIT_DESCRIBE_IN_EXECUTE.yaml":"taxi/uservices/userver/postgresql/dynamic_configs/POSTGRES_OMIT_DESCRIBE_IN_EXECUTE.yaml",
   "postgresql/dynamic_configs/POSTGRES_QUERIES_COMMAND_CONTROL.yaml":"taxi/uservices/userver/postgresql/dynamic_configs/POSTGRES_QUERIES_COMMAND_CONTROL.yaml",

postgresql/dynamic_configs/POSTGRES_DISTLOCK_SETTINGS.yaml

@@ -0,0 +1,57 @@
+default: {}
+description: |
+    Configuration of periodic tasks executed with a distributed lock.
+    A dictionary where the keys are task names.
+    If hosts are specified in run_on_hosts, the task will switch to
+    any of them at the first opportunity and will keep running
+    there until a new preferred host is specified or a failover occurs due to
+    an error. By adjusting the timing settings,
+    you can override the lock acquisition policy.
+schema:
+    type: object
+    additionalProperties:
+        $ref: '#/definitions/TaskSettings'
+    definitions:
+        TaskSettings:
+            type: object
+            description: Periodic task policy settings
+            additionalProperties: false
+            properties:
+                run_on_hosts:
+                    description: |
+                        USE IsCancelAdvised() OR IT DOESNT WORK
+                        List of hosts on which the task
+                        is allowed to run. The task will start on
+                        any of the specified hosts—without
+                        prioritization.
+                    type: array
+                    x-usrv-cpp-type: std::unordered_set
+                    items:
+                        type: string
+                acquire_interval:
+                    description: |
+                        How often to try to acquire the lock.
+                    type: integer
+                    x-usrv-cpp-type: std::chrono::milliseconds
+                prolong_interval:
+                    description: |
+                        How often to attempt to renew
+                        the lock while holding it.
+                    type: integer
+                    x-usrv-cpp-type: std::chrono::milliseconds
+                lock_ttl:
+                    description: Duration of lock
+                    type: integer
+                    x-usrv-cpp-type: std::chrono::milliseconds
+                forced_stop_margin:
+                    description: |
+                        How long to allow the component to
+                        finish work when it cannot acquire the lock.
+                    type: integer
+                    x-usrv-cpp-type: std::chrono::milliseconds
+                worker_func_restart_delay:
+                    description: |
+                        Delay before attempting to restart a
+                        crashed component.
+                    type: integer
+                    x-usrv-cpp-type: std::chrono::milliseconds

postgresql/include/userver/storages/postgres/dist_lock_component_base.hpp

@@ -5,6 +5,8 @@
 
 #include <userver/components/component_base.hpp>
 #include <userver/dist_lock/dist_locked_worker.hpp>
+#include <userver/dynamic_config/snapshot.hpp>
+#include <userver/dynamic_config/source.hpp>
 #include <userver/storages/postgres/dist_lock_strategy.hpp>
 #include <userver/utils/statistics/entry.hpp>
 
@@ -39,6 +41,7 @@ namespace storages::postgres {
 ///            lock-ttl: 10s
 ///            autostart: true
 /// ```
+/// See config `POSTGRES_DISTLOCK_SETTINGS`, some of parameters can be dynamically overridden.
 ///
 /// ## Static options:
 /// name           | Description  | Default value
@@ -92,7 +95,11 @@ class DistLockComponentBase : public components::ComponentBase {
     /// ```cpp
     /// void MyDistLockComponent::DoWork()
     /// {
-    ///     while (!engine::ShouldCancel())
+    ///     // `IsCancelAdvised` is advisory/soft signal to stop the task.
+    ///     // Check it in every independent processing iteration.
+    ///     // Whereas @ref engine::current_task::ShouldCancel() checks as frequently,
+    ///     // as you can to honor low-level task cancellation.
+    ///     while (!IsCancelAdvised())
     ///     {
     ///         // Start a new trace_id
     ///         auto span = tracing::Span::MakeRootSpan("my-dist-lock");
@@ -125,10 +132,24 @@ class DistLockComponentBase : public components::ComponentBase {
     /// Must be called in dtr
     void StopDistLock();
 
+    /// Check this method when going for the next independent processing
+    /// iteration. Whereas @ref engine::current_task::ShouldCancel()
+    /// checks as frequently, as you can to honor low-level task cancellation.
+    bool IsCancelAdvised() const;
+
 private:
+    bool ShouldRunOnHost(const dynamic_config::Snapshot& config) const;
+    void OnConfigUpdate(const dynamic_config::Diff& diff);
+
+    dynamic_config::Source config_;
+    const std::string name_;
+    const std::string real_host_name_;
     std::unique_ptr<dist_lock::DistLockedWorker> worker_;
     bool autostart_;
     bool testsuite_enabled_{false};
+    dist_lock::DistLockSettings default_settings_;
+
+    concurrent::AsyncEventSubscriberScope subscription_token_;
 
     // Subscriptions must be the last fields.
     USERVER_NAMESPACE::utils::statistics::Entry statistics_holder_;

postgresql/library.yaml

@@ -22,3 +22,4 @@ configs:
       - POSTGRES_OMIT_DESCRIBE_IN_EXECUTE
       - POSTGRES_QUERIES_COMMAND_CONTROL
       - POSTGRES_STATEMENT_METRICS_SETTINGS
+      - POSTGRES_DISTLOCK_SETTINGS

postgresql/src/storages/postgres/dist_lock_component_base.cpp

@@ -3,14 +3,19 @@
 #include <userver/components/component.hpp>
 #include <userver/components/statistics_storage.hpp>
 #include <userver/dist_lock/dist_lock_settings.hpp>
+#include <userver/dynamic_config/storage/component.hpp>
+#include <userver/hostinfo/blocking/get_hostname.hpp>
 #include <userver/storages/postgres/component.hpp>
 #include <userver/testsuite/tasks.hpp>
+#include <userver/utils/algo.hpp>
 #include <userver/yaml_config/merge_schemas.hpp>
 
 #ifndef ARCADIA_ROOT
 #include "generated/src/storages/postgres/dist_lock_component_base.yaml.hpp"  // Y_IGNORE
 #endif
 
+#include <dynamic_config/variables/POSTGRES_DISTLOCK_SETTINGS.hpp>
+
 USERVER_NAMESPACE_BEGIN
 
 namespace storages::postgres {
@@ -19,7 +24,11 @@ DistLockComponentBase::DistLockComponentBase(
     const components::ComponentConfig& component_config,
     const components::ComponentContext& component_context
 )
-    : components::ComponentBase(component_config, component_context) {
+    : components::ComponentBase(component_config, component_context),
+      config_(component_context.FindComponent<components::DynamicConfig>().GetSource()),
+      name_(component_config.Name()),
+      real_host_name_(hostinfo::blocking::GetRealHostName())
+{
     auto cluster =
         component_context.FindComponent<components::Postgres>(component_config["cluster"].As<std::string>())
             .GetCluster();
@@ -38,29 +47,32 @@ DistLockComponentBase::DistLockComponentBase(
     settings.worker_func_restart_delay =
         component_config["restart-delay"].As<std::chrono::milliseconds>(settings.worker_func_restart_delay);
 
+    default_settings_ = settings;
     auto strategy = std::make_shared<DistLockStrategy>(std::move(cluster), table, lock_name, settings);
 
     auto task_processor_name = component_config["task-processor"].As<std::optional<std::string>>();
     auto* task_processor =
         task_processor_name ? &component_context.GetTaskProcessor(task_processor_name.value()) : nullptr;
 
     auto locker_log_level = logging::LevelFromString(component_config["locker-log-level"].As<std::string>("info"));
-
     worker_ = std::make_unique<dist_lock::DistLockedWorker>(
         lock_name,
         [this]() {
-            if (testsuite_enabled_) {
-                DoWorkTestsuite();
-            } else {
-                DoWork();
+            const auto snapshot = config_.GetSnapshot();
+            if (ShouldRunOnHost(snapshot)) {
+                if (testsuite_enabled_) {
+                    DoWorkTestsuite();
+                } else {
+                    DoWork();
+                }
             }
         },
         std::move(strategy),
         settings,
         task_processor,
         locker_log_level
     );
-
+    subscription_token_ = config_.UpdateAndListen(this, name_, &DistLockComponentBase::OnConfigUpdate);
     autostart_ = component_config["autostart"].As<bool>(false);
 
     auto& statistics_storage = component_context.FindComponent<components::StatisticsStorage>();
@@ -80,7 +92,10 @@ DistLockComponentBase::DistLockComponentBase(
     }
 }
 
-DistLockComponentBase::~DistLockComponentBase() { statistics_holder_.Unregister(); }
+DistLockComponentBase::~DistLockComponentBase() {
+    statistics_holder_.Unregister();
+    subscription_token_.Unsubscribe();
+}
 
 dist_lock::DistLockedWorker& DistLockComponentBase::GetWorker() { return *worker_; }
 
@@ -97,6 +112,54 @@ void DistLockComponentBase::AutostartDistLock() {
 
 void DistLockComponentBase::StopDistLock() { worker_->Stop(); }
 
+void DistLockComponentBase::OnConfigUpdate(const dynamic_config::Diff& diff) {
+    const auto& old_snapshot_opt = diff.previous;
+    const auto& new_snapshot = diff.current;
+
+    const auto& new_settings = new_snapshot[::dynamic_config::POSTGRES_DISTLOCK_SETTINGS];
+    const auto* new_overrides = utils::FindOrNullptr(new_settings.extra, name_);
+
+    if (old_snapshot_opt.has_value()) {
+        const auto& old_settings = (*old_snapshot_opt)[::dynamic_config::POSTGRES_DISTLOCK_SETTINGS];
+        const auto* old_overrides = utils::FindOrNullptr(old_settings.extra, name_);
+
+        if (!old_overrides && !new_overrides) {
+            return;
+        }
+        if (old_overrides && new_overrides && *old_overrides == *new_overrides) {
+            return;
+        }
+    }
+    auto& worker = GetWorker();
+    if (new_overrides) {
+        dist_lock::DistLockSettings settings;
+        settings.acquire_interval = new_overrides->acquire_interval.value_or(default_settings_.acquire_interval);
+        settings.prolong_interval = new_overrides->prolong_interval.value_or(default_settings_.prolong_interval);
+        settings.lock_ttl = new_overrides->lock_ttl.value_or(default_settings_.lock_ttl);
+        settings.forced_stop_margin = new_overrides->forced_stop_margin.value_or(default_settings_.forced_stop_margin);
+        settings.worker_func_restart_delay =
+            new_overrides->worker_func_restart_delay.value_or(default_settings_.worker_func_restart_delay);
+
+        worker.UpdateSettings(settings);
+    } else {
+        worker.UpdateSettings(default_settings_);
+    }
+}
+bool DistLockComponentBase::ShouldRunOnHost(const dynamic_config::Snapshot& config) const {
+    const auto& worker_settings_config = config[::dynamic_config::POSTGRES_DISTLOCK_SETTINGS];
+    if (const auto* settings = utils::FindOrNullptr(worker_settings_config.extra, name_)) {
+        if (const auto& hosts = settings->run_on_hosts) {
+            return hosts->count(real_host_name_);
+        }
+    }
+    return true;
+}
+
+bool DistLockComponentBase::IsCancelAdvised() const {
+    const auto snapshot = config_.GetSnapshot();
+    return !ShouldRunOnHost(snapshot);
+}
+
 yaml_config::Schema DistLockComponentBase::GetStaticConfigSchema() {
     return yaml_config::MergeSchemasFromResource<
         components::ComponentBase>("src/storages/postgres/dist_lock_component_base.yaml");