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