Rewire Bouncer configuration flow

Removes own `ConfigFetcher` in favor of pushing reconfiguration
responsibilities onto the components owning the Bouncer instance.

The current "superclass calls into subclass" approach isn't
ideal, but the longer term plan is to hoist all config subscriptions
out of `StorageNode` and into the higher-level `Process` structure.

9 files changed, 82 insertions, 55 deletions

diff --git a/storage/src/tests/storageserver/bouncertest.cpp b/storage/src/tests/storageserver/bouncertest.cpp
index 5b7d279537e..225b3c94120 100644
--- a/storage/src/tests/storageserver/bouncertest.cpp
+++ b/storage/src/tests/storageserver/bouncertest.cpp
@@ -1,20 +1,22 @@
 // Copyright Vespa.ai. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 
-#include <vespa/storageapi/message/bucket.h>
-#include <vespa/storageapi/message/state.h>
-#include <vespa/storageapi/message/stat.h>
-#include <vespa/storage/storageserver/bouncer.h>
-#include <vespa/storage/storageserver/bouncer_metrics.h>
-#include <tests/common/teststorageapp.h>
-#include <tests/common/testhelper.h>
 #include <tests/common/dummystoragelink.h>
+#include <tests/common/testhelper.h>
+#include <tests/common/teststorageapp.h>
+#include <vespa/config/common/exceptions.h>
+#include <vespa/config/helper/configgetter.hpp>
 #include <vespa/document/bucket/fixed_bucket_spaces.h>
-#include <vespa/document/test/make_document_bucket.h>
 #include <vespa/document/fieldset/fieldsets.h>
-#include <vespa/storageapi/message/persistence.h>
+#include <vespa/document/test/make_document_bucket.h>
 #include <vespa/persistence/spi/bucket_limits.h>
+#include <vespa/storage/config/config-stor-bouncer.h>
+#include <vespa/storage/storageserver/bouncer.h>
+#include <vespa/storage/storageserver/bouncer_metrics.h>
+#include <vespa/storageapi/message/bucket.h>
+#include <vespa/storageapi/message/persistence.h>
+#include <vespa/storageapi/message/stat.h>
+#include <vespa/storageapi/message/state.h>
 #include <vespa/vdslib/state/clusterstate.h>
-#include <vespa/config/common/exceptions.h>
 #include <vespa/vespalib/gtest/gtest.h>
 
 using document::test::makeDocumentBucket;
@@ -73,7 +75,10 @@ void BouncerTest::setUpAsNode(const lib::NodeType& type) {
         _node.reset(new TestDistributorApp(NodeIndex(2), config.getConfigId()));
     }
     _upper.reset(new DummyStorageLink());
-    _manager = new Bouncer(_node->getComponentRegister(), config::ConfigUri(config.getConfigId()));
+    using StorBouncerConfig = vespa::config::content::core::StorBouncerConfig;
+    auto cfg_uri = config::ConfigUri(config.getConfigId());
+    auto cfg = config::ConfigGetter<StorBouncerConfig>::getConfig(cfg_uri.getConfigId(), cfg_uri.getContext());
+    _manager = new Bouncer(_node->getComponentRegister(), *cfg);
     _lower = new DummyStorageLink();
     _upper->push_back(std::unique_ptr<StorageLink>(_manager));
     _upper->push_back(std::unique_ptr<StorageLink>(_lower));
@@ -225,9 +230,9 @@ void
 BouncerTest::configureRejectionThreshold(int newThreshold)
 {
     using Builder = vespa::config::content::core::StorBouncerConfigBuilder;
-    auto config = std::make_unique<Builder>();
-    config->feedRejectionPriorityThreshold = newThreshold;
-    _manager->configure(std::move(config));
+    Builder config;
+    config.feedRejectionPriorityThreshold = newThreshold;
+    _manager->on_configure(config);
 }
 
 TEST_F(BouncerTest, reject_lower_prioritized_feed_messages_when_configured) {
diff --git a/storage/src/vespa/storage/storageserver/bouncer.cpp b/storage/src/vespa/storage/storageserver/bouncer.cpp
index 78f248e2b90..7e3b21ef33a 100644
--- a/storage/src/vespa/storage/storageserver/bouncer.cpp
+++ b/storage/src/vespa/storage/storageserver/bouncer.cpp
@@ -21,34 +21,19 @@ LOG_SETUP(".bouncer");
 
 namespace storage {
 
-Bouncer::Bouncer(StorageComponentRegister& compReg, const config::ConfigUri & configUri)
+Bouncer::Bouncer(StorageComponentRegister& compReg, const StorBouncerConfig& bootstrap_config)
     : StorageLink("Bouncer", MsgDownOnFlush::Disallowed, MsgUpOnClosed::Allowed),
-      _config(new vespa::config::content::core::StorBouncerConfig()),
+      _config(std::make_unique<vespa::config::content::core::StorBouncerConfig>(bootstrap_config)),
       _component(compReg, "bouncer"),
       _lock(),
       _baselineNodeState("s:i"),
       _derivedNodeStates(),
       _clusterState(&lib::State::UP),
-      _configFetcher(std::make_unique<config::ConfigFetcher>(configUri.getContext())),
       _metrics(std::make_unique<BouncerMetrics>()),
       _closed(false)
 {
     _component.getStateUpdater().addStateListener(*this);
     _component.registerMetric(*_metrics);
-    // Register for config. Normally not critical, so catching config
-    // exception allowing program to continue if missing/faulty config.
-    try {
-        if (!configUri.empty()) {
-            _configFetcher->subscribe<vespa::config::content::core::StorBouncerConfig>(configUri.getConfigId(), this);
-            _configFetcher->start();
-        } else {
-            LOG(info, "No config id specified. Using defaults rather than config");
-        }
-    } catch (config::InvalidConfigException& e) {
-        LOG(info, "Bouncer failed to load config '%s'. This "
-                  "is not critical since it has sensible defaults: %s",
-            configUri.getConfigId().c_str(), e.what());
-    }
 }
 
 Bouncer::~Bouncer()
@@ -68,19 +53,18 @@ Bouncer::print(std::ostream& out, bool verbose,
 void
 Bouncer::onClose()
 {
-    _configFetcher->close();
     _component.getStateUpdater().removeStateListener(*this);
     std::lock_guard guard(_lock);
     _closed = true;
 }
 
 void
-Bouncer::configure(std::unique_ptr<vespa::config::content::core::StorBouncerConfig> config)
+Bouncer::on_configure(const vespa::config::content::core::StorBouncerConfig& config)
 {
-    log_config_received(*config);
-    validateConfig(*config);
+    validateConfig(config);
+    auto new_config = std::make_unique<StorBouncerConfig>(config);
     std::lock_guard lock(_lock);
-    _config = std::move(config);
+    _config = std::move(new_config);
 }
 
 const BouncerMetrics& Bouncer::metrics() const noexcept {
diff --git a/storage/src/vespa/storage/storageserver/bouncer.h b/storage/src/vespa/storage/storageserver/bouncer.h
index 1038e94ee94..44c7a16f3dc 100644
--- a/storage/src/vespa/storage/storageserver/bouncer.h
+++ b/storage/src/vespa/storage/storageserver/bouncer.h
@@ -1,12 +1,7 @@
 // Copyright Vespa.ai. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 /**
- * @class storage::Bouncer
- * @ingroup storageserver
- *
- * @brief Denies messages from entering if state is not good.
- *
- * If we are not in up state, but process is still running, only a few
- * messages should be allowed through. This link stops all messages not allowed.
+ * Component which rejects messages that can not be accepted by the node in
+ * its current state.
  */
 
 #pragma once
@@ -29,28 +24,28 @@ namespace storage {
 struct BouncerMetrics;
 
 class Bouncer : public StorageLink,
-                private StateListener,
-                private config::IFetcherCallback<vespa::config::content::core::StorBouncerConfig>
+                private StateListener
 {
-    std::unique_ptr<vespa::config::content::core::StorBouncerConfig> _config;
+    using StorBouncerConfig = vespa::config::content::core::StorBouncerConfig;
+
+    std::unique_ptr<StorBouncerConfig> _config;
     StorageComponent _component;
     std::mutex       _lock;
     lib::NodeState   _baselineNodeState;
     using BucketSpaceNodeStateMapping = std::unordered_map<document::BucketSpace, lib::NodeState, document::BucketSpace::hash>;
     BucketSpaceNodeStateMapping _derivedNodeStates;
     const lib::State* _clusterState;
-    std::unique_ptr<config::ConfigFetcher> _configFetcher;
     std::unique_ptr<BouncerMetrics> _metrics;
     bool _closed;
 
 public:
-    Bouncer(StorageComponentRegister& compReg, const config::ConfigUri & configUri);
+    Bouncer(StorageComponentRegister& compReg, const StorBouncerConfig& bootstrap_config);
     ~Bouncer() override;
 
     void print(std::ostream& out, bool verbose,
                const std::string& indent) const override;
 
-    void configure(std::unique_ptr<vespa::config::content::core::StorBouncerConfig> config) override;
+    void on_configure(const StorBouncerConfig& config);
     const BouncerMetrics& metrics() const noexcept;
 
 private:
diff --git a/storage/src/vespa/storage/storageserver/distributornode.cpp b/storage/src/vespa/storage/storageserver/distributornode.cpp
index 31785e19681..0c7bee01715 100644
--- a/storage/src/vespa/storage/storageserver/distributornode.cpp
+++ b/storage/src/vespa/storage/storageserver/distributornode.cpp
@@ -32,7 +32,8 @@ DistributorNode::DistributorNode(
       _timestamp_second_counter(0),
       _intra_second_pseudo_usec_counter(0),
       _num_distributor_stripes(num_distributor_stripes),
-      _retrievedCommunicationManager(std::move(communicationManager)) // may be nullptr
+      _retrievedCommunicationManager(std::move(communicationManager)), // may be nullptr
+      _bouncer(nullptr)
 {
     if (storage_chain_builder) {
         set_storage_chain_builder(std::move(storage_chain_builder));
@@ -94,7 +95,9 @@ DistributorNode::createChain(IStorageChainBuilder &builder)
     }
     std::unique_ptr<StateManager> stateManager(releaseStateManager());
 
-    builder.add(std::make_unique<Bouncer>(dcr, _configUri));
+    auto bouncer = std::make_unique<Bouncer>(dcr, bouncer_config());
+    _bouncer = bouncer.get();
+    builder.add(std::move(bouncer));
     // Distributor instance registers a host info reporter with the state
     // manager, which is safe since the lifetime of said state manager
     // extends to the end of the process.
@@ -140,4 +143,9 @@ DistributorNode::pause()
     return {};
 }
 
+void DistributorNode::on_bouncer_config_changed() {
+    assert(_bouncer);
+    _bouncer->on_configure(bouncer_config());
+}
+
 } // storage
diff --git a/storage/src/vespa/storage/storageserver/distributornode.h b/storage/src/vespa/storage/storageserver/distributornode.h
index ac3cad30036..b725f320851 100644
--- a/storage/src/vespa/storage/storageserver/distributornode.h
+++ b/storage/src/vespa/storage/storageserver/distributornode.h
@@ -19,6 +19,7 @@ namespace storage {
 
 namespace distributor { class DistributorStripePool; }
 
+class Bouncer;
 class IStorageChainBuilder;
 
 class DistributorNode
@@ -34,6 +35,7 @@ class DistributorNode
     uint32_t _intra_second_pseudo_usec_counter;
     uint32_t _num_distributor_stripes;
     std::unique_ptr<StorageLink> _retrievedCommunicationManager;
+    Bouncer* _bouncer;
 
     // If the current wall clock is more than the below number of seconds into the
     // past when compared to the highest recorded wall clock second time stamp, abort
@@ -65,6 +67,7 @@ private:
     void initializeNodeSpecific() override;
     void createChain(IStorageChainBuilder &builder) override;
     api::Timestamp generate_unique_timestamp() override;
+    void on_bouncer_config_changed() override;
 
     /**
      * Shut down necessary distributor-specific components before shutting
diff --git a/storage/src/vespa/storage/storageserver/servicelayernode.cpp b/storage/src/vespa/storage/storageserver/servicelayernode.cpp
index 0cd5ffeba68..e76625ccca7 100644
--- a/storage/src/vespa/storage/storageserver/servicelayernode.cpp
+++ b/storage/src/vespa/storage/storageserver/servicelayernode.cpp
@@ -32,6 +32,7 @@ ServiceLayerNode::ServiceLayerNode(const config::ConfigUri & configUri, ServiceL
       _context(context),
       _persistenceProvider(persistenceProvider),
       _externalVisitors(externalVisitors),
+      _bouncer(nullptr),
       _bucket_manager(nullptr),
       _fileStorManager(nullptr),
       _init_has_been_called(false)
@@ -166,7 +167,9 @@ ServiceLayerNode::createChain(IStorageChainBuilder &builder)
     auto communication_manager = std::make_unique<CommunicationManager>(compReg, _configUri, communication_manager_config());
     _communicationManager = communication_manager.get();
     builder.add(std::move(communication_manager));
-    builder.add(std::make_unique<Bouncer>(compReg, _configUri));
+    auto bouncer = std::make_unique<Bouncer>(compReg, bouncer_config());
+    _bouncer = bouncer.get();
+    builder.add(std::move(bouncer));
     auto merge_throttler_up = std::make_unique<MergeThrottler>(_configUri, compReg);
     auto merge_throttler = merge_throttler_up.get();
     builder.add(std::move(merge_throttler_up));
@@ -215,4 +218,9 @@ void ServiceLayerNode::perform_post_chain_creation_init_steps() {
     _fileStorManager->complete_internal_initialization();
 }
 
+void ServiceLayerNode::on_bouncer_config_changed() {
+    assert(_bouncer);
+    _bouncer->on_configure(bouncer_config());
+}
+
 } // storage
diff --git a/storage/src/vespa/storage/storageserver/servicelayernode.h b/storage/src/vespa/storage/storageserver/servicelayernode.h
index 91a799c1295..4a5c2e37bb3 100644
--- a/storage/src/vespa/storage/storageserver/servicelayernode.h
+++ b/storage/src/vespa/storage/storageserver/servicelayernode.h
@@ -20,6 +20,7 @@ namespace storage {
 
 namespace spi { struct PersistenceProvider; }
 
+class Bouncer;
 class BucketManager;
 class FileStorManager;
 
@@ -35,8 +36,9 @@ class ServiceLayerNode
 
     // FIXME: Should probably use the fetcher in StorageNode
     std::unique_ptr<config::ConfigFetcher>   _configFetcher;
-    BucketManager                          * _bucket_manager;
-    FileStorManager                        * _fileStorManager;
+    Bouncer*                                 _bouncer;
+    BucketManager*                           _bucket_manager;
+    FileStorManager*                         _fileStorManager;
     bool                                     _init_has_been_called;
 
 public:
@@ -67,6 +69,7 @@ private:
     documentapi::Priority::Value toDocumentPriority(uint8_t storagePriority) const override;
     void createChain(IStorageChainBuilder &builder) override;
     void removeConfigSubscriptions() override;
+    void on_bouncer_config_changed() override;
 };
 
 } // storage
diff --git a/storage/src/vespa/storage/storageserver/storagenode.cpp b/storage/src/vespa/storage/storageserver/storagenode.cpp
index f835b286165..3b0f3d875cd 100644
--- a/storage/src/vespa/storage/storageserver/storagenode.cpp
+++ b/storage/src/vespa/storage/storageserver/storagenode.cpp
@@ -106,6 +106,7 @@ void
 StorageNode::subscribeToConfigs()
 {
     _configFetcher = std::make_unique<config::ConfigFetcher>(_configUri.getContext());
+    _configFetcher->subscribe<StorBouncerConfig>(_configUri.getConfigId(), this);
     _configFetcher->subscribe<BucketspacesConfig>(_configUri.getConfigId(), this);
     _configFetcher->subscribe<CommunicationManagerConfig>(_configUri.getConfigId(), this);
     _configFetcher->subscribe<StorDistributionConfig>(_configUri.getConfigId(), this);
@@ -115,6 +116,7 @@ StorageNode::subscribeToConfigs()
 
     // All the below config instances were synchronously populated as part of start()ing the config fetcher
     std::lock_guard configLockGuard(_configLock);
+    _bouncer_config.promote_staging_to_active();
     _bucket_spaces_config.promote_staging_to_active();
     _comm_mgr_config.promote_staging_to_active();
     _distribution_config.promote_staging_to_active();
@@ -318,6 +320,10 @@ StorageNode::handleLiveConfigUpdate(const InitialGuard & initGuard)
         _comm_mgr_config.promote_staging_to_active();
         _communicationManager->on_configure(communication_manager_config());
     }
+    if (_bouncer_config.staging) {
+        _bouncer_config.promote_staging_to_active();
+        on_bouncer_config_changed();
+    }
 }
 
 void
@@ -441,10 +447,15 @@ StorageNode::configure(std::unique_ptr<BucketspacesConfig> config) {
 }
 
 void
-StorageNode::configure(std::unique_ptr<CommunicationManagerConfig > config) {
+StorageNode::configure(std::unique_ptr<CommunicationManagerConfig> config) {
     stage_config_change(_comm_mgr_config, std::move(config));
 }
 
+void
+StorageNode::configure(std::unique_ptr<StorBouncerConfig> config) {
+    stage_config_change(_bouncer_config, std::move(config));
+}
+
 template <typename ConfigT>
 void
 StorageNode::stage_config_change(ConfigWrapper<ConfigT>& cfg, std::unique_ptr<ConfigT> new_cfg) {
diff --git a/storage/src/vespa/storage/storageserver/storagenode.h b/storage/src/vespa/storage/storageserver/storagenode.h
index fb616eb2475..ef2879d8f8a 100644
--- a/storage/src/vespa/storage/storageserver/storagenode.h
+++ b/storage/src/vespa/storage/storageserver/storagenode.h
@@ -15,6 +15,7 @@
 #include <vespa/config/subscription/configuri.h>
 #include <vespa/document/config/config-documenttypes.h>
 #include <vespa/storage/common/doneinitializehandler.h>
+#include <vespa/storage/config/config-stor-bouncer.h>
 #include <vespa/storage/config/config-stor-communicationmanager.h>
 #include <vespa/storage/config/config-stor-server.h>
 #include <vespa/storage/storageutil/resumeguard.h>
@@ -52,6 +53,7 @@ class StorageNode : private config::IFetcherCallback<vespa::config::content::cor
                     private config::IFetcherCallback<vespa::config::content::StorDistributionConfig>,
                     private config::IFetcherCallback<vespa::config::content::core::BucketspacesConfig>,
                     private config::IFetcherCallback<vespa::config::content::core::StorCommunicationmanagerConfig>,
+                    private config::IFetcherCallback<vespa::config::content::core::StorBouncerConfig>,
                     private framework::MetricUpdateHook,
                     private DoneInitializeHandler,
                     private framework::defaultimplementation::ShutdownListener
@@ -92,6 +94,7 @@ public:
 protected:
     using BucketspacesConfig         = vespa::config::content::core::BucketspacesConfig;
     using CommunicationManagerConfig = vespa::config::content::core::StorCommunicationmanagerConfig;
+    using StorBouncerConfig          = vespa::config::content::core::StorBouncerConfig;
     using StorDistributionConfig     = vespa::config::content::StorDistributionConfig;
     using StorServerConfig           = vespa::config::content::core::StorServerConfig;
 private:
@@ -142,6 +145,7 @@ private:
     void configure(std::unique_ptr<StorDistributionConfig> config) override;
     void configure(std::unique_ptr<BucketspacesConfig>) override;
     void configure(std::unique_ptr<CommunicationManagerConfig> config) override;
+    void configure(std::unique_ptr<StorBouncerConfig> config) override;
 
 protected:
     // Lock taken while doing configuration of the server.
@@ -149,11 +153,15 @@ protected:
     std::mutex _initial_config_mutex;
     using InitialGuard = std::lock_guard<std::mutex>;
 
+    ConfigWrapper<StorBouncerConfig>          _bouncer_config;
     ConfigWrapper<BucketspacesConfig>         _bucket_spaces_config;
     ConfigWrapper<CommunicationManagerConfig> _comm_mgr_config;
     ConfigWrapper<StorDistributionConfig>     _distribution_config;
     ConfigWrapper<StorServerConfig>           _server_config;
 
+    [[nodiscard]] const StorBouncerConfig& bouncer_config() const noexcept {
+        return *_bouncer_config.active;
+    }
     [[nodiscard]] const BucketspacesConfig& bucket_spaces_config() const noexcept {
         return *_bucket_spaces_config.active;
     }
@@ -192,6 +200,8 @@ protected:
     virtual void handleLiveConfigUpdate(const InitialGuard & initGuard);
     void shutdown();
     virtual void removeConfigSubscriptions();
+
+    virtual void on_bouncer_config_changed() { /* no-op by default */ }
 public:
     void set_storage_chain_builder(std::unique_ptr<IStorageChainBuilder> builder);
 };