diff options
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); }; |