You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by fg...@apache.org on 2024/11/15 14:13:58 UTC
(nifi-minifi-cpp) 06/06: MINIFICPP-2489 Send reason for rejecting flow update
This is an automated email from the ASF dual-hosted git repository.
fgerlits pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit a8f18ee6e6491fb5ad7d52139c9557b2efade44b
Author: Adam Debreceni <ad...@apache.org>
AuthorDate: Fri Nov 8 14:13:08 2024 +0100
MINIFICPP-2489 Send reason for rejecting flow update
Signed-off-by: Ferenc Gerlits <fg...@gmail.com>
This closes #1894
---
controller/tests/ControllerTests.cpp | 10 +++-------
libminifi/include/FlowController.h | 5 ++---
libminifi/include/core/state/UpdateController.h | 14 +++----------
libminifi/src/FlowController.cpp | 24 ++++++++++++-----------
libminifi/src/c2/C2Agent.cpp | 8 ++++----
libminifi/test/flow-tests/FlowControllerTests.cpp | 2 +-
6 files changed, 26 insertions(+), 37 deletions(-)
diff --git a/controller/tests/ControllerTests.cpp b/controller/tests/ControllerTests.cpp
index 9fdca0268..08ae1a4bf 100644
--- a/controller/tests/ControllerTests.cpp
+++ b/controller/tests/ControllerTests.cpp
@@ -164,14 +164,10 @@ class TestUpdateSink : public minifi::state::StateMonitor {
return {};
}
- int16_t applyUpdate(const std::string& /*source*/, const std::string& /*configuration*/, bool /*persist*/ = false, const std::optional<std::string>& /*flow_id*/ = std::nullopt) override {
+ nonstd::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) override {
update_calls++;
- return 0;
- }
-
- int16_t applyUpdate(const std::string& /*source*/, const std::shared_ptr<minifi::state::Update>& /*updateController*/) override {
- update_calls++;
- return 0;
+ return {};
}
uint64_t getUptime() override {
diff --git a/libminifi/include/FlowController.h b/libminifi/include/FlowController.h
index c4a3402b9..8cd266360 100644
--- a/libminifi/include/FlowController.h
+++ b/libminifi/include/FlowController.h
@@ -102,7 +102,7 @@ class FlowController : public core::controller::ForwardingControllerServiceProvi
int16_t resume() override;
// Unload the current flow, clean the root process group and all its children
int16_t stop() override;
- int16_t applyUpdate(const std::string &source, const std::string &configuration, bool persist, const std::optional<std::string>& flow_id) override;
+ nonstd::expected<void, std::string> applyUpdate(const std::string &source, const std::string &configuration, bool persist, const std::optional<std::string>& flow_id) override;
int16_t drainRepositories() override {
return -1;
}
@@ -114,7 +114,6 @@ class FlowController : public core::controller::ForwardingControllerServiceProvi
std::vector<std::string> getSupportedConfigurationFormats() const override;
- int16_t applyUpdate(const std::string& /*source*/, const std::shared_ptr<state::Update>&) override { return -1; }
// Asynchronous function trigger unloading and wait for a period of time
virtual void waitUnload(const std::chrono::milliseconds time_to_wait);
void updatePropertyValue(std::string processorName, std::string propertyName, std::string propertyValue) {
@@ -125,7 +124,7 @@ class FlowController : public core::controller::ForwardingControllerServiceProvi
// first it will validate the payload with the current root node config for flowController
// like FlowController id/name is the same and new version is greater than the current version
// after that, it will apply the configuration
- bool applyConfiguration(const std::string &source, const std::string &configurePayload, const std::optional<std::string>& flow_id = std::nullopt);
+ nonstd::expected<void, std::string> applyConfiguration(const std::string &source, const std::string &configurePayload, const std::optional<std::string>& flow_id = std::nullopt);
std::string getName() const override {
return root_wrapper_.getName();
diff --git a/libminifi/include/core/state/UpdateController.h b/libminifi/include/core/state/UpdateController.h
index c63b385bc..c2c98ca5c 100644
--- a/libminifi/include/core/state/UpdateController.h
+++ b/libminifi/include/core/state/UpdateController.h
@@ -26,6 +26,7 @@
#include "utils/ThreadPool.h"
#include "utils/BackTrace.h"
#include "io/InputStream.h"
+#include "utils/expected.h"
namespace org {
namespace apache {
@@ -162,18 +163,9 @@ class StateMonitor : public StateController {
/**
* Apply an update with the provided string.
- *
- * < 0 is an error code
- * 0 is success
*/
- virtual int16_t applyUpdate(const std::string & source, const std::string &configuration, bool persist = false, const std::optional<std::string>& flow_id = std::nullopt) = 0;
-
- /**
- * Apply an update that the agent must decode. This is useful for certain operations
- * that can't be encapsulated within these definitions.
- */
- virtual int16_t applyUpdate(const std::string &source, const std::shared_ptr<Update> &updateController) = 0;
-
+ virtual nonstd::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.
diff --git a/libminifi/src/FlowController.cpp b/libminifi/src/FlowController.cpp
index 39c24e320..89d210fce 100644
--- a/libminifi/src/FlowController.cpp
+++ b/libminifi/src/FlowController.cpp
@@ -106,23 +106,23 @@ FlowController::~FlowController() {
logger_->log_trace("Destroying FlowController");
}
-bool FlowController::applyConfiguration(const std::string &source, const std::string &configurePayload, const std::optional<std::string>& flow_id) {
+nonstd::expected<void, std::string> FlowController::applyConfiguration(const std::string &source, const std::string &configurePayload, const std::optional<std::string>& flow_id) {
std::unique_ptr<core::ProcessGroup> newRoot;
try {
newRoot = updateFromPayload(source, configurePayload, flow_id);
} catch (const std::exception& ex) {
logger_->log_error("Invalid configuration payload, type: {}, what: {}", typeid(ex).name(), ex.what());
- return false;
+ return nonstd::make_unexpected(fmt::format("Invalid configuration payload, type: {}, what: {}", typeid(ex).name(), ex.what()));
} catch (...) {
logger_->log_error("Invalid configuration payload, type: {}", getCurrentExceptionTypeName());
- return false;
+ return nonstd::make_unexpected(fmt::format("Invalid configuration payload, type: {}", getCurrentExceptionTypeName()));
}
if (newRoot == nullptr)
- return false;
+ return nonstd::make_unexpected("Could not create root process group");
if (!isRunning())
- return false;
+ return nonstd::make_unexpected("FlowController is not running");
logger_->log_info("Starting to reload Flow Controller with flow control name {}, version {}", newRoot->getName(), newRoot->getVersion());
@@ -164,7 +164,10 @@ bool FlowController::applyConfiguration(const std::string &source, const std::st
}
}
- return started;
+ if (!started) {
+ return nonstd::make_unexpected("Failed to start flow");
+ }
+ return {};
}
int16_t FlowController::stop() {
@@ -382,17 +385,16 @@ std::vector<std::string> FlowController::getSupportedConfigurationFormats() cons
return flow_configuration_->getSupportedFormats();
}
-int16_t FlowController::applyUpdate(const std::string &source, const std::string &configuration, bool persist, const std::optional<std::string>& flow_id) {
- if (applyConfiguration(source, configuration, flow_id)) {
+nonstd::expected<void, std::string> FlowController::applyUpdate(const std::string &source, const std::string &configuration, bool persist, const std::optional<std::string>& flow_id) {
+ auto result = applyConfiguration(source, configuration, flow_id);
+ if (result) {
if (persist) {
const auto* process_group = root_wrapper_.getRoot();
gsl_Expects(process_group);
flow_configuration_->persist(*process_group);
}
- return 0;
- } else {
- return -1;
}
+ return result;
}
int16_t FlowController::clearConnection(const std::string &connection) {
diff --git a/libminifi/src/c2/C2Agent.cpp b/libminifi/src/c2/C2Agent.cpp
index 5e0efabd3..33f9fbd01 100644
--- a/libminifi/src/c2/C2Agent.cpp
+++ b/libminifi/src/c2/C2Agent.cpp
@@ -1043,11 +1043,11 @@ bool C2Agent::handleConfigurationUpdate(const C2ContentResponse &resp) {
return utils::string::equalsIgnoreCase(persist->second.to_string(), "true");
}();
- int16_t err = {update_sink_->applyUpdate(file_uri, configuration_str, should_persist, getFlowIdFromConfigUpdate(resp))};
- if (err != 0) {
- logger_->log_error("Flow configuration update failed with error code {}", err);
+ auto update_result = update_sink_->applyUpdate(file_uri, configuration_str, should_persist, getFlowIdFromConfigUpdate(resp));
+ if (!update_result) {
+ logger_->log_error("Flow configuration update failed: {}", update_result.error());
C2Payload response(Operation::acknowledge, state::UpdateState::SET_ERROR, resp.ident, true);
- response.setRawData("Error while applying flow. Likely missing processors");
+ response.setRawData(fmt::format("Error while applying flow: {}", update_result.error()));
enqueue_c2_response(std::move(response));
return false;
}
diff --git a/libminifi/test/flow-tests/FlowControllerTests.cpp b/libminifi/test/flow-tests/FlowControllerTests.cpp
index 633029f89..40c917a52 100644
--- a/libminifi/test/flow-tests/FlowControllerTests.cpp
+++ b/libminifi/test/flow-tests/FlowControllerTests.cpp
@@ -283,7 +283,7 @@ Controller Services: []
REQUIRE(LogTestController::getInstance().countOccurrences("Creating scheduling agent") == 3);
LogTestController::getInstance().clear();
- bool update_successful = controller.controller_->applyConfiguration("/flows/1", empty_flow);
+ bool update_successful = controller.controller_->applyConfiguration("/flows/1", empty_flow).has_value();
REQUIRE(update_successful);
REQUIRE(LogTestController::getInstance().countOccurrences("Creating scheduling agent") == 3);