Skip to content

Commit a1c6e5c

Browse files
lordgamezlordgamez
committed
MINIFICPP-2525 Handle errors when enabling controllers
1 parent 7b6b5de commit a1c6e5c

File tree

13 files changed

+480
-146
lines changed

13 files changed

+480
-146
lines changed

‎extensions/couchbase/controllerservices/CouchbaseClusterService.h

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -138,11 +138,11 @@ class CouchbaseClusterService : public core::controller::ControllerServiceImpl {
138138
void initialize() override;
139139

140140
void yield() override {
141-
};
141+
}
142142

143143
bool isWorkAvailable() override {
144144
return false;
145-
};
145+
}
146146

147147
bool isRunning() const override {
148148
return getState() == core::controller::ControllerServiceState::ENABLED;

‎libminifi/include/core/controller/StandardControllerServiceNode.h

Lines changed: 2 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -46,39 +46,17 @@ class StandardControllerServiceNode : public ControllerServiceNodeImpl {
4646
StandardControllerServiceNode(const StandardControllerServiceNode &other) = delete;
4747
StandardControllerServiceNode &operator=(const StandardControllerServiceNode &parent) = delete;
4848

49-
/**
50-
* Initializes the controller service node.
51-
*/
5249
void initialize() override {
5350
ControllerServiceNodeImpl::initialize();
5451
active = false;
5552
}
5653

57-
bool canEnable() override {
58-
if (!active.load()) {
59-
for (auto linked_service : linked_controller_services_) {
60-
if (!linked_service->canEnable()) {
61-
return false;
62-
}
63-
}
64-
return true;
65-
} else {
66-
return false;
67-
}
68-
}
69-
54+
bool canEnable() override;
7055
bool enable() override;
71-
72-
bool disable() override {
73-
controller_service_->setState(DISABLED);
74-
active = false;
75-
return true;
76-
}
56+
bool disable() override;
7757

7858
protected:
79-
// controller service provider.
8059
std::shared_ptr<ControllerServiceProvider> provider;
81-
8260
std::mutex mutex_;
8361

8462
private:

‎libminifi/include/core/controller/StandardControllerServiceProvider.h

Lines changed: 21 additions & 53 deletions
Original file line numberDiff line numberDiff line change
@@ -15,19 +15,15 @@
1515
* See the License for the specific language governing permissions and
1616
* limitations under the License.
1717
*/
18-
1918
#pragma once
2019

2120
#include <string>
2221
#include <utility>
23-
#include <memory>
24-
#include <vector>
25-
#include "core/ProcessGroup.h"
26-
#include "core/ClassLoader.h"
22+
#include <memory>
23+
#include <unordered_set>
24+
#include <thread>
2725
#include "core/controller/ControllerService.h"
2826
#include "ControllerServiceNodeMap.h"
29-
#include "ControllerServiceNode.h"
30-
#include "StandardControllerServiceNode.h"
3127
#include "ControllerServiceProvider.h"
3228
#include "core/logging/LoggerFactory.h"
3329

@@ -39,6 +35,7 @@ class StandardControllerServiceProvider : public ControllerServiceProviderImpl
3935
: ControllerServiceProviderImpl(std::move(services)),
4036
extension_loader_(loader),
4137
configuration_(std::move(configuration)),
38+
admin_yield_duration_(readAdministrativeYieldDuration()),
4239
logger_(logging::LoggerFactory<StandardControllerServiceProvider>::getLogger()) {
4340
}
4441

@@ -47,64 +44,35 @@ class StandardControllerServiceProvider : public ControllerServiceProviderImpl
4744

4845
StandardControllerServiceProvider& operator=(const StandardControllerServiceProvider &other) = delete;
4946
StandardControllerServiceProvider& operator=(StandardControllerServiceProvider &&other) = delete;
50-
51-
std::shared_ptr<ControllerServiceNode> createControllerService(const std::string& type, const std::string&, const std::string& id, bool) override {
52-
std::shared_ptr<ControllerService> new_controller_service = extension_loader_.instantiate<ControllerService>(type, id);
53-
54-
if (!new_controller_service) {
55-
return nullptr;
56-
}
57-
58-
std::shared_ptr<ControllerServiceNode> new_service_node = std::make_shared<StandardControllerServiceNode>(new_controller_service,
59-
sharedFromThis<ControllerServiceProvider>(), id,
60-
configuration_);
61-
62-
controller_map_->put(id, new_service_node);
63-
return new_service_node;
64-
}
65-
66-
void enableAllControllerServices() override {
67-
logger_->log_info("Enabling {} controller services", controller_map_->getAllControllerServices().size());
68-
for (const auto& service : controller_map_->getAllControllerServices()) {
69-
logger_->log_info("Enabling {}", service->getName());
70-
if (!service->canEnable()) {
71-
logger_->log_warn("Service {} cannot be enabled", service->getName());
72-
continue;
73-
}
74-
if (!service->enable()) {
75-
logger_->log_warn("Could not enable {}", service->getName());
76-
}
77-
}
78-
}
79-
80-
void disableAllControllerServices() override {
81-
logger_->log_info("Disabling {} controller services", controller_map_->getAllControllerServices().size());
82-
for (const auto& service : controller_map_->getAllControllerServices()) {
83-
logger_->log_info("Disabling {}", service->getName());
84-
if (!service->enabled()) {
85-
logger_->log_warn("Service {} is not enabled", service->getName());
86-
continue;
87-
}
88-
if (!service->disable()) {
89-
logger_->log_warn("Could not disable {}", service->getName());
90-
}
91-
}
47+
~StandardControllerServiceProvider() override {
48+
stopEnableRetryThread();
9249
}
9350

94-
void clearControllerServices() override {
95-
controller_map_->clear();
96-
}
51+
std::shared_ptr<ControllerServiceNode> createControllerService(const std::string& type, const std::string&, const std::string& id, bool) override;
52+
void enableAllControllerServices() override;
53+
void disableAllControllerServices() override;
54+
void clearControllerServices() override;
9755

9856
protected:
57+
void stopEnableRetryThread();
58+
void startEnableRetryThread();
59+
9960
bool canEdit() override {
10061
return false;
10162
}
10263

10364
ClassLoader &extension_loader_;
104-
10565
std::shared_ptr<Configure> configuration_;
10666

10767
private:
68+
std::chrono::milliseconds readAdministrativeYieldDuration() const;
69+
70+
std::thread controller_service_enable_retry_thread_;
71+
std::atomic_bool enable_retry_thread_running_{false};
72+
std::mutex enable_retry_mutex_;
73+
std::condition_variable enable_retry_condition_;
74+
std::unordered_set<std::shared_ptr<ControllerServiceNode>> controller_services_to_enable_;
75+
std::chrono::milliseconds admin_yield_duration_;
10876
std::shared_ptr<logging::Logger> logger_;
10977
};
11078

‎libminifi/src/core/FlowConfiguration.cpp

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -98,7 +98,13 @@ std::unique_ptr<core::ProcessGroup> FlowConfiguration::updateFromPayload(const s
9898
auto old_parameter_contexts = std::move(parameter_contexts_);
9999
service_provider_ = std::make_shared<core::controller::StandardControllerServiceProvider>(std::make_unique<core::controller::ControllerServiceNodeMap>(), configuration_);
100100
auto payload = getRootFromPayload(yamlConfigPayload);
101-
if (!url.empty() && payload != nullptr) {
101+
if (!payload) {
102+
service_provider_ = old_provider;
103+
parameter_contexts_ = std::move(old_parameter_contexts);
104+
return nullptr;
105+
}
106+
107+
if (!url.empty()) {
102108
std::string payload_flow_id;
103109
std::string bucket_id;
104110
auto path_split = utils::string::split(url, "/");
@@ -110,10 +116,8 @@ std::unique_ptr<core::ProcessGroup> FlowConfiguration::updateFromPayload(const s
110116
}
111117
}
112118
flow_version_->setFlowVersion(url, bucket_id, flow_id ? *flow_id : payload_flow_id);
113-
} else {
114-
service_provider_ = old_provider;
115-
parameter_contexts_ = std::move(old_parameter_contexts);
116119
}
120+
117121
return payload;
118122
}
119123

‎libminifi/src/core/controller/ControllerServiceNodeMap.cpp

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -81,6 +81,7 @@ void ControllerServiceNodeMap::clear() {
8181
node->disable();
8282
}
8383
controller_service_nodes_.clear();
84+
process_groups_.clear();
8485
}
8586

8687
std::vector<std::shared_ptr<ControllerServiceNode>> ControllerServiceNodeMap::getAllControllerServices() const {

‎libminifi/src/core/controller/StandardControllerServiceNode.cpp

Lines changed: 30 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -19,9 +19,20 @@
1919
#include "core/controller/StandardControllerServiceNode.h"
2020
#include <memory>
2121
#include <mutex>
22+
#include <algorithm>
2223

2324
namespace org::apache::nifi::minifi::core::controller {
2425

26+
bool StandardControllerServiceNode::canEnable() {
27+
if (active) {
28+
return false;
29+
}
30+
31+
return std::all_of(linked_controller_services_.begin(), linked_controller_services_.end(), [](auto linked_service) {
32+
return linked_service->canEnable();
33+
});
34+
}
35+
2536
bool StandardControllerServiceNode::enable() {
2637
logger_->log_trace("Enabling CSN {}", getName());
2738
if (active) {
@@ -47,16 +58,32 @@ bool StandardControllerServiceNode::enable() {
4758
for (const auto& service : linked_controller_services_) {
4859
services.push_back(service->getControllerServiceImplementation());
4960
if (!service->enable()) {
50-
logger_->log_debug("Linked Service '{}' could not be enabled", service->getName());
61+
logger_->log_warn("Linked Service '{}' could not be enabled", service->getName());
5162
return false;
5263
}
5364
}
54-
impl->setLinkedControllerServices(services);
55-
impl->onEnable();
65+
try {
66+
impl->setLinkedControllerServices(services);
67+
impl->onEnable();
68+
} catch(const std::exception& e) {
69+
logger_->log_warn("Service '{}' failed to enable: {}", getName(), e.what());
70+
controller_service_->setState(ENABLING);
71+
return false;
72+
}
73+
} else {
74+
logger_->log_warn("Service '{}' service implementation could not be found", controller_service_->getName());
75+
controller_service_->setState(ENABLING);
76+
return false;
5677
}
5778
active = true;
5879
controller_service_->setState(ENABLED);
5980
return true;
6081
}
6182

83+
bool StandardControllerServiceNode::disable() {
84+
controller_service_->setState(DISABLED);
85+
active = false;
86+
return true;
87+
}
88+
6289
} // namespace org::apache::nifi::minifi::core::controller

‎libminifi/src/core/controller/StandardControllerServiceProvider.cpp

Lines changed: 127 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,127 @@
1+
/**
2+
*
3+
* Licensed to the Apache Software Foundation (ASF) under one or more
4+
* contributor license agreements. See the NOTICE file distributed with
5+
* this work for additional information regarding copyright ownership.
6+
* The ASF licenses this file to You under the Apache License, Version 2.0
7+
* (the "License"); you may not use this file except in compliance with
8+
* the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
#include "core/controller/StandardControllerServiceProvider.h"
19+
20+
#include "core/controller/StandardControllerServiceNode.h"
21+
22+
using namespace std::literals::chrono_literals;
23+
24+
namespace org::apache::nifi::minifi::core::controller {
25+
26+
std::shared_ptr<ControllerServiceNode> StandardControllerServiceProvider::createControllerService(const std::string& type, const std::string&, const std::string& id, bool) {
27+
std::shared_ptr<ControllerService> new_controller_service = extension_loader_.instantiate<ControllerService>(type, id);
28+
29+
if (!new_controller_service) {
30+
return nullptr;
31+
}
32+
33+
std::shared_ptr<ControllerServiceNode> new_service_node = std::make_shared<StandardControllerServiceNode>(new_controller_service,
34+
sharedFromThis<ControllerServiceProvider>(), id,
35+
configuration_);
36+
37+
controller_map_->put(id, new_service_node);
38+
return new_service_node;
39+
}
40+
41+
void StandardControllerServiceProvider::enableAllControllerServices() {
42+
gsl_Expects(!enable_retry_thread_running_);
43+
{
44+
std::lock_guard<std::mutex> lock(enable_retry_mutex_);
45+
logger_->log_info("Enabling {} controller services", controller_map_->getAllControllerServices().size());
46+
for (const auto& service : controller_map_->getAllControllerServices()) {
47+
logger_->log_info("Enabling {}", service->getName());
48+
if (!service->canEnable()) {
49+
logger_->log_warn("Service {} cannot be enabled", service->getName());
50+
continue;
51+
}
52+
if (!service->enable()) {
53+
logger_->log_warn("Could not enable {}", service->getName());
54+
controller_services_to_enable_.insert(service);
55+
}
56+
}
57+
}
58+
startEnableRetryThread();
59+
}
60+
61+
void StandardControllerServiceProvider::disableAllControllerServices() {
62+
stopEnableRetryThread();
63+
logger_->log_info("Disabling {} controller services", controller_map_->getAllControllerServices().size());
64+
for (const auto& service : controller_map_->getAllControllerServices()) {
65+
logger_->log_info("Disabling {}", service->getName());
66+
if (!service->disable()) {
67+
logger_->log_warn("Could not disable {}", service->getName());
68+
}
69+
}
70+
}
71+
72+
void StandardControllerServiceProvider::clearControllerServices() {
73+
stopEnableRetryThread();
74+
controller_map_->clear();
75+
}
76+
77+
void StandardControllerServiceProvider::stopEnableRetryThread() {
78+
enable_retry_thread_running_ = false;
79+
enable_retry_condition_.notify_all();
80+
if (controller_service_enable_retry_thread_.joinable()) {
81+
controller_service_enable_retry_thread_.join();
82+
}
83+
}
84+
85+
void StandardControllerServiceProvider::startEnableRetryThread() {
86+
enable_retry_thread_running_ = true;
87+
controller_service_enable_retry_thread_ = std::thread([this]() {
88+
if (controller_services_to_enable_.empty()) {
89+
return;
90+
}
91+
std::unique_lock<std::mutex> lock(enable_retry_mutex_);
92+
enable_retry_condition_.wait_for(lock, admin_yield_duration_, [this]() {
93+
return !enable_retry_thread_running_;
94+
});
95+
while (enable_retry_thread_running_) {
96+
for (auto it = controller_services_to_enable_.begin(); it != controller_services_to_enable_.end();) {
97+
if ((*it)->enable()) {
98+
it = controller_services_to_enable_.erase(it);
99+
} else {
100+
++it;
101+
}
102+
}
103+
if (controller_services_to_enable_.empty()) {
104+
break;
105+
}
106+
enable_retry_condition_.wait_for(lock, admin_yield_duration_, [this]() {
107+
return !enable_retry_thread_running_;
108+
});
109+
}
110+
controller_services_to_enable_.clear();
111+
});
112+
}
113+
114+
std::chrono::milliseconds StandardControllerServiceProvider::readAdministrativeYieldDuration() const {
115+
std::chrono::milliseconds admin_yield_duration = 30s;
116+
std::string yield_value_str;
117+
118+
if (configuration_->get(Configure::nifi_administrative_yield_duration, yield_value_str)) {
119+
std::optional<core::TimePeriodValue> value = core::TimePeriodValue::fromString(yield_value_str);
120+
if (value) {
121+
admin_yield_duration = value->getMilliseconds();
122+
}
123+
}
124+
return admin_yield_duration;
125+
}
126+
127+
} // namespace org::apache::nifi::minifi::core::controller

0 commit comments

Comments
 (0)