This is an automated email from the ASF dual-hosted git repository. fgerlits pushed a commit to branch MINIFICPP-2877_Fix-a-lock-ordering-issue in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 4e2423c318dae71577121d1533b69ce75527e255 Author: Ferenc Gerlits <[email protected]> AuthorDate: Fri Nov 6 10:34:52 2026 +0100 MINIFICPP-2877 Fix a lock ordering issue ThreadSanitizer has found a mutex lock order issue between FlowController::mutex_ (M1) and C2MetricsPublisher::metrics_mutex_ (M2): during a flow update, we lock M1 first then M2, and during heartbeats, we lock M2 first and then some metrics nodes call executeOnAllComponents(), which locks M1. This is unlikely to happen in practice, because executeOnAllComponents checks if an update is in progress before trying to acquire the M1 lock -- but the flow update could start in the tiny window between these two lines. Also, if executeOnAllComponents finds that an update is in progress, it returns early without doing its stuff, so some metrics could have been missing from the heartbeat. Claude also found a problem with metrics node types FlowInformation, QueueMetrics and ProcessorMetrics: these store raw pointers to processors or connections inside the current flow, which can become dangling after a flow update. This PR fixes both issues by removing executeOnAllComponents() and replacing it with a component state cache, which has its own leaf mutex. The cache is cleared before the flow update and resnapped afterwards, which happens under this mutex. The new MetricsDuringReloadTests unit test exercises these code paths by running several metrics serializations and flow updates in parallel. Before this change, running under ThreadSanitizer, it fails with both kinds of errors (lock order inversion, invalid read); after the change, it passes. --- controller/tests/ControllerTests.cpp | 4 +- libminifi/include/FlowController.h | 3 +- libminifi/include/RootProcessGroupWrapper.h | 10 +- libminifi/include/core/state/ComponentStateCache.h | 53 ++++++ libminifi/include/core/state/UpdateController.h | 45 ++--- .../include/core/state/nodes/ResponseNodeLoader.h | 3 +- libminifi/src/FlowController.cpp | 23 +-- libminifi/src/RootProcessGroupWrapper.cpp | 53 +++--- libminifi/src/c2/ControllerSocketProtocol.cpp | 11 +- libminifi/src/c2/FlowStatusBuilder.cpp | 2 + libminifi/src/core/state/ComponentStateCache.cpp | 55 ++++++ libminifi/src/core/state/MetricsPublisherStore.cpp | 3 +- .../src/core/state/nodes/AgentInformation.cpp | 23 ++- libminifi/src/core/state/nodes/FlowInformation.cpp | 10 +- .../src/core/state/nodes/ResponseNodeLoader.cpp | 46 +++-- .../src/core/state/nodes/SupportedOperations.cpp | 6 +- .../test/flow-tests/MetricsDuringReloadTests.cpp | 190 +++++++++++++++++++++ .../integration/OnScheduleErrorHandlingTests.cpp | 30 ++-- .../integration/StateTransactionalityTests.cpp | 29 ++-- 19 files changed, 413 insertions(+), 186 deletions(-) diff --git a/controller/tests/ControllerTests.cpp b/controller/tests/ControllerTests.cpp index 6955a13d0..4d815e057 100644 --- a/controller/tests/ControllerTests.cpp +++ b/controller/tests/ControllerTests.cpp @@ -102,8 +102,8 @@ class TestUpdateSink : public minifi::state::StateMonitor { func(*controller); } - void executeOnAllComponents(std::function<void(minifi::state::StateController&)> func) override { - func(*controller); + std::vector<minifi::state::ComponentState> getComponentStates() const override { + return {{.name = controller->getComponentName(), .uuid = controller->getComponentUUID(), .running = controller->isRunning()}}; } std::string getComponentName() const override { diff --git a/libminifi/include/FlowController.h b/libminifi/include/FlowController.h index 5177e2957..8ce5fbe72 100644 --- a/libminifi/include/FlowController.h +++ b/libminifi/include/FlowController.h @@ -106,7 +106,7 @@ class FlowController : public core::controller::ForwardingControllerServiceProvi } void executeOnComponent(const std::string& id_or_name, std::function<void(state::StateController&)> func) override; - void executeOnAllComponents(std::function<void(state::StateController&)> func) override; + std::vector<state::ComponentState> getComponentStates() const override; int16_t clearConnection(const std::string &connection) override; @@ -167,7 +167,6 @@ class FlowController : public core::controller::ForwardingControllerServiceProvi std::unique_ptr<core::ProcessGroup> loadInitialFlow(); void loadFlowRepo(); - std::vector<state::StateController*> getAllComponents(); state::StateController* getComponent(const std::string& id_or_name); gsl::not_null<std::unique_ptr<state::ProcessorController>> createController(core::Processor& processor); diff --git a/libminifi/include/RootProcessGroupWrapper.h b/libminifi/include/RootProcessGroupWrapper.h index e69585cdf..7c28985d8 100644 --- a/libminifi/include/RootProcessGroupWrapper.h +++ b/libminifi/include/RootProcessGroupWrapper.h @@ -32,6 +32,8 @@ #include "minifi-cpp/utils/gsl.h" #include "minifi-cpp/core/logging/Logger.h" #include "c2/ControllerSocketProtocol.h" +#include "core/state/UpdateController.h" +#include "core/state/ComponentStateCache.h" namespace org::apache::nifi::minifi { @@ -46,12 +48,15 @@ class RootProcessGroupWrapper { if (metrics_publisher_store_) { metrics_publisher_store_->clearMetricNodes(); } + component_cache_.setRoot(nullptr); } void updatePropertyValue(const std::string& processor_name, const std::string& property_name, const std::string& property_value); std::string getName() const; utils::Identifier getComponentUUID() const; std::string getVersion() const; + + std::vector<state::ComponentState> getProcessorStates() const; const core::ProcessGroup* getRoot() const { return root_.get(); } void setNewRoot(std::unique_ptr<core::ProcessGroup> new_root); void restoreBackup(); @@ -74,9 +79,6 @@ class RootProcessGroupWrapper { state::StateController* getProcessorController(const std::string& id_or_name, const std::function<gsl::not_null<std::unique_ptr<state::ProcessorController>>(core::Processor&)>& controllerFactory); - std::optional<std::vector<state::StateController*>> getAllProcessorControllers( - const std::function<gsl::not_null<std::unique_ptr<state::ProcessorController>>(core::Processor&)>& controllerFactory); - void setControllerSocketProtocol(c2::ControllerSocketProtocol* controller_socket_protocol) { controller_socket_protocol_ = controller_socket_protocol; } @@ -91,6 +93,8 @@ class RootProcessGroupWrapper { std::unordered_map<utils::Identifier, std::unique_ptr<state::ProcessorController>> processor_to_controller_; std::chrono::milliseconds shutdown_check_interval_{1000}; c2::ControllerSocketProtocol* controller_socket_protocol_{}; + state::ComponentStateCache component_cache_; + std::shared_ptr<core::logging::Logger> logger_ = core::logging::LoggerFactory<RootProcessGroupWrapper>::getLogger(); }; diff --git a/libminifi/include/core/state/ComponentStateCache.h b/libminifi/include/core/state/ComponentStateCache.h new file mode 100644 index 000000000..4edae4bde --- /dev/null +++ b/libminifi/include/core/state/ComponentStateCache.h @@ -0,0 +1,53 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#pragma once + +#include <shared_mutex> +#include <string> +#include <vector> + +#include "utils/Id.h" +#include "core/state/UpdateController.h" + +namespace org::apache::nifi::minifi::core { +class Processor; +class ProcessGroup; +} // namespace org::apache::nifi::minifi::core + +namespace org::apache::nifi::minifi::state { + +// Identity of a root process group, copied out of it whenever the root changes, so that it can be read from another +// thread (eg. a metrics thread) without touching the process group itself. +struct RootIdentity { + std::string name; + utils::Identifier uuid; + std::string version{"0"}; +}; + +// Thread-safe snapshot of a root process group's identity and processor list, guarded by a leaf mutex: it is never +// held while any other minifi mutex is taken, so metrics threads can read it without inverting the lock order. +struct ComponentStateCache { + void setRoot(core::ProcessGroup* root); + RootIdentity getIdentity() const; + std::vector<ComponentState> getProcessorStates() const; + + mutable std::shared_mutex mutex; + std::vector<core::Processor*> processors; + RootIdentity identity; +}; + +} // namespace org::apache::nifi::minifi::state diff --git a/libminifi/include/core/state/UpdateController.h b/libminifi/include/core/state/UpdateController.h index bed22a5b6..96d7ece99 100644 --- a/libminifi/include/core/state/UpdateController.h +++ b/libminifi/include/core/state/UpdateController.h @@ -26,6 +26,7 @@ #include "utils/BackTrace.h" #include "minifi-cpp/io/InputStream.h" #include "utils/expected.h" +#include "utils/Id.h" namespace org::apache::nifi::minifi::state { @@ -116,61 +117,39 @@ class StateController : public Pausable { [[nodiscard]] virtual std::string getComponentName() const = 0; [[nodiscard]] virtual utils::Identifier getComponentUUID() const = 0; - /** - * Start the client - */ + virtual int16_t start() = 0; - /** - * Stop the client - */ + virtual int16_t stop() = 0; virtual bool isRunning() const = 0; }; -/** - * Justification and Purpose: Update sink is an abstract class with most functions being purely virtual. - * This is meant to reflect the operational sink for the state of the client. Functions, below, represent - * a small and tight interface for operations that can be performed from external controllers. - * - */ +struct ComponentState { + std::string name; + utils::Identifier uuid; + bool running = false; +}; + class StateMonitor : public StateController { public: ~StateMonitor() override = default; - // Execute callback func on the component. Thread safe, locking mutex_, preventing concurrent flow update virtual void executeOnComponent(const std::string &id_or_name, std::function<void(state::StateController&)> func) = 0; - // Execute callback func on the all components. Thread safe, locking mutex_, preventing concurrent flow update - virtual void executeOnAllComponents(std::function<void(state::StateController&)> func) = 0; + virtual std::vector<ComponentState> getComponentStates() const = 0; - /** - * Drain repositories - */ virtual int16_t drainRepositories() = 0; - /** - * Clear connection for the agent. - */ virtual int16_t clearConnection(const std::string &connection) = 0; virtual std::vector<std::string> getSupportedConfigurationFormats() const = 0; - /** - * Apply an update with the provided string. - */ virtual std::expected<void, std::string> applyUpdate(const std::string & source, const std::string &configuration, - bool persist = false, const std::optional<std::string>& flow_id = std::nullopt) = 0; - /** - * Returns uptime for this module. - * @return uptime for the current state monitor. - */ + bool persist = false, const std::optional<std::string>& flow_id = std::nullopt) = 0; + virtual uint64_t getUptime() = 0; - /** - * Returns a vector of backtraces - * @return backtraces from the state monitor. - */ virtual std::vector<BackTrace> getTraces() = 0; virtual std::map<std::string, std::unique_ptr<io::InputStream>> getDebugInfo() = 0; diff --git a/libminifi/include/core/state/nodes/ResponseNodeLoader.h b/libminifi/include/core/state/nodes/ResponseNodeLoader.h index d4c4240af..ac503b555 100644 --- a/libminifi/include/core/state/nodes/ResponseNodeLoader.h +++ b/libminifi/include/core/state/nodes/ResponseNodeLoader.h @@ -57,6 +57,7 @@ class ResponseNodeLoaderImpl : public ResponseNodeLoader { std::vector<SharedResponseNode> getComponentMetricsNodes(const std::string& metrics_class) const; std::expected<SharedResponseNode, std::string> getSystemMetricsNode(const std::string& clazz); std::vector<SharedResponseNode> getResponseNodes(const std::string& clazz); + void initializeSystemMetricsNode(const SharedResponseNode& response_node) const; void initializeRepositoryMetrics(const SharedResponseNode& response_node) const; void initializeQueueMetrics(const SharedResponseNode& response_node) const; void initializeAgentIdentifier(const SharedResponseNode& response_node) const; @@ -71,11 +72,9 @@ class ResponseNodeLoaderImpl : public ResponseNodeLoader { mutable std::mutex root_mutex_; mutable std::mutex component_metrics_mutex_; mutable std::mutex system_metrics_mutex_; - mutable std::mutex initialization_mutex_; core::ProcessGroup* root_{}; std::unordered_map<std::string, std::vector<SharedResponseNode>> component_metrics_; std::unordered_map<std::string, SharedResponseNode> system_metrics_; - std::unordered_set<std::string> initialized_metrics_; std::shared_ptr<Configure> configuration_; std::vector<std::shared_ptr<core::RepositoryMetricsSource>> repository_metric_sources_; std::shared_ptr<core::FlowConfiguration> flow_configuration_; diff --git a/libminifi/src/FlowController.cpp b/libminifi/src/FlowController.cpp index e315ae3b5..0cb4e1b68 100644 --- a/libminifi/src/FlowController.cpp +++ b/libminifi/src/FlowController.cpp @@ -404,14 +404,12 @@ int16_t FlowController::clearConnection(const std::string &connection) { return -1; } -void FlowController::executeOnAllComponents(std::function<void(state::StateController&)> func) { - if (updating_.isUpdating()) { - return; - } - std::lock_guard<std::recursive_mutex> lock(mutex_); - for (auto* component: getAllComponents()) { - func(*component); - } +std::vector<state::ComponentState> FlowController::getComponentStates() const { + // Deliberately does not lock mutex_: this is called from metrics threads while a metrics publisher's mutex + // is held, and locking mutex_ here would invert the lock order against the flow update path. + auto states = root_wrapper_.getProcessorStates(); + states.push_back({.name = getComponentName(), .uuid = root_wrapper_.getComponentUUID(), .running = isRunning()}); + return states; } void FlowController::executeOnComponent(const std::string &id_or_name, std::function<void(state::StateController&)> func) { @@ -426,15 +424,6 @@ void FlowController::executeOnComponent(const std::string &id_or_name, std::func } } -std::vector<state::StateController*> FlowController::getAllComponents() { - if (auto components = root_wrapper_.getAllProcessorControllers([this](core::Processor& p) { return createController(p); })) { - components->push_back(this); - return *components; - } - - return {this}; -} - state::StateController* FlowController::getComponent(const std::string& id_or_name) { if (id_or_name == getUUIDStr() || id_or_name == "FlowController") { return this; diff --git a/libminifi/src/RootProcessGroupWrapper.cpp b/libminifi/src/RootProcessGroupWrapper.cpp index ca910f96e..423703ddc 100644 --- a/libminifi/src/RootProcessGroupWrapper.cpp +++ b/libminifi/src/RootProcessGroupWrapper.cpp @@ -25,24 +25,19 @@ void RootProcessGroupWrapper::updatePropertyValue(const std::string& processor_n } std::string RootProcessGroupWrapper::getName() const { - if (root_ != nullptr) { - return root_->getName(); - } - return ""; + return component_cache_.getIdentity().name; } utils::Identifier RootProcessGroupWrapper::getComponentUUID() const { - if (!root_) { - return {}; - } - return root_->getUUID(); + return component_cache_.getIdentity().uuid; } std::string RootProcessGroupWrapper::getVersion() const { - if (root_ != nullptr) { - return std::to_string(root_->getVersion()); - } - return "0"; + return component_cache_.getIdentity().version; +} + +std::vector<state::ComponentState> RootProcessGroupWrapper::getProcessorStates() const { + return component_cache_.getProcessorStates(); } void RootProcessGroupWrapper::setNewRoot(std::unique_ptr<core::ProcessGroup> new_root) { @@ -55,9 +50,11 @@ void RootProcessGroupWrapper::setNewRoot(std::unique_ptr<core::ProcessGroup> new if (controller_socket_protocol_) { controller_socket_protocol_->setRoot(nullptr); } + component_cache_.setRoot(nullptr); backup_root_ = std::move(root_); root_ = std::move(new_root); processor_to_controller_.clear(); + component_cache_.setRoot(root_.get()); if (metrics_publisher_store_) { metrics_publisher_store_->loadMetricNodes(root_.get()); } if (controller_socket_protocol_) { controller_socket_protocol_->setRoot(root_.get()); @@ -66,9 +63,17 @@ void RootProcessGroupWrapper::setNewRoot(std::unique_ptr<core::ProcessGroup> new void RootProcessGroupWrapper::restoreBackup() { if (metrics_publisher_store_) { metrics_publisher_store_->clearMetricNodes(); } + if (controller_socket_protocol_) { + controller_socket_protocol_->setRoot(nullptr); + } + component_cache_.setRoot(nullptr); root_ = std::move(backup_root_); processor_to_controller_.clear(); + component_cache_.setRoot(root_.get()); if (metrics_publisher_store_) { metrics_publisher_store_->loadMetricNodes(root_.get()); } + if (controller_socket_protocol_) { + controller_socket_protocol_->setRoot(root_.get()); + } } void RootProcessGroupWrapper::clearBackup() { @@ -140,27 +145,6 @@ void RootProcessGroupWrapper::clearConnection(const std::string &connection) { } } -std::optional<std::vector<state::StateController*>> RootProcessGroupWrapper::getAllProcessorControllers( - const std::function<gsl::not_null<std::unique_ptr<state::ProcessorController>>(core::Processor&)>& controllerFactory) { - if (!root_) { - return std::nullopt; - } - std::vector<state::StateController*> controllerVec; - std::vector<core::Processor*> processorVec; - root_->getAllProcessors(processorVec); - - for (const auto& processor : processorVec) { - // reference to the existing or newly created controller - auto& controller = processor_to_controller_[processor->getUUID()]; - if (!controller) { - controller = controllerFactory(*processor); - } - controllerVec.push_back(controller.get()); - } - - return controllerVec; -} - state::StateController* RootProcessGroupWrapper::getProcessorController(const std::string& id_or_name, const std::function<gsl::not_null<std::unique_ptr<state::ProcessorController>>(core::Processor&)>& controllerFactory) { if (!root_) { @@ -170,7 +154,8 @@ state::StateController* RootProcessGroupWrapper::getProcessorController(const st return utils::Identifier::parse(id_or_name) | utils::andThen([this](utils::Identifier id) { return utils::optional_from_ptr(root_->findProcessorById(id)); }) | utils::orElse([this, &id_or_name] { return utils::optional_from_ptr(root_->findProcessorByName(id_or_name)); }) - | utils::transform([this, &controllerFactory](gsl::not_null<core::Processor*> proc) -> gsl::not_null<state::ProcessorController*> { + | utils::transform([this, &controllerFactory](core::Processor* proc) -> gsl::not_null<state::ProcessorController*> { + gsl_Expects(proc != nullptr); return utils::optional_from_ptr(processor_to_controller_[proc->getUUID()].get()) | utils::valueOrElse([this, proc, &controllerFactory] { return gsl::make_not_null((processor_to_controller_[proc->getUUID()] = controllerFactory(*proc)).get()); diff --git a/libminifi/src/c2/ControllerSocketProtocol.cpp b/libminifi/src/c2/ControllerSocketProtocol.cpp index 9b64b6907..c7f6a78a5 100644 --- a/libminifi/src/c2/ControllerSocketProtocol.cpp +++ b/libminifi/src/c2/ControllerSocketProtocol.cpp @@ -271,17 +271,14 @@ void ControllerSocketProtocol::writeQueueSizesResponse(io::BaseStream &stream) { } void ControllerSocketProtocol::writeComponentsResponse(io::BaseStream &stream) { - std::vector<std::pair<std::string, bool>> components; - update_sink_.executeOnAllComponents([&components](state::StateController& component) { - components.emplace_back(component.getComponentName(), component.isRunning()); - }); + const auto components = update_sink_.getComponentStates(); io::BufferStream resp; auto op = static_cast<uint8_t>(Operation::describe); resp.write(&op, 1); resp.write(gsl::narrow<uint16_t>(components.size())); - for (const auto& [name, is_running] : components) { - resp.write(name); - resp.write(is_running ? "true" : "false"); + for (const auto& component : components) { + resp.write(component.name); + resp.write(component.running ? "true" : "false"); } stream.write(resp.getBuffer()); diff --git a/libminifi/src/c2/FlowStatusBuilder.cpp b/libminifi/src/c2/FlowStatusBuilder.cpp index 7512c31a1..2f3c1a815 100644 --- a/libminifi/src/c2/FlowStatusBuilder.cpp +++ b/libminifi/src/c2/FlowStatusBuilder.cpp @@ -219,6 +219,8 @@ std::expected<void, std::string> FlowStatusBuilder::addInstanceStatus(rapidjson: bulletins = bulletin_store_->getBulletins(std::chrono::minutes(5)); } + std::lock_guard<std::mutex> guard(root_mutex_); + if (options.contains(FlowStatusQueryOption::health)) { instance_status.AddMember("instanceHealth", rapidjson::Value(rapidjson::kObjectType), allocator); uint64_t queued_count = 0; diff --git a/libminifi/src/core/state/ComponentStateCache.cpp b/libminifi/src/core/state/ComponentStateCache.cpp new file mode 100644 index 000000000..602f1572e --- /dev/null +++ b/libminifi/src/core/state/ComponentStateCache.cpp @@ -0,0 +1,55 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#include "core/state/ComponentStateCache.h" + +#include <mutex> +#include <string> +#include <utility> + +#include "core/ProcessGroup.h" + +namespace org::apache::nifi::minifi::state { + +RootIdentity ComponentStateCache::getIdentity() const { + std::shared_lock lock(mutex); + return identity; +} + +std::vector<ComponentState> ComponentStateCache::getProcessorStates() const { + std::shared_lock lock(mutex); + std::vector<ComponentState> states; + states.reserve(processors.size()); + for (const auto* processor : processors) { + states.push_back({.name = processor->getName(), .uuid = processor->getUUID(), .running = processor->isRunning()}); + } + return states; +} + +void ComponentStateCache::setRoot(core::ProcessGroup* root) { + std::vector<core::Processor*> new_processors; + RootIdentity new_identity; + if (root) { + root->getAllProcessors(new_processors); + new_identity = {.name = root->getName(), .uuid = root->getUUID(), .version = std::to_string(root->getVersion())}; + } + + std::unique_lock lock(mutex); + processors = std::move(new_processors); + identity = std::move(new_identity); +} + +} // namespace org::apache::nifi::minifi::state diff --git a/libminifi/src/core/state/MetricsPublisherStore.cpp b/libminifi/src/core/state/MetricsPublisherStore.cpp index 933434cbe..879bbde7f 100644 --- a/libminifi/src/core/state/MetricsPublisherStore.cpp +++ b/libminifi/src/core/state/MetricsPublisherStore.cpp @@ -59,10 +59,11 @@ void MetricsPublisherStore::loadMetricNodes(core::ProcessGroup* root) { } void MetricsPublisherStore::clearMetricNodes() { + response_node_loader_->clearConfigRoot(); + // clearConfigRoot() must be before the clearMetricNodes() calls, otherwise a reader could create and cache a system metrics node bound to the old root for (const auto& [name, publisher]: metrics_publishers_) { publisher->clearMetricNodes(); } - response_node_loader_->clearConfigRoot(); } std::weak_ptr<state::MetricsPublisher> MetricsPublisherStore::getMetricsPublisher(const std::string& name) const { diff --git a/libminifi/src/core/state/nodes/AgentInformation.cpp b/libminifi/src/core/state/nodes/AgentInformation.cpp index f82b0a50f..458bd8843 100644 --- a/libminifi/src/core/state/nodes/AgentInformation.cpp +++ b/libminifi/src/core/state/nodes/AgentInformation.cpp @@ -243,10 +243,10 @@ std::vector<PublishedMetric> AgentStatus::calculateMetrics() { } if (nullptr != monitor_) { - monitor_->executeOnAllComponents([this, &metrics](StateController& component){ - metrics.push_back({"is_running", (component.isRunning() ? 1.0 : 0.0), - {{"component_uuid", component.getComponentUUID().to_string()}, {"component_name", component.getComponentName()}, {"metric_class", getName()}}}); - }); + for (const auto& component : monitor_->getComponentStates()) { + metrics.push_back({"is_running", (component.running ? 1.0 : 0.0), + {{"component_uuid", component.uuid.to_string()}, {"component_name", component.name}, {"metric_class", getName()}}}); + } } metrics.push_back({"agent_memory_usage_bytes", static_cast<double>(utils::OsUtils::getCurrentProcessPhysicalMemoryUsage()), {{"metric_class", getName()}}}); @@ -285,17 +285,16 @@ SerializedResponseNode AgentStatus::serializeComponents() const { components_node.collapsible = false; components_node.name = "components"; if (monitor_ != nullptr) { - monitor_->executeOnAllComponents([&components_node](StateController& component){ - SerializedResponseNode component_node { - .name = component.getComponentName(), + for (const auto& component : monitor_->getComponentStates()) { + components_node.children.push_back(SerializedResponseNode{ + .name = component.name, .collapsible = false, .children = { - {.name = "running", .value = component.isRunning()}, - {.name = "uuid", .value = std::string{component.getComponentUUID().to_string()}}, + {.name = "running", .value = component.running}, + {.name = "uuid", .value = std::string{component.uuid.to_string()}}, } - }; - components_node.children.push_back(component_node); - }); + }); + } } return components_node; } diff --git a/libminifi/src/core/state/nodes/FlowInformation.cpp b/libminifi/src/core/state/nodes/FlowInformation.cpp index 6fdb57f57..7f2febb48 100644 --- a/libminifi/src/core/state/nodes/FlowInformation.cpp +++ b/libminifi/src/core/state/nodes/FlowInformation.cpp @@ -39,9 +39,7 @@ std::vector<SerializedResponseNode> FlowInformation::serialize() { }; if (nullptr != monitor_) { - monitor_->executeOnComponent("FlowController", [&serialized](StateController& component) { - serialized.push_back({.name = "runStatus", .value = (component.isRunning() ? "RUNNING" : "STOPPED")}); - }); + serialized.push_back({.name = "runStatus", .value = (monitor_->isRunning() ? "RUNNING" : "STOPPED")}); } SerializedResponseNode uri; @@ -132,10 +130,8 @@ std::vector<SerializedResponseNode> FlowInformation::serialize() { std::vector<PublishedMetric> FlowInformation::calculateMetrics() { std::vector<PublishedMetric> metrics = connection_store_.calculateConnectionMetrics("FlowInformation"); if (nullptr != monitor_) { - monitor_->executeOnComponent("FlowController", [&metrics](StateController& component) { - metrics.push_back({"is_running", (component.isRunning() ? 1.0 : 0.0), - {{"component_uuid", component.getComponentUUID().to_string()}, {"component_name", component.getComponentName()}, {"metric_class", "FlowInformation"}}}); - }); + metrics.push_back({"is_running", (monitor_->isRunning() ? 1.0 : 0.0), + {{"component_uuid", monitor_->getComponentUUID().to_string()}, {"component_name", monitor_->getComponentName()}, {"metric_class", "FlowInformation"}}}); } for (const auto& processor : processors_) { diff --git a/libminifi/src/core/state/nodes/ResponseNodeLoader.cpp b/libminifi/src/core/state/nodes/ResponseNodeLoader.cpp index 76a73f0b5..3690201fa 100644 --- a/libminifi/src/core/state/nodes/ResponseNodeLoader.cpp +++ b/libminifi/src/core/state/nodes/ResponseNodeLoader.cpp @@ -43,10 +43,6 @@ ResponseNodeLoaderImpl::ResponseNodeLoaderImpl(std::shared_ptr<Configure> config } void ResponseNodeLoaderImpl::clearConfigRoot() { - { - std::lock_guard<std::mutex> guard(initialization_mutex_); - initialized_metrics_.clear(); - } { std::lock_guard<std::mutex> guard(system_metrics_mutex_); system_metrics_.clear(); @@ -110,10 +106,26 @@ std::expected<SharedResponseNode, std::string> ResponseNodeLoaderImpl::getSystem if (!response_node) { return std::unexpected{"Instantiated class '" + clazz + "' is not a ResponseNode!"}; } - system_metrics_.emplace(clazz, gsl::make_not_null(response_node)); + // Initialize before publishing to the cache, and while still holding system_metrics_mutex_, so that no + // other thread can ever get hold of a node whose dependencies have not been injected yet. + auto new_node = gsl::make_not_null(std::move(response_node)); + initializeSystemMetricsNode(new_node); + system_metrics_.emplace(clazz, new_node); return system_metrics_.at(clazz); } +void ResponseNodeLoaderImpl::initializeSystemMetricsNode(const SharedResponseNode& response_node) const { + initializeRepositoryMetrics(response_node); + initializeQueueMetrics(response_node); + initializeAgentIdentifier(response_node); + initializeAgentMonitor(response_node); + initializeAgentNode(response_node); + initializeAgentStatus(response_node); + initializeConfigurationChecksums(response_node); + initializeFlowInformation(response_node); + initializeAssetInformation(response_node); +} + std::vector<SharedResponseNode> ResponseNodeLoaderImpl::getResponseNodes(const std::string& clazz) { auto component_metrics = getComponentMetricsNodes(clazz); if (!component_metrics.empty()) { @@ -246,28 +258,8 @@ void ResponseNodeLoaderImpl::initializeFlowInformation(const SharedResponseNode& std::vector<SharedResponseNode> ResponseNodeLoaderImpl::loadResponseNodes(const std::string& clazz, core::ProcessGroup* root) { auto response_nodes = getResponseNodes(clazz); - if (response_nodes.empty()) { - if (root) { - logger_->log_error("No metric defined for {}", clazz); - } - return {}; - } - - std::lock_guard<std::mutex> guard(initialization_mutex_); - for (const auto& response_node : response_nodes) { - if (initialized_metrics_.contains(response_node->getName())) { - continue; - } - initializeRepositoryMetrics(response_node); - initializeQueueMetrics(response_node); - initializeAgentIdentifier(response_node); - initializeAgentMonitor(response_node); - initializeAgentNode(response_node); - initializeAgentStatus(response_node); - initializeConfigurationChecksums(response_node); - initializeFlowInformation(response_node); - initializeAssetInformation(response_node); - initialized_metrics_.insert(response_node->getName()); + if (response_nodes.empty() && root) { + logger_->log_error("No metric defined for {}", clazz); } return response_nodes; } diff --git a/libminifi/src/core/state/nodes/SupportedOperations.cpp b/libminifi/src/core/state/nodes/SupportedOperations.cpp index 302ad81d6..c82fc642d 100644 --- a/libminifi/src/core/state/nodes/SupportedOperations.cpp +++ b/libminifi/src/core/state/nodes/SupportedOperations.cpp @@ -89,9 +89,9 @@ void SupportedOperations::fillProperties(SerializedResponseNode& properties, min case minifi::c2::Operation::stop: { addProperty(properties, "c2"); if (monitor_) { - monitor_->executeOnAllComponents([&properties](StateController& component){ - addProperty(properties, component.getComponentUUID().to_string()); - }); + for (const auto& component : monitor_->getComponentStates()) { + addProperty(properties, component.uuid.to_string()); + } } break; } diff --git a/libminifi/test/flow-tests/MetricsDuringReloadTests.cpp b/libminifi/test/flow-tests/MetricsDuringReloadTests.cpp new file mode 100644 index 000000000..96398e8ba --- /dev/null +++ b/libminifi/test/flow-tests/MetricsDuringReloadTests.cpp @@ -0,0 +1,190 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#include <array> +#include <atomic> +#include <chrono> +#include <memory> +#include <ranges> +#include <string> +#include <string_view> +#include <thread> +#include <unordered_set> +#include <vector> + +#include "c2/C2MetricsPublisher.h" +#include "c2/C2Utils.h" +#include "core/state/MetricsPublisherStore.h" +#include "properties/Configuration.h" +#include "CustomProcessors.h" +#include "unit/Catch.h" +#include "unit/TestControllerWithFlow.h" + +namespace { + +constexpr std::string_view SinkProcessorName = "Sink Processor"; + +std::string createFlow(const std::string& generator_processor_name) { + return R"( +Flow Controller: + name: MiNiFi Flow +Processors: +- name: )" + generator_processor_name + R"( + id: 2438e3c8-015a-1000-79ca-83af40ec1991 + class: org.apache.nifi.processors.standard.TestFlowFileGenerator + scheduling strategy: TIMER_DRIVEN + scheduling period: 10 ms + auto-terminated relationships list: + Properties: + Batch Size: 1 +- name: )" + std::string{SinkProcessorName} + R"( + id: 2438e3c8-015a-1000-79ca-83af40ec1992 + class: org.apache.nifi.processors.standard.TestProcessor + scheduling strategy: TIMER_DRIVEN + scheduling period: 10 ms + auto-terminated relationships list: + - apple + - banana +Connections: +- name: GeneratorToSink + id: 2438e3c8-015a-1000-79ca-83af40ec1997 + source name: )" + generator_processor_name + R"( + source id: 2438e3c8-015a-1000-79ca-83af40ec1991 + source relationship name: success + destination name: )" + std::string{SinkProcessorName} + R"( + destination id: 2438e3c8-015a-1000-79ca-83af40ec1992 + max work queue size: 0 + max work queue data size: 1 MB + flowfile expiration: 60 sec +Remote Processing Groups: [] +Controller Services: [] +)"; +} + +constexpr std::string_view UnparseableFlow = "this is not a valid flow configuration"; + +constexpr std::array DescribedClasses = {"FlowInformation", "QueueMetrics", "processorMetrics/.*"}; + +std::unique_ptr<TestControllerWithFlow> createTestController(const std::string& flow) { + auto test_controller = std::make_unique<TestControllerWithFlow>(flow.c_str(), /* setup_flow = */ false); + test_controller->configuration_->set(minifi::Configuration::nifi_c2_root_classes, + "DeviceInfoNode,AgentInformation,FlowInformation,QueueMetrics"); + test_controller->setupFlow(); + return test_controller; +} + +std::unordered_set<std::string> processorNames(const std::vector<minifi::state::ComponentState>& components) { + REQUIRE_FALSE(components.empty()); + CHECK(components.back().name == "FlowController"); + return components | std::views::take(components.size() - 1) + | std::views::transform([](const auto& component) { return component.name; }) + | std::ranges::to<std::unordered_set>(); +} + +} // namespace + +TEST_CASE("Metrics can be collected while the flow is reloaded", "[MetricsDuringReload]") { + auto test_controller = createTestController(createFlow("Processor One")); + test_controller->startFlow(); + + auto* metrics_publisher_store = test_controller->metrics_publisher_store_; + REQUIRE(metrics_publisher_store != nullptr); + auto c2_metrics_publisher = std::dynamic_pointer_cast<minifi::c2::C2MetricsPublisher>( + metrics_publisher_store->getMetricsPublisher(minifi::c2::C2_METRICS_PUBLISHER).lock()); + REQUIRE(c2_metrics_publisher != nullptr); + REQUIRE_FALSE(c2_metrics_publisher->getHeartbeatNodes(true).empty()); + + std::atomic<bool> reloads_done{false}; + std::atomic<size_t> heartbeat_nodes{0}; + std::atomic<size_t> described_nodes{0}; + + // Readers: the heartbeat and the describe path, i.e. what the C2 agent's two worker threads do. A failure + // here shows up as a hang, a TSan report or an ASan report rather than a failed assertion, because REQUIRE + // is not safe to call from a non-Catch2 thread. + auto heartbeat_thread = std::thread([&] { + while (!reloads_done) { + heartbeat_nodes += c2_metrics_publisher->getHeartbeatNodes(false).size(); + std::this_thread::sleep_for(std::chrono::milliseconds{1}); + } + }); + auto describe_thread = std::thread([&] { + for (size_t i = 0; !reloads_done; ++i) { + if (const auto node = c2_metrics_publisher->getMetricsNode(DescribedClasses[i % DescribedClasses.size()])) { + described_nodes += node->serialized_nodes.size(); + } + std::this_thread::sleep_for(std::chrono::milliseconds{1}); + } + }); + + // Writer: alternate between two flows. The unparseable payload is rejected before the live flow is + // touched, which must leave the running flow and the readers alone. + size_t successful_reloads = 0; + for (int i = 0; i < 12; ++i) { + if (i % 4 == 3) { + CHECK_FALSE(test_controller->controller_->applyConfiguration("/flows/bad", std::string{UnparseableFlow}).has_value()); + } else if (test_controller->controller_->applyConfiguration( + "/flows/" + std::to_string(i), createFlow(i % 2 == 0 ? "Processor Two" : "Processor One")).has_value()) { + ++successful_reloads; + } + } + reloads_done = true; + heartbeat_thread.join(); + describe_thread.join(); + + CHECK(successful_reloads == 9); + CHECK(heartbeat_nodes > 0); + CHECK(described_nodes > 0); + + // The flow survived all that and still reports itself. + CHECK(processorNames(test_controller->controller_->getComponentStates()) + == std::unordered_set<std::string>{"Processor Two", std::string{SinkProcessorName}}); +} + +TEST_CASE("getComponentStates lists every processor and then the agent", "[MetricsDuringReload]") { + auto test_controller = createTestController(createFlow("Processor One")); + auto& flow_controller = *test_controller->controller_; + + const std::unordered_set<std::string> flow_one{"Processor One", std::string{SinkProcessorName}}; + + SECTION("before the flow is started") { + const auto components = flow_controller.getComponentStates(); + CHECK(processorNames(components) == flow_one); + CHECK_FALSE(components.back().running); + } + + SECTION("after the flow is started") { + test_controller->startFlow(); + const auto components = flow_controller.getComponentStates(); + CHECK(processorNames(components) == flow_one); + CHECK(components.back().running); + } + + const std::unordered_set<std::string> flow_two{"Processor Two", std::string{SinkProcessorName}}; + + SECTION("the processor list follows a flow update") { + test_controller->startFlow(); + REQUIRE(flow_controller.applyConfiguration("/flows/2", createFlow("Processor Two")).has_value()); + const auto components = flow_controller.getComponentStates(); + CHECK(processorNames(components) == flow_two); + } + + SECTION("a rejected flow update leaves the component list alone") { + test_controller->startFlow(); + REQUIRE_FALSE(flow_controller.applyConfiguration("/flows/bad", std::string{UnparseableFlow}).has_value()); + const auto components = flow_controller.getComponentStates(); + CHECK(processorNames(components) == flow_one); + } +} diff --git a/libminifi/test/integration/OnScheduleErrorHandlingTests.cpp b/libminifi/test/integration/OnScheduleErrorHandlingTests.cpp index ae18fcd27..926f70eac 100644 --- a/libminifi/test/integration/OnScheduleErrorHandlingTests.cpp +++ b/libminifi/test/integration/OnScheduleErrorHandlingTests.cpp @@ -84,25 +84,19 @@ class EventDriverScheduleErrorHandlingTests: public IntegrationBase { * second is the controller itself. * Added here some assertions to make it clear. In case any of these fail without changing the corresponding yml file, * that most probably means a breaking change. */ - size_t controllerVecIdx = 0; - - fc.executeOnAllComponents([&controllerVecIdx](minifi::state::StateController& component){ - if (controllerVecIdx == 1) { - REQUIRE(component.getComponentName() == "FlowController"); - } else if (controllerVecIdx == 0) { - REQUIRE(component.getComponentName() == "kamikaze"); - - auto process_controller = dynamic_cast<minifi::state::ProcessorController*>(&component); - REQUIRE(process_controller != nullptr); - - process_controller->getProcessor().setSchedulingStrategy(minifi::core::SchedulingStrategy::EVENT_DRIVEN); - } - - ++controllerVecIdx; + const auto components = fc.getComponentStates(); + REQUIRE(components.size() == 2); + REQUIRE(components[0].name == "kamikaze"); + REQUIRE(components[1].name == "FlowController"); + + bool processor_found = false; + fc.executeOnComponent("kamikaze", [&processor_found](minifi::state::StateController& component){ + auto process_controller = dynamic_cast<minifi::state::ProcessorController*>(&component); + REQUIRE(process_controller != nullptr); + process_controller->getProcessor().setSchedulingStrategy(minifi::core::SchedulingStrategy::EVENT_DRIVEN); + processor_found = true; }); - - // check controller vector size - REQUIRE(controllerVecIdx == 2); + REQUIRE(processor_found); } void runAssertions() override { diff --git a/libminifi/test/integration/StateTransactionalityTests.cpp b/libminifi/test/integration/StateTransactionalityTests.cpp index 9d96c68ef..42db2a5eb 100644 --- a/libminifi/test/integration/StateTransactionalityTests.cpp +++ b/libminifi/test/integration/StateTransactionalityTests.cpp @@ -60,26 +60,19 @@ class StatefulIntegrationTest : public IntegrationBase { * second is the controller itself. * Added here some assertions to make it clear. In case any of these fail without changing the corresponding yml file, * that most probably means a breaking change. */ - size_t controllerVecIdx = 0; + const auto components = fc.getComponentStates(); + REQUIRE(components.size() == 2); + REQUIRE(components[0].name == "statefulProcessor"); + REQUIRE(components[1].name == "FlowController"); - fc.executeOnAllComponents([this, &controllerVecIdx](minifi::state::StateController& component){ - if (controllerVecIdx == 1) { - REQUIRE(component.getComponentName() == "FlowController"); - } else if (controllerVecIdx == 0) { - REQUIRE(component.getComponentName() == "statefulProcessor"); - // set hooks - const auto processController = dynamic_cast<ProcessorController*>(&component); - REQUIRE(processController != nullptr); - stateful_processor_ = &processController->getProcessor(); - REQUIRE(stateful_processor_); - stateful_processor_.get().setHooks(on_schedule_hook_, on_trigger_hooks_); - } - - ++controllerVecIdx; + fc.executeOnComponent("statefulProcessor", [this](minifi::state::StateController& component){ + // set hooks + const auto processController = dynamic_cast<ProcessorController*>(&component); + REQUIRE(processController != nullptr); + stateful_processor_ = &processController->getProcessor(); + stateful_processor_.get().setHooks(on_schedule_hook_, on_trigger_hooks_); }); - - // check controller vector size - REQUIRE(controllerVecIdx == 2); + REQUIRE(stateful_processor_); } void runAssertions() override {
