Line data Source code
1 : /*
2 : * Copyright (c) 2016 Juniper Networks, Inc. All rights reserved.
3 : * Copyright Contributors to the OpenSDN project
4 : */
5 : #include "config_client_manager.h"
6 :
7 : #include <boost/assign/list_of.hpp>
8 : #include <sandesh/request_pipeline.h>
9 : #include <sstream>
10 : #include <string>
11 :
12 : #include "base/connection_info.h"
13 : #include "base/task.h"
14 : #include "base/task_trigger.h"
15 : #include "config_amqp_client.h"
16 : #include "config_db_client.h"
17 : #include "config_cassandra_client.h"
18 : #include "config_client_log.h"
19 : #include "config_client_log_types.h"
20 : #include "config_client_show_types.h"
21 : #include "config_factory.h"
22 : #include "io/event_manager.h"
23 : #include "schema/bgp_schema_types.h"
24 : #include "schema/vnc_cfg_types.h"
25 :
26 : using namespace boost::assign;
27 : using namespace std;
28 :
29 : const set<string> ConfigClientManager::skip_properties =
30 : list_of("perms2")("draft_mode_state");
31 : bool ConfigClientManager::end_of_rib_computed_;
32 :
33 4217 : int ConfigClientManager::GetNumConfigReader() {
34 : static bool init_ = false;
35 : static int num_config_readers = 0;
36 :
37 4217 : if (!init_) {
38 : // XXX To be used for testing purposes only.
39 17 : char *count_str = getenv("CONFIG_NUM_WORKERS");
40 17 : if (count_str) {
41 0 : num_config_readers = strtol(count_str, NULL, 0);
42 : } else {
43 17 : num_config_readers = kNumConfigReaderTasks;
44 : }
45 17 : init_ = true;
46 : }
47 4217 : return num_config_readers;
48 : }
49 :
50 3677 : void ConfigClientManager::SetDefaultSchedulingPolicy() {
51 : static bool config_policy_set;
52 3677 : if (config_policy_set)
53 3660 : return;
54 17 : config_policy_set = true;
55 :
56 17 : TaskScheduler *scheduler = TaskScheduler::GetInstance();
57 : // Policy for config_client::Reader Task.
58 : TaskPolicy cassadra_reader_policy = boost::assign::list_of
59 34 : (TaskExclusion(scheduler->GetTaskId("config_client::Init")))
60 17 : (TaskExclusion(scheduler->GetTaskId("config_client::DBReader")));
61 153 : for (int idx = 0; idx < ConfigClientManager::GetNumConfigReader(); ++idx) {
62 136 : cassadra_reader_policy.push_back(
63 272 : TaskExclusion(scheduler->GetTaskId("config_client::ObjectProcessor"), idx));
64 : }
65 17 : scheduler->SetPolicy(scheduler->GetTaskId("config_client::Reader"),
66 : cassadra_reader_policy);
67 :
68 : // Policy for config_client::ObjectProcessor Task.
69 : TaskPolicy cassadra_obj_process_policy = boost::assign::list_of
70 34 : (TaskExclusion(scheduler->GetTaskId("config_client::Init")));
71 153 : for (int idx = 0; idx < ConfigClientManager::GetNumConfigReader(); ++idx) {
72 136 : cassadra_obj_process_policy.push_back(
73 272 : TaskExclusion(scheduler->GetTaskId("config_client::Reader"), idx));
74 : }
75 17 : scheduler->SetPolicy(scheduler->GetTaskId("config_client::ObjectProcessor"),
76 : cassadra_obj_process_policy);
77 :
78 : // Policy for config_client::DBReader Task.
79 : TaskPolicy fq_name_reader_policy = boost::assign::list_of
80 34 : (TaskExclusion(scheduler->GetTaskId("config_client::Init")))
81 17 : (TaskExclusion(scheduler->GetTaskId("config_client::Reader")));
82 17 : scheduler->SetPolicy(scheduler->GetTaskId("config_client::DBReader"),
83 : fq_name_reader_policy);
84 :
85 : // Policy for config_client::Init process
86 : TaskPolicy cassandra_init_policy = boost::assign::list_of
87 34 : (TaskExclusion(scheduler->GetTaskId("amqp::RabbitMQReader")))
88 34 : (TaskExclusion(scheduler->GetTaskId("config_client::ObjectProcessor")))
89 34 : (TaskExclusion(scheduler->GetTaskId("config_client::DBReader")))
90 17 : (TaskExclusion(scheduler->GetTaskId("config_client::Reader")));
91 17 : scheduler->SetPolicy(scheduler->GetTaskId("config_client::Init"),
92 : cassandra_init_policy);
93 :
94 : // Policy for amqp::RabbitMQReader process
95 : TaskPolicy rabbitmq_reader_policy = boost::assign::list_of
96 34 : (TaskExclusion(scheduler->GetTaskId("config_client::Init")));
97 17 : scheduler->SetPolicy(scheduler->GetTaskId("amqp::RabbitMQReader"),
98 : rabbitmq_reader_policy);
99 :
100 : // Policy for etcd::EtcdWatcher process
101 : TaskPolicy etcd_watcher_policy = boost::assign::list_of
102 34 : (TaskExclusion(scheduler->GetTaskId("config_client::Init")));
103 17 : scheduler->SetPolicy(scheduler->GetTaskId("etcd::EtcdWatcher"),
104 : etcd_watcher_policy);
105 :
106 17 : }
107 :
108 3677 : void ConfigClientManager::SetUp(ConfigJsonParserBase *cfg_json_base) {
109 3677 : config_json_parser_.reset(cfg_json_base);
110 3677 : config_json_parser_->Init(this);
111 3677 : thread_count_ = GetNumConfigReader();
112 3677 : end_of_rib_computed_at_ = UTCTimestampUsec();
113 3677 : if (!config_options_.config_db_use_etcd) {
114 7354 : config_db_client_.reset(
115 3677 : ConfigStaticObjectFactory::CreateRef<ConfigCassandraClient>(
116 : this,
117 : evm_,
118 3677 : config_options_,
119 : thread_count_));
120 7354 : config_amqp_client_.reset(new ConfigAmqpClient(this, hostname_,
121 3677 : module_name_, config_options_));
122 : }
123 3677 : SetDefaultSchedulingPolicy();
124 :
125 : int task_id;
126 3677 : task_id = TaskScheduler::GetInstance()->GetTaskId("config_client::Init");
127 :
128 7354 : init_trigger_.reset(new
129 : TaskTrigger(boost::bind(&ConfigClientManager::InitConfigClient, this),
130 3677 : task_id, 0));
131 :
132 3677 : reinit_triggered_ = false;
133 3677 : }
134 :
135 3677 : ConfigClientManager::ConfigClientManager(EventManager *evm,
136 : ConfigJsonParserBase *cfg_json_base,
137 : std::string hostname,
138 : std::string module_name,
139 3677 : const ConfigClientOptions& config_options)
140 3677 : : evm_(evm),
141 3677 : generation_number_(0),
142 3677 : hostname_(hostname), module_name_(module_name),
143 7354 : config_options_(config_options) {
144 3677 : end_of_rib_computed_ = false;
145 3677 : SetUp(cfg_json_base);
146 3677 : }
147 :
148 3677 : ConfigClientManager::~ConfigClientManager() {
149 3677 : }
150 :
151 7738 : void ConfigClientManager::Initialize() {
152 7738 : if (init_trigger_.get() != nullptr) {
153 7738 : init_trigger_->Set();
154 : }
155 7738 : }
156 :
157 177 : ConfigDbClient *ConfigClientManager::config_db_client() const {
158 177 : return config_db_client_.get();
159 : }
160 :
161 354 : ConfigAmqpClient *ConfigClientManager::config_amqp_client() const {
162 354 : return config_amqp_client_.get();
163 : }
164 :
165 7 : bool ConfigClientManager::GetEndOfRibComputed() const {
166 7 : std::scoped_lock lock(end_of_rib_sync_mutex_);
167 7 : return end_of_rib_computed_;
168 7 : }
169 :
170 0 : uint64_t ConfigClientManager::GetEndOfRibComputedAt() const {
171 0 : std::scoped_lock lock(end_of_rib_sync_mutex_);
172 0 : return end_of_rib_computed_at_;
173 0 : }
174 :
175 44496 : void ConfigClientManager::EnqueueUUIDRequest(string oper, string obj_type,
176 : string uuid_str) {
177 44496 : config_db_client_->EnqueueUUIDRequest(oper, obj_type, uuid_str);
178 44496 : }
179 :
180 7793 : void ConfigClientManager::EndOfConfig() {
181 : {
182 : // Notify waiting caller with the result
183 7793 : std::scoped_lock lock(end_of_rib_sync_mutex_);
184 7793 : assert(!end_of_rib_computed_);
185 7793 : end_of_rib_computed_ = true;
186 7793 : cond_var_.notify_all();
187 7793 : end_of_rib_computed_at_ = UTCTimestampUsec();
188 7793 : CONFIG_CLIENT_DEBUG(ConfigClientMgrDebug,
189 : "Config Client Mgr SM: End of RIB computed and notification sent");
190 7793 : }
191 :
192 : // Once we have finished reading the complete cassandra DB, we should verify
193 : // whether all DBEntries(node/link) are as per the new generation number.
194 : // The stale entry cleanup task ensure this.
195 : // There is no need to run stale clean up during first time startup
196 7793 : if (GetGenerationNumber())
197 0 : config_json_parser()->EndOfConfig();
198 :
199 7793 : process::ConnectionState::GetInstance()->Update();
200 7793 : }
201 :
202 : // This function waits forever for bulk sync of cassandra config to finish
203 : // The condition variable is triggered even in case of "reinit". In such a case
204 : // wait is terminated and function returns.
205 : // AMQP reader task starts consuming messages only after bulk sync.
206 : // During reinit, the tight loop is broken by triggering the condition variable
207 2 : void ConfigClientManager::WaitForEndOfConfig() {
208 2 : std::unique_lock<std::mutex> lock(end_of_rib_sync_mutex_);
209 : // Wait for End of config
210 2 : while (!end_of_rib_computed_) {
211 0 : cond_var_.wait(lock);
212 0 : if (is_reinit_triggered()) break;
213 : }
214 2 : string message;
215 : message = "Config Client Mgr SM: End of RIB notification received, "
216 2 : "re init triggered" + is_reinit_triggered()?"TRUE":"FALSE";
217 2 : CONFIG_CLIENT_DEBUG(ConfigClientMgrDebug, message);
218 4 : return;
219 2 : }
220 :
221 0 : void ConfigClientManager::GetClientManagerInfo(
222 : ConfigClientManagerInfo &info) const {
223 0 : std::scoped_lock lock(end_of_rib_sync_mutex_);
224 0 : info.end_of_rib_computed = end_of_rib_computed_;
225 0 : info.end_of_rib_computed_at = end_of_rib_computed_at_;
226 0 : info.end_of_rib_computed_at = UTCUsecToString(end_of_rib_computed_at_);
227 0 : }
228 :
229 0 : void ConfigClientManager::PostShutdown() {
230 0 : config_db_client_->PostShutdown();
231 0 : reinit_triggered_ = false;
232 0 : end_of_rib_computed_ = false;
233 :
234 : // All set to read next version of the config. Increment the generation
235 0 : IncrementGenerationNumber();
236 :
237 : // scoped ptr reset deletes the previous config db object
238 : // Create new config db client and amqp client
239 : // Delete of config db client object guarantees the flusing of
240 : // object uuid cache and uuid read request list.
241 0 : if (!config_options_.config_db_use_etcd) {
242 0 : config_db_client_.reset(ConfigStaticObjectFactory::
243 0 : CreateRef<ConfigCassandraClient>(
244 : this,
245 : evm_,
246 0 : config_options_,
247 : thread_count_));
248 0 : config_amqp_client_.reset(new ConfigAmqpClient(this, hostname_,
249 0 : module_name_, config_options_));
250 : }
251 0 : stringstream ss;
252 0 : ss << GetGenerationNumber();
253 0 : CONFIG_CLIENT_DEBUG(ConfigClientMgrDebug,
254 : "Config Client Mgr SM: Post shutdown, next version of config: "
255 : + ss.str());
256 0 : }
257 :
258 7738 : bool ConfigClientManager::InitConfigClient() {
259 7738 : if (is_reinit_triggered()) {
260 : // "config_client::Init" task is mutually exclusive to
261 : // 1. FQName reader task
262 : // 2. Object UUID Table reader task
263 : // 3. AMQP reader task
264 : // 4. Object processing Work queue task
265 : // Due to this task policy, if the reinit task is running, it ensured
266 : // that above mutually exclusive tasks have finished/aborted
267 : // Perform PostShutdown to prepare for new connection
268 : // However, it is possible that these tasks have been scheduled
269 : // but yet to begin execution. For Task and WorkQueue events, their
270 : // destructor takes core of this but for TaskTrigger events, the
271 : // destructor will crash (See Bug #1786154) as it expects the task
272 : // to not be scheduled. Taking care of that case here by checking
273 : // if TaskTrigger events are scheduled (but not executing) and
274 : // return if they are so that this will be retried.
275 0 : if (config_db_client_->IsTaskTriggered()) {
276 0 : return false;
277 : }
278 0 : PostShutdown();
279 : }
280 : // Common code path for both init/reinit
281 7738 : if (!config_options_.config_db_use_etcd) {
282 7738 : CONFIG_CLIENT_DEBUG(ConfigClientMgrDebug,
283 : "Config Client Mgr SM: Start RabbitMqReader and init Database");
284 7738 : config_amqp_client_->StartRabbitMQReader();
285 : }
286 7738 : CONFIG_CLIENT_DEBUG(ConfigClientMgrDebug,
287 : "Config Client Mgr SM: Init Database");
288 7738 : config_db_client_->InitDatabase();
289 7738 : if (is_reinit_triggered()) return false;
290 7738 : return true;
291 : }
292 :
293 0 : void ConfigClientManager::ReinitConfigClient(
294 : const ConfigClientOptions &config) {
295 0 : config_options_ = config;
296 0 : ReinitConfigClient();
297 0 : }
298 :
299 0 : void ConfigClientManager::ReinitConfigClient() {
300 : {
301 : // Wake up the amqp task waiting for EOR for config reading
302 0 : std::scoped_lock lock(end_of_rib_sync_mutex_);
303 0 : cond_var_.notify_all();
304 0 : }
305 0 : reinit_triggered_ = true;
306 0 : if (init_trigger_.get() != nullptr) {
307 0 : init_trigger_->Set();
308 : }
309 0 : CONFIG_CLIENT_DEBUG(ConfigClientMgrDebug,
310 : "Config Client Mgr SM: Re init triggered!");
311 0 : }
|
