Line data Source code
1 : /*
2 : * Copyright (c) 2016 Juniper Networks, Inc. All rights reserved.
3 : * Copyright Contributors to the OpenSDN project
4 : */
5 :
6 : #include "config-client-mgr/config_cassandra_client.h"
7 :
8 : #include <sandesh/request_pipeline.h>
9 :
10 : #include <boost/algorithm/string/join.hpp>
11 : #include <boost/algorithm/string/predicate.hpp>
12 : #include <boost/foreach.hpp>
13 : #include <boost/functional/hash.hpp>
14 : #include <boost/ptr_container/ptr_map.hpp>
15 : #include <boost/uuid/uuid.hpp>
16 : #include <map>
17 : #include <set>
18 : #include <string>
19 : #include <utility>
20 : #include <vector>
21 :
22 : #include "base/connection_info.h"
23 : #include "base/logging.h"
24 : #include "base/regex.h"
25 : #include "base/task.h"
26 : #include "base/task_annotations.h"
27 : #include "base/task_trigger.h"
28 : #include "config_cass2json_adapter.h"
29 : #include "io/event_manager.h"
30 : #include "database/cassandra/cql/cql_if.h"
31 : #include "config_factory.h"
32 : #include "config_client_log.h"
33 : #include "config_client_log_types.h"
34 : #include "config_client_show_types.h"
35 : #include "sandesh/common/vns_constants.h"
36 :
37 : using contrail::regex;
38 : using contrail::regex_match;
39 : using contrail::regex_search;
40 : using std::unique_ptr;
41 : using std::multimap;
42 : using std::set;
43 : using std::string;
44 :
45 : const string ConfigCassandraClient::kUuidTableName = "obj_uuid_table";
46 : const string ConfigCassandraClient::kFqnTableName = "obj_fq_name_table";
47 : const string ConfigCassandraClient::kCassClientTaskId = "config_client::Reader";
48 : const string ConfigCassandraClient::kObjectProcessTaskId =
49 : "config_client::ObjectProcessor";
50 :
51 3677 : ConfigCassandraClient::ConfigCassandraClient(ConfigClientManager *mgr,
52 : EventManager *evm, const ConfigClientOptions &options,
53 3677 : int num_workers)
54 3677 : : ConfigDbClient(mgr, evm, options), num_workers_(num_workers) {
55 3677 : dbif_.reset(ConfigStaticObjectFactory::CreateRef<cass::cql::CqlIf>(
56 : evm, config_db_ips(),
57 3677 : GetFirstConfigDbPort(), config_db_user(),
58 3677 : config_db_password(),
59 3677 : static_cast<bool>(options.config_db_use_ssl), options.config_db_ca_certs));
60 :
61 : // Initialized the casssadra connection status;
62 3677 : InitConnectionInfo();
63 3677 : bulk_sync_status_ = 0;
64 :
65 33093 : for (int i = 0; i < num_workers_; i++) {
66 58832 : partitions_.push_back(
67 : ConfigStaticObjectFactory::Create<ConfigCassandraPartition>
68 29416 : (this, static_cast<size_t>(i)));
69 : }
70 :
71 7354 : fq_name_reader_.reset(new
72 : TaskTrigger(boost::bind(&ConfigCassandraClient::FQNameReader, this),
73 7354 : TaskScheduler::GetInstance()->GetTaskId("config_client::DBReader"),
74 3677 : 0));
75 3677 : }
76 :
77 3679 : ConfigCassandraClient::~ConfigCassandraClient() {
78 3677 : if (dbif_) {
79 : // dbif_->Db_Uninit(....);
80 : }
81 :
82 3677 : STLDeleteValues(&partitions_);
83 3679 : }
84 :
85 7738 : void ConfigCassandraClient::InitDatabase() {
86 7738 : HandleCassandraConnectionStatus(false, true);
87 : while (true) {
88 19002 : CONFIG_CLIENT_DEBUG(ConfigClientMgrDebug, "Cassandra SM: Db Init");
89 19002 : if (!dbif_->Db_Init()) {
90 2816 : CONFIG_CLIENT_DEBUG(ConfigCassInitErrorMessage,
91 : "Database initialization failed");
92 2816 : if (!InitRetry()) return;
93 2816 : continue;
94 : }
95 16186 : CONFIG_CLIENT_DEBUG(ConfigClientMgrDebug,
96 : "Cassandra SM: Db SetTableSpace");
97 16186 : if (!dbif_->Db_SetTablespace(
98 : g_vns_constants.API_SERVER_KEYSPACE_NAME)) {
99 2816 : CONFIG_CLIENT_DEBUG(ConfigCassInitErrorMessage,
100 : "Setting database keyspace failed");
101 2816 : if (!InitRetry()) return;
102 2816 : continue;
103 : }
104 13370 : CONFIG_CLIENT_DEBUG(ConfigClientMgrDebug,
105 : "Cassandra SM: Db UseColumnFamily uuidTableName");
106 13370 : if (!dbif_->Db_UseColumnfamily(kUuidTableName)) {
107 2816 : if (!InitRetry()) return;
108 2816 : continue;
109 : }
110 10554 : CONFIG_CLIENT_DEBUG(ConfigClientMgrDebug,
111 : "Cassandra SM: Db UseColumnFamily fqnTableName");
112 10554 : if (!dbif_->Db_UseColumnfamily(kFqnTableName)) {
113 2816 : if (!InitRetry()) return;
114 2816 : continue;
115 : }
116 7738 : break;
117 : }
118 7738 : HandleCassandraConnectionStatus(true);
119 7738 : BulkDataSync();
120 : }
121 :
122 11264 : bool ConfigCassandraClient::InitRetry() {
123 11264 : CONFIG_CLIENT_DEBUG(ConfigClientMgrDebug, "Cassandra SM: DB uninit");
124 11264 : dbif_->Db_Uninit();
125 : // If reinit is triggered, return false to abort connection attempt
126 11264 : if (mgr()->is_reinit_triggered()) return false;
127 11264 : usleep(GetInitRetryTimeUSec());
128 11264 : return true;
129 : }
130 :
131 : ConfigCassandraPartition *
132 68061 : ConfigCassandraClient::GetPartition(const string &uuid) {
133 68061 : int worker_id = HashUUID(uuid);
134 68061 : return partitions_[worker_id];
135 : }
136 :
137 : const ConfigCassandraPartition *
138 0 : ConfigCassandraClient::GetPartition(const string &uuid) const {
139 0 : int worker_id = HashUUID(uuid);
140 0 : return partitions_[worker_id];
141 : }
142 :
143 : const ConfigCassandraPartition *
144 208 : ConfigCassandraClient::GetPartition(int worker_id) const {
145 208 : assert(worker_id < num_workers_);
146 208 : return partitions_[worker_id];
147 : }
148 :
149 73283 : int ConfigCassandraClient::HashUUID(const string &uuid_str) const {
150 : boost::hash<string> string_hash;
151 73283 : return string_hash(uuid_str) % num_workers_;
152 : }
153 :
154 20764 : bool ConfigCassandraPartition::ReadObjUUIDTable(const set<string> &req_list) {
155 20764 : GenDb::ColListVec col_list_vec;
156 :
157 20764 : set<string> uuid_list = req_list;
158 20764 : vector<GenDb::DbDataValueVec> keys;
159 20764 : for (set<string>::const_iterator it = uuid_list.begin();
160 52384 : it != uuid_list.end(); it++) {
161 31619 : GenDb::DbDataValueVec key;
162 63236 : key.push_back(GenDb::Blob(reinterpret_cast<const uint8_t *>
163 31618 : (it->c_str()), it->size()));
164 31619 : keys.push_back(key);
165 31620 : }
166 :
167 20764 : GenDb::Blob col_filter(reinterpret_cast<const uint8_t *>("d"), 1);
168 20764 : GenDb::ColumnNameRange crange;
169 : crange.start_ =
170 20764 : boost::assign::list_of(GenDb::DbDataValue(col_filter)).convert_to_container<GenDb::DbDataValueVec>();
171 :
172 20763 : GenDb::FieldNamesToReadVec field_vec;
173 20763 : field_vec.push_back(boost::make_tuple("key", true, false, false));
174 20764 : field_vec.push_back(boost::make_tuple("column1", false, true, false));
175 20764 : field_vec.push_back(boost::make_tuple("value", false, false, true));
176 :
177 20764 : if (client()->dbif_->Db_GetMultiRow(&col_list_vec,
178 : ConfigCassandraClient::kUuidTableName, keys,
179 : crange, field_vec,
180 : GenDb::DbConsistency::QUORUM)) {
181 20764 : client()->HandleCassandraConnectionStatus(true);
182 84004 : BOOST_FOREACH(const GenDb::ColList &col_list, col_list_vec) {
183 31620 : assert(col_list.rowkey_.size() == 1);
184 31620 : assert(col_list.rowkey_[0].which() == GenDb::DB_VALUE_BLOB);
185 31620 : if (col_list.columns_.size()) {
186 31620 : GenDb::Blob uuid(boost::get<GenDb::Blob>(col_list.rowkey_[0]));
187 31620 : string uuid_str(reinterpret_cast<const char *>(uuid.data()),
188 63240 : uuid.size());
189 31620 : ProcessObjUUIDTableEntry(uuid_str, col_list);
190 31620 : uuid_list.erase(uuid_str);
191 31620 : }
192 : }
193 : } else {
194 : // Failure is returned due to connectivity issue or consistency
195 : // issues in reading from cassandra
196 0 : client()->HandleCassandraConnectionStatus(false);
197 0 : CONFIG_CLIENT_WARN(ConfigClientGetRowError,
198 : "GetMultiRow failed for table",
199 : ConfigCassandraClient::kUuidTableName, "");
200 : //
201 : // Task is rescheduled to read the request queue
202 : // Due to a bug CQL driver from datastax, connection status is
203 : // not notified asynchronously. Because of this, polling is the only
204 : // choice to determine the cql connection status.
205 : // Since there are dedicated threads to read config,
206 : // and it is ok to retry by rescheduling the reader task
207 : // TODO: Sleep or No Sleep?
208 : //
209 0 : return false;
210 : }
211 :
212 : // Delete all stale entries from the data base.
213 20763 : BOOST_FOREACH(string uuid_key, uuid_list) {
214 0 : CONFIG_CLIENT_WARN(ConfigClientGetRowError, "Missing row in the table",
215 : ConfigCassandraClient::kUuidTableName, uuid_key);
216 0 : HandleObjectDelete(uuid_key, false);
217 0 : }
218 :
219 : // Clear the uuid list.
220 20764 : uuid_list.clear();
221 20764 : return true;
222 20764 : }
223 :
224 : // Notes on list map property processing:
225 : // Separate entries per list/map keys are stored in the ObjUuidCache partition
226 : // based on the uuid.
227 : // The cache map entries contain a refreshed bit and timestamp in addition to
228 : // the values etc.
229 : // A set containing list/map property names (updated_list_map_properties) that
230 : // have key/value pairs with a new timestamp, a second set
231 : // (candidate_list_map_properties) also containing list/map property names that
232 : // may require an update given some key/value pairs have been deleted, and a
233 : // multimap (list_map_properties) for all the list/map key value pairs in the
234 : // new configuration, are build and held in the context(temporary).
235 : // These lists are used to determine which list/map properties need to be pushed
236 : // to the backend, they are built as columns are parsed. Once all columns are
237 : // parsed,
238 : // in ListMapPropReviseUpdateList, for each property name in
239 : // candidate_list_map_properties we check it is already in the
240 : // updated_list_map_property list, if not we proceed to find at least one stale
241 : // list/map key value pair with the property name in the ObJUuidCache, if one is
242 : // found that requires an update, the property name is added to
243 : // updated_list_map_properties.
244 : // Once updated_list_map_properties is revised, we iterate through each property
245 : // name in it and push all matching key/value pairs in list_map_properties.
246 : // ConfigCass2JsonAdapter groups the key value pairs belonging to the same
247 : // property so that a single DB request is sent to the
248 : // backend.
249 : // Deletes are handled by FormDeleteRequestList. Note that deletes are sent
250 : // only when all key/value pairs for a given list/map property are removed.
251 : // Additionally, the resulting DB request only resets the property_set bit, it
252 : // does not clear the entries in the backend.
253 : //
254 : // parent_or_ref_fq_name_unknown indicates that at least one parent or
255 : // ref cannot be found in the FQNameCache, this can happen if the parent or
256 : // referred object is not yet read.
257 : struct ConfigCassandraParseContext {
258 34231 : ConfigCassandraParseContext() : obj_type(""), fq_name_present(false),
259 68462 : ignore_object(false), parent_or_ref_fq_name_unknown(false) {
260 34231 : }
261 : std::multimap<string, JsonAdapterDataType> list_map_properties;
262 : set<string> updated_list_map_properties;
263 : set<string> candidate_list_map_properties;
264 : string obj_type;
265 : string fq_name;
266 : bool fq_name_present;
267 : bool ignore_object;
268 : bool parent_or_ref_fq_name_unknown;
269 :
270 : private:
271 : DISALLOW_COPY_AND_ASSIGN(ConfigCassandraParseContext);
272 : };
273 :
274 34231 : bool ConfigCassandraPartition::ProcessObjUUIDTableEntry(const string &uuid_key,
275 : const GenDb::ColList &col_list) {
276 34231 : CassColumnKVVec cass_data_vec;
277 :
278 34231 : ConfigCassandraParseContext context;
279 :
280 34231 : ConfigCassandraPartition::ObjCacheEntry *obj = MarkCacheDirty(uuid_key);
281 :
282 34231 : ParseObjUUIDTableEntry(uuid_key, col_list, &cass_data_vec, context);
283 : // Ignore draft objects.
284 34231 : if (context.ignore_object) {
285 4 : client()->PurgeFQNameCache(uuid_key);
286 4 : DeleteCacheMap(uuid_key);
287 4 : return false;
288 : }
289 : // If type or fq-name is not present in the db object, ignore the object
290 : // and trigger delete of the object.
291 34227 : if (context.obj_type.empty() || !context.fq_name_present) {
292 : // Handle as delete
293 20 : CONFIG_CLIENT_WARN(ConfigClientGetRowError,
294 : "Parsing row response for type/fq_name failed for table",
295 : ConfigCassandraClient::kUuidTableName, uuid_key);
296 20 : obj->DisableCassandraReadRetry(uuid_key);
297 20 : HandleObjectDelete(uuid_key, false);
298 20 : return false;
299 : }
300 :
301 34207 : obj->SetFQName(context.fq_name);
302 34207 : obj->SetObjType(context.obj_type);
303 :
304 34206 : if (context.parent_or_ref_fq_name_unknown) {
305 10 : obj->EnableCassandraReadRetry(uuid_key);
306 : } else {
307 34196 : obj->DisableCassandraReadRetry(uuid_key);
308 : }
309 :
310 34207 : ListMapPropReviseUpdateList(uuid_key, context);
311 :
312 : // Read the context for map and list properties
313 34207 : if (context.updated_list_map_properties.size()) {
314 188 : for (set<string>::iterator it =
315 188 : context.updated_list_map_properties.begin();
316 400 : it != context.updated_list_map_properties.end(); it++) {
317 : pair<multimap<string, JsonAdapterDataType>::iterator,
318 : multimap<string, JsonAdapterDataType>::iterator> ret =
319 212 : context.list_map_properties.equal_range(*it);
320 212 : for (multimap<string, JsonAdapterDataType>::iterator mit =
321 640 : ret.first; mit != ret.second; mit++) {
322 428 : cass_data_vec.push_back(mit->second);
323 : }
324 : }
325 : }
326 34207 : GenerateAndPushJson(uuid_key, context.obj_type, cass_data_vec, true);
327 34207 : HandleObjectDelete(uuid_key, true);
328 34207 : return true;
329 34231 : }
330 :
331 31620 : void ConfigCassandraPartition::ParseObjUUIDTableEntry(const string &uuid,
332 : const GenDb::ColList &col_list, CassColumnKVVec *cass_data_vec,
333 : ConfigCassandraParseContext &context) {
334 284566 : BOOST_FOREACH(const GenDb::NewCol &ncol, col_list.columns_) {
335 126468 : assert(ncol.name->size() == 1);
336 126468 : assert(ncol.value->size() == 1);
337 126465 : assert(ncol.timestamp->size() == 1);
338 :
339 126465 : const GenDb::DbDataValue &dname(ncol.name->at(0));
340 126464 : assert(dname.which() == GenDb::DB_VALUE_BLOB);
341 126463 : GenDb::Blob dname_blob(boost::get<GenDb::Blob>(dname));
342 126464 : string key(reinterpret_cast<const char *>(dname_blob.data()),
343 252925 : dname_blob.size());
344 :
345 126467 : const GenDb::DbDataValue &dvalue(ncol.value->at(0));
346 126465 : assert(dvalue.which() == GenDb::DB_VALUE_STRING);
347 126464 : string value(boost::get<string>(dvalue));
348 :
349 126472 : const GenDb::DbDataValue &dtimestamp(ncol.timestamp->at(0));
350 126471 : assert(dtimestamp.which() == GenDb::DB_VALUE_UINT64);
351 126469 : uint64_t timestamp = boost::get<uint64_t>(dtimestamp);
352 126467 : ParseObjUUIDTableEachColumnBuildContext(uuid, key, value, timestamp,
353 : cass_data_vec, context);
354 126478 : }
355 31620 : }
356 :
357 142604 : void ConfigCassandraPartition::ParseObjUUIDTableEachColumnBuildContext(
358 : const string &uuid, const string &key, const string &value,
359 : uint64_t timestamp, CassColumnKVVec *cass_data_vec,
360 : ConfigCassandraParseContext &context) {
361 : // Check whether there was an update to property of ref
362 142604 : JsonAdapterDataType adapter(key, value);
363 142616 : if (StoreKeyIfUpdated(uuid, &adapter, timestamp, context)) {
364 : // Field is updated.. enqueue to parsing
365 141474 : cass_data_vec->push_back(adapter);
366 : }
367 142617 : }
368 :
369 2752 : void ConfigCassandraPartition::GenerateAndPushJson(
370 : const string &uuid_key, const string &obj_type,
371 : const CassColumnKVVec &cass_data_vec, bool add_change) {
372 :
373 : ConfigCass2JsonAdapter ccja(uuid_key, client(), obj_type,
374 2752 : cass_data_vec);
375 2752 : client()->mgr()->config_json_parser()->Receive(ccja, add_change);
376 2752 : }
377 :
378 : // Post shutdown during reinit, cleanup all previous states and connections
379 : // 1. Disconnect from cassandra cluster
380 : // 2. Clean FQ Name cache
381 : // 3. Delete partitions which inturn will clear up the object cache and
382 : // previously enqueued uuid read requests
383 0 : void ConfigCassandraClient::PostShutdown() {
384 0 : CONFIG_CLIENT_DEBUG(ConfigClientMgrDebug,
385 : "Cassandra SM: Post shutdown during re init");
386 0 : CONFIG_CLIENT_DEBUG(ConfigClientMgrDebug, "Cassandra SM: Db Uninit");
387 0 : dbif_->Db_Uninit();
388 0 : STLDeleteValues(&partitions_);
389 0 : ClearFQNameCache();
390 0 : }
391 :
392 7738 : bool ConfigCassandraClient::BulkDataSync() {
393 7738 : CONFIG_CLIENT_DEBUG(
394 : ConfigClientMgrDebug, "Cassandra SM: BulkDataSync Started");
395 7738 : bulk_sync_status_ = num_workers_;
396 7738 : fq_name_reader_->Set();
397 7738 : return true;
398 : }
399 :
400 0 : bool ConfigCassandraClient::IsTaskTriggered() const {
401 : // If FQNameReader task has been triggered return true.
402 0 : if (fq_name_reader_->IsSet()) {
403 0 : return true;
404 : }
405 :
406 : /**
407 : * Walk the partitions and check if ConfigReader task has
408 : * been triggered in any of them. If so, return true.
409 : */
410 0 : BOOST_FOREACH(ConfigCassandraPartition *partition, partitions_) {
411 0 : if (partition->IsTaskTriggered()) {
412 0 : return true;
413 : }
414 : }
415 0 : return false;
416 : }
417 :
418 7738 : bool ConfigCassandraClient::FQNameReader() {
419 7738 : for (ConfigClientManager::ObjectTypeList::const_iterator it =
420 7738 : mgr()->config_json_parser()->ObjectTypeListToRead().begin();
421 9844 : it != mgr()->config_json_parser()->ObjectTypeListToRead().end();
422 2106 : it++) {
423 2106 : string column_name;
424 : while (true) {
425 : // Ensure that FQName reader task aborts on reinit trigger.
426 6318 : if (mgr()->is_reinit_triggered()) {
427 0 : CONFIG_CLIENT_DEBUG(ConfigClientMgrDebug,
428 : "Cassandra SM: Abort FQName reader on reinit trigger");
429 0 : return true;
430 : }
431 :
432 : // Rowkey is obj-type
433 6318 : GenDb::DbDataValueVec key;
434 12636 : key.push_back(GenDb::Blob(reinterpret_cast<const uint8_t *>
435 6318 : (it->c_str()), it->size()));
436 6318 : GenDb::ColumnNameRange crange;
437 6318 : if (!column_name.empty()) {
438 : GenDb::Blob col_filter(reinterpret_cast<const uint8_t *>
439 4212 : (column_name.c_str()), column_name.size());
440 : // Start reading the next set of entries from where we ended in
441 : // last read
442 : crange.start_ =
443 4212 : boost::assign::list_of(
444 8424 : GenDb::DbDataValue(col_filter)).convert_to_container
445 4212 : <GenDb::DbDataValueVec>();
446 4212 : crange.start_op_ = GenDb::Op::GT;
447 4212 : }
448 :
449 : // In large scale scenarios, each object type may have a large
450 : // number of uuid entries. Read a fixed number of entries at a time
451 : // to avoid cpu hogging by this thread.
452 6318 : crange.count_ = GetFQNameEntriesToRead();
453 :
454 6318 : GenDb::FieldNamesToReadVec field_vec;
455 6318 : field_vec.push_back(boost::make_tuple("key", true, false, false));
456 6318 : field_vec.push_back(boost::make_tuple("column1", false, true,
457 6318 : false));
458 :
459 6318 : GenDb::ColList col_list;
460 6318 : if (dbif_->Db_GetRow(&col_list, kFqnTableName, key,
461 : GenDb::DbConsistency::QUORUM, crange, field_vec)) {
462 6318 : HandleCassandraConnectionStatus(true);
463 :
464 : // No entries for this obj-type
465 6318 : if (!col_list.columns_.size())
466 2106 : break;
467 :
468 6318 : ObjTypeUUIDList uuid_list;
469 6318 : ParseFQNameRowGetUUIDList(*it, col_list, uuid_list,
470 : &column_name);
471 6318 : EnqueueDBSyncRequest(uuid_list);
472 :
473 : // If we read less than what we sought, it means there are
474 : // no more entries for current obj-type. We move to next
475 : // obj-type.
476 6318 : if (col_list.columns_.size() < GetFQNameEntriesToRead())
477 2106 : break;
478 6318 : } else {
479 0 : HandleCassandraConnectionStatus(false);
480 0 : CONFIG_CLIENT_WARN(ConfigClientGetRowError,
481 : "GetRow failed for table", kFqnTableName, *it);
482 0 : usleep(GetInitRetryTimeUSec());
483 : }
484 16848 : }
485 2106 : }
486 : // At the end of task trigger
487 131546 : BOOST_FOREACH(ConfigCassandraPartition *partition, partitions_) {
488 61904 : ObjectProcessReq *req = new ObjectProcessReq("EndOfConfig", "", "");
489 61904 : partition->Enqueue(req);
490 : }
491 :
492 7738 : return true;
493 : }
494 :
495 6318 : bool ConfigCassandraClient::ParseFQNameRowGetUUIDList(const string &obj_type,
496 : const GenDb::ColList &col_list, ObjTypeUUIDList &uuid_list,
497 : string *last_column) {
498 6318 : string column_name;
499 48438 : BOOST_FOREACH(const GenDb::NewCol &ncol, col_list.columns_) {
500 21060 : assert(ncol.name->size() == 1);
501 21060 : const GenDb::DbDataValue &dname(ncol.name->at(0));
502 21060 : assert(dname.which() == GenDb::DB_VALUE_BLOB);
503 21060 : GenDb::Blob dname_blob(boost::get<GenDb::Blob>(dname));
504 42120 : column_name = string(reinterpret_cast<const char *>(dname_blob.data()),
505 21060 : dname_blob.size());
506 21060 : UpdateFQNameCache(column_name, obj_type, uuid_list);
507 21060 : }
508 :
509 6318 : *last_column = column_name;
510 6318 : return true;
511 6318 : }
512 :
513 23244 : void ConfigCassandraClient::UpdateFQNameCache(const string &key,
514 : const string &obj_type, ObjTypeUUIDList &uuid_list) {
515 23244 : string uuid_str = FetchUUIDFromFQNameEntry(key);
516 23244 : if (uuid_str.empty())
517 0 : return;
518 23244 : uuid_list.push_back(make_pair(obj_type, uuid_str));
519 23244 : AddFQNameCache(uuid_str, obj_type, key.substr(0, key.rfind(':')));
520 23244 : }
521 :
522 21060 : string ConfigCassandraClient::FetchUUIDFromFQNameEntry(
523 : const string &key) const {
524 21060 : size_t temp = key.rfind(':');
525 21060 : return (temp == string::npos) ? "" : key.substr(temp+1);
526 : }
527 :
528 6358 : bool ConfigCassandraClient::EnqueueDBSyncRequest(
529 : const ObjTypeUUIDList &uuid_list) {
530 6358 : for (ObjTypeUUIDList::const_iterator it = uuid_list.begin();
531 29602 : it != uuid_list.end(); it++) {
532 23244 : EnqueueUUIDRequest("CREATE", it->first, it->second);
533 : }
534 6358 : return true;
535 : }
536 :
537 208 : bool ConfigCassandraClient::UUIDToObjCacheShow(
538 : const string &search_string, int inst_num, const string &last_uuid,
539 : uint32_t num_entries, vector<ConfigDBUUIDCacheEntry> *entries) const {
540 208 : return GetPartition(inst_num)->UUIDToObjCacheShow(search_string, last_uuid,
541 208 : num_entries, entries);
542 : }
543 :
544 67740 : void ConfigCassandraClient::EnqueueUUIDRequest(string oper, string obj_type,
545 : string uuid_str) {
546 67740 : ObjectProcessReq *req = new ObjectProcessReq(oper, uuid_str, obj_type);
547 67740 : GetPartition(uuid_str)->Enqueue(req);
548 67740 : }
549 :
550 61894 : void ConfigCassandraClient::BulkSyncDone() {
551 : long num_config_readers_still_processing =
552 61894 : bulk_sync_status_.fetch_sub(1);
553 61894 : if (num_config_readers_still_processing == 1) {
554 7738 : CONFIG_CLIENT_DEBUG(ConfigClientMgrDebug,
555 : "Cassandra SM: BulkSyncDone by all readers");
556 7738 : mgr()->EndOfConfig();
557 : } else {
558 54156 : CONFIG_CLIENT_DEBUG(ConfigClientMgrDebug,
559 : "Cassandra SM: One reader finished BulkSync");
560 : }
561 61904 : }
562 :
563 42558 : void ConfigCassandraClient::HandleCassandraConnectionStatus(bool success,
564 : bool force_update) {
565 42558 : UpdateConnectionInfo(success, force_update);
566 :
567 42558 : if (success) {
568 : // Update connection info
569 104460 : process::ConnectionState::GetInstance()->Update(
570 : process::ConnectionType::DATABASE, "Cassandra",
571 : process::ConnectionStatus::UP,
572 69640 : dbif_->Db_GetEndpoints(), "Established Cassandra connection");
573 34820 : CONFIG_CLIENT_DEBUG(ConfigClientMgrDebug,
574 : "Cassandra SM: Established Cassandra connection");
575 : } else {
576 23214 : process::ConnectionState::GetInstance()->Update(
577 : process::ConnectionType::DATABASE, "Cassandra",
578 : process::ConnectionStatus::DOWN,
579 15476 : dbif_->Db_GetEndpoints(), "Lost Cassandra connection");
580 7738 : CONFIG_CLIENT_DEBUG(ConfigClientMgrDebug,
581 : "Cassandra SM: Lost Cassandra connection");
582 : }
583 42558 : }
584 :
585 190 : bool ConfigCassandraClient::IsListOrMapPropEmpty(const string &uuid_key,
586 : const string &lookup_key) {
587 190 : return GetPartition(uuid_key)->IsListOrMapPropEmpty(uuid_key, lookup_key);
588 : }
589 :
590 29416 : ConfigCassandraPartition::ConfigCassandraPartition(
591 29416 : ConfigCassandraClient *client, size_t idx)
592 29416 : : config_client_(client), worker_id_(idx) {
593 29416 : int task_id = TaskScheduler::GetInstance()->GetTaskId("config_client::Reader");
594 58832 : config_reader_.reset(new
595 : TaskTrigger(boost::bind(&ConfigCassandraPartition::ConfigReader, this),
596 29416 : task_id, idx));
597 : task_id =
598 29416 : TaskScheduler::GetInstance()->GetTaskId("config_client::ObjectProcessor");
599 58832 : obj_process_queue_.reset(new WorkQueue<ObjectProcessReq *>(
600 : task_id, idx, bind(&ConfigCassandraPartition::RequestHandler, this, _1),
601 29416 : WorkQueue<ObjectProcessReq *>::kMaxSize, 512));
602 29416 : }
603 :
604 40808 : ConfigCassandraPartition::~ConfigCassandraPartition() {
605 29416 : obj_process_queue_->Shutdown();
606 40808 : }
607 :
608 129644 : void ConfigCassandraPartition::Enqueue(ObjectProcessReq *req) {
609 129644 : obj_process_queue_->Enqueue(req);
610 129644 : }
611 :
612 129558 : bool ConfigCassandraPartition::RequestHandler(ObjectProcessReq *req) {
613 129558 : AddUUIDToRequestList(req->oper_, req->value_, req->uuid_str_);
614 129636 : delete req;
615 129639 : return true;
616 : }
617 :
618 129582 : void ConfigCassandraPartition::AddUUIDToRequestList(const string &oper,
619 : const string &obj_type,
620 : const string &uuid_str) {
621 129582 : pair<UUIDProcessSet::iterator, bool> ret;
622 129590 : bool trigger = uuid_read_set_.empty();
623 : ObjectProcessRequestType *req =
624 129588 : new ObjectProcessRequestType(oper, obj_type, uuid_str);
625 129642 : ret = uuid_read_set_.insert(make_pair(client()->GetUUID(uuid_str), req));
626 129632 : if (ret.second) {
627 127823 : if (trigger) {
628 95580 : config_reader_->Set();
629 : }
630 : } else {
631 1809 : delete req;
632 1808 : ret.first->second->oper = oper;
633 1807 : ret.first->second->uuid = uuid_str;
634 : }
635 129637 : }
636 :
637 65925 : void ConfigCassandraPartition::HandleObjectDelete(
638 : const string &uuid, bool add_change) {
639 65925 : if (!add_change) {
640 : ConfigCassandraClient::ObjTypeFQNPair obj_type_fq_name_pair =
641 31718 : client()->UUIDToFQName(uuid, true);
642 31719 : if (obj_type_fq_name_pair.second == "ERROR") {
643 10529 : return;
644 : }
645 31719 : }
646 :
647 55397 : bool needNotify = false;
648 55397 : std::string obj_type("");
649 55396 : ObjectCacheMap::iterator uuid_iter = object_cache_map_.find(uuid);
650 55396 : if (uuid_iter == object_cache_map_.end()) {
651 2 : assert(!add_change);
652 2 : return;
653 : }
654 :
655 55394 : CassColumnKVVec cass_data_vec;
656 55391 : for (FieldDetailMap::iterator it =
657 55394 : uuid_iter->second->GetFieldDetailMap().begin(), itnext;
658 282105 : it != uuid_iter->second->GetFieldDetailMap().end();
659 226714 : it = itnext) {
660 226695 : itnext = it;
661 226695 : ++itnext;
662 226693 : if (it->first.key == "type") {
663 55379 : obj_type = it->first.value;
664 110758 : obj_type.erase(remove(obj_type.begin(),
665 110758 : obj_type.end(), '\"'), obj_type.end());
666 55377 : cass_data_vec.push_back(it->first);
667 : }
668 :
669 226689 : if (it->first.key == "fq_name") {
670 55376 : cass_data_vec.push_back(it->first);
671 : }
672 :
673 226709 : if (!add_change || !it->second.refreshed) {
674 84734 : if (it->first.key == "type" || it->first.key == "fq_name") {
675 42343 : continue;
676 : }
677 :
678 42392 : needNotify = true;
679 42392 : cass_data_vec.push_back(it->first);
680 42397 : if (add_change) {
681 77 : uuid_iter->second->GetFieldDetailMap().erase(it);
682 : }
683 : }
684 : }
685 :
686 55393 : if (add_change != true) {
687 21188 : object_cache_map_.erase(uuid_iter);
688 : }
689 55393 : if (needNotify) {
690 21225 : GenerateAndPushJson(uuid, obj_type, cass_data_vec, false);
691 : }
692 55393 : if (!add_change) {
693 21188 : client()->PurgeFQNameCache(uuid);
694 : }
695 55395 : }
696 :
697 190 : bool ConfigCassandraPartition::IsListOrMapPropEmpty(const string &uuid_key,
698 : const string &lookup_key) {
699 190 : string key;
700 190 : ObjectCacheMap::iterator uuid_iter = object_cache_map_.find(uuid_key);
701 190 : if (uuid_iter == object_cache_map_.end()) {
702 132 : return true;
703 : }
704 :
705 58 : key = "propm:" + lookup_key;
706 : FieldDetailMap::iterator lower_bound_it =
707 116 : uuid_iter->second->GetFieldDetailMap().lower_bound(
708 116 : JsonAdapterDataType(key, ""));
709 116 : if (lower_bound_it != uuid_iter->second->GetFieldDetailMap().end() &&
710 58 : boost::starts_with(lower_bound_it->first.key, key)) {
711 16 : return false;
712 : }
713 42 : key = "propl:" + lookup_key;
714 : lower_bound_it =
715 84 : uuid_iter->second->GetFieldDetailMap().lower_bound(
716 84 : JsonAdapterDataType(key, ""));
717 84 : if (lower_bound_it != uuid_iter->second->GetFieldDetailMap().end() &&
718 42 : boost::starts_with(lower_bound_it->first.key, key)) {
719 16 : return false;
720 : }
721 26 : return true;
722 190 : }
723 :
724 0 : bool ConfigCassandraPartition::IsTaskTriggered() const {
725 0 : return (config_reader_->IsSet());
726 : }
727 :
728 97684 : bool ConfigCassandraPartition::ConfigReader() {
729 97684 : CHECK_CONCURRENCY("config_client::Reader");
730 :
731 97690 : set<string> bunch_req_list;
732 97687 : int num_req_handled = 0;
733 : // Config reader task should stop on reinit trigger
734 97687 : for (UUIDProcessSet::iterator it = uuid_read_set_.begin(), itnext;
735 223412 : it != uuid_read_set_.end() && !client()->mgr()->is_reinit_triggered();
736 125728 : it = itnext) {
737 127818 : itnext = it;
738 127818 : ++itnext;
739 127819 : ObjectProcessRequestType *obj_req = it->second;
740 :
741 221410 : if (obj_req->oper == "CREATE" || obj_req->oper == "UPDATE" ||
742 93588 : obj_req->oper == "UPDATE-IMPLICIT") {
743 34229 : bunch_req_list.insert(obj_req->uuid);
744 34231 : bool is_last = (itnext == uuid_read_set_.end());
745 48494 : if (is_last ||
746 14263 : bunch_req_list.size() == client()->GetNumReadRequestToBunch()) {
747 19968 : if (!ReadObjUUIDTable(bunch_req_list)) {
748 2104 : return false;
749 : }
750 19968 : num_req_handled += bunch_req_list.size();
751 19968 : RemoveObjReqEntries(bunch_req_list);
752 19968 : if (num_req_handled >= client()->GetMaxRequestsToYield()) {
753 2104 : return false;
754 : }
755 : }
756 32127 : continue;
757 125718 : } else if (obj_req->oper == "DELETE") {
758 31699 : HandleObjectDelete(obj_req->uuid, false);
759 61892 : } else if (obj_req->oper == "EndOfConfig") {
760 61894 : client()->BulkSyncDone();
761 : }
762 93599 : RemoveObjReqEntry(obj_req->uuid);
763 93603 : if (++num_req_handled == client()->GetMaxRequestsToYield()) {
764 0 : return false;
765 : }
766 : }
767 :
768 : // No need to read the object uuid table if reinit is triggered
769 95587 : if (!bunch_req_list.empty() && !client()->mgr()->is_reinit_triggered()) {
770 1792 : if (!ReadObjUUIDTable(bunch_req_list))
771 0 : return false;
772 1792 : RemoveObjReqEntries(bunch_req_list);
773 : }
774 : // Clear the UUID read set if we are currently processing reinit request
775 95586 : if (client()->mgr()->is_reinit_triggered()) {
776 0 : CONFIG_CLIENT_DEBUG(ConfigClientMgrDebug,
777 : "Cassandra SM: Clear UUID read set due to reinit");
778 0 : uuid_read_set_.clear();
779 : }
780 95586 : assert(uuid_read_set_.empty());
781 95586 : return true;
782 97690 : }
783 :
784 21760 : void ConfigCassandraPartition::RemoveObjReqEntries(set<string> &req_list) {
785 90222 : BOOST_FOREACH(string uuid, req_list) {
786 34231 : RemoveObjReqEntry(uuid);
787 34231 : }
788 21760 : req_list.clear();
789 21760 : }
790 :
791 127834 : void ConfigCassandraPartition::RemoveObjReqEntry(string &uuid) {
792 : UUIDProcessSet::iterator req_it =
793 127834 : uuid_read_set_.find(client()->GetUUID(uuid));
794 127834 : delete req_it->second;
795 127834 : uuid_read_set_.erase(req_it);
796 127834 : }
797 :
798 :
799 0 : boost::asio::io_context *ConfigCassandraPartition::ioservice() {
800 0 : return client()->event_manager()->io_service();
801 : }
802 :
803 : ConfigCassandraPartition::ObjCacheEntry *
804 2 : ConfigCassandraPartition::GetObjCacheEntry(const string &uuid) {
805 2 : ObjectCacheMap::iterator uuid_iter = object_cache_map_.find(uuid);
806 2 : if (uuid_iter == object_cache_map_.end())
807 0 : return NULL;
808 2 : return uuid_iter->second;
809 : }
810 :
811 : const ConfigCassandraPartition::ObjCacheEntry *
812 125 : ConfigCassandraPartition::GetObjCacheEntry(const string &uuid) const {
813 125 : ObjectCacheMap::const_iterator uuid_iter = object_cache_map_.find(uuid);
814 125 : if (uuid_iter == object_cache_map_.end())
815 0 : return NULL;
816 125 : return uuid_iter->second;
817 : }
818 :
819 0 : int ConfigCassandraPartition::UUIDRetryTimeInMSec(
820 : const ObjCacheEntry *obj) const {
821 : uint32_t retry_time_pow_of_two =
822 0 : obj->GetRetryCount() > kMaxUUIDRetryTimePowOfTwo ?
823 0 : kMaxUUIDRetryTimePowOfTwo : obj->GetRetryCount();
824 0 : return ((1 << retry_time_pow_of_two) * kMinUUIDRetryTimeMSec);
825 : }
826 :
827 47038 : ConfigCassandraPartition::ObjCacheEntry::~ObjCacheEntry() {
828 23519 : if (retry_timer_) {
829 1 : TimerManager::DeleteTimer(retry_timer_);
830 : }
831 47038 : }
832 :
833 10 : void ConfigCassandraPartition::ObjCacheEntry::EnableCassandraReadRetry(
834 : const string uuid) {
835 10 : if (!retry_timer_) {
836 18 : retry_timer_ = TimerManager::CreateTimer(
837 6 : *parent_->client()->event_manager()->io_service(),
838 12 : "UUID retry timer for " + uuid,
839 : TaskScheduler::GetInstance()->GetTaskId(
840 : "config_client::Reader"),
841 6 : parent_->worker_id_);
842 6 : CONFIG_CLIENT_DEBUG(ConfigClientReadRetry,
843 : "Created UUID read retry timer ", uuid);
844 : }
845 10 : retry_timer_->Cancel();
846 20 : retry_timer_->Start(parent_->UUIDRetryTimeInMSec(this),
847 20 : boost::bind(
848 : &ConfigCassandraPartition::ObjCacheEntry::CassReadRetryTimerExpired,
849 : this, uuid),
850 : boost::bind(
851 : &ConfigCassandraPartition::ObjCacheEntry::CassReadRetryTimerErrorHandler,
852 : this));
853 10 : CONFIG_CLIENT_DEBUG(ConfigClientReadRetry,
854 : "Start/restart UUID Read Retry timer due to configuration", uuid);
855 10 : }
856 :
857 34217 : void ConfigCassandraPartition::ObjCacheEntry::DisableCassandraReadRetry(
858 : const string uuid) {
859 34217 : if (retry_timer_) {
860 5 : retry_timer_->Cancel();
861 5 : TimerManager::DeleteTimer(retry_timer_);
862 5 : retry_timer_ = NULL;
863 5 : retry_count_ = 0;
864 5 : CONFIG_CLIENT_DEBUG(ConfigClientReadRetry,
865 : "UUID Read retry timer - deleted timer due to configuration",
866 : uuid);
867 : }
868 34217 : }
869 :
870 36 : bool ConfigCassandraPartition::ObjCacheEntry::IsRetryTimerRunning() const {
871 36 : if (retry_timer_)
872 0 : return (retry_timer_->running());
873 36 : return false;
874 : }
875 :
876 4 : bool ConfigCassandraPartition::ObjCacheEntry::CassReadRetryTimerExpired(
877 : const string uuid) {
878 8 : parent_->client()->mgr()->EnqueueUUIDRequest(
879 8 : "UPDATE", GetObjType(), parent_->client()->uuid_str(uuid));
880 4 : retry_count_++;
881 4 : CONFIG_CLIENT_DEBUG(ConfigClientReadRetry, "timer expired ", uuid);
882 4 : return false;
883 : }
884 :
885 : void
886 0 : ConfigCassandraPartition::ObjCacheEntry::CassReadRetryTimerErrorHandler() {
887 0 : std::string message = "Timer";
888 0 : CONFIG_CLIENT_WARN(ConfigClientGetRowError,
889 : "UUID Read Retry Timer error ", message, message);
890 0 : }
891 :
892 34207 : void ConfigCassandraPartition::ListMapPropReviseUpdateList(
893 : const string &uuid, ConfigCassandraParseContext &context) {
894 34207 : for (set<string>::iterator it =
895 34207 : context.candidate_list_map_properties.begin();
896 34231 : it != context.candidate_list_map_properties.end(); it++) {
897 24 : if (context.updated_list_map_properties.find(*it) !=
898 48 : context.updated_list_map_properties.end()) {
899 8 : continue;
900 : }
901 16 : ObjectCacheMap::iterator uuid_iter = object_cache_map_.find(uuid);
902 16 : assert(uuid_iter != object_cache_map_.end());
903 : FieldDetailMap::iterator field_iter =
904 32 : uuid_iter->second->GetFieldDetailMap().lower_bound(
905 32 : JsonAdapterDataType(*it, ""));
906 16 : assert(field_iter != uuid_iter->second->GetFieldDetailMap().end());
907 16 : assert(it->compare(0, it->size() - 1, field_iter->first.key,
908 : 0, it->size() - 1) == 0);
909 80 : while (it->compare(0, it->size() - 1, field_iter->first.key,
910 80 : 0, it->size() - 1) == 0) {
911 32 : if (field_iter->second.refreshed == false) {
912 8 : context.updated_list_map_properties.insert(*it);
913 8 : break;
914 : }
915 24 : field_iter++;
916 : }
917 : }
918 34207 : }
919 :
920 142617 : bool ConfigCassandraPartition::StoreKeyIfUpdated(const string &uuid,
921 : JsonAdapterDataType *adapter, uint64_t timestamp,
922 : ConfigCassandraParseContext &context) {
923 142617 : ObjectCacheMap::iterator uuid_iter = object_cache_map_.find(uuid);
924 142609 : assert(uuid_iter != object_cache_map_.end());
925 142606 : size_t from_front_pos = adapter->key.find(':');
926 142607 : size_t from_back_pos = adapter->key.rfind(':');
927 142607 : string type_field = adapter->key.substr(0, from_front_pos+1);
928 142608 : bool is_ref = (type_field == ConfigCass2JsonAdapter::ref_prefix);
929 142608 : bool is_parent = (type_field == ConfigCass2JsonAdapter::parent_prefix);
930 142606 : bool is_propl = (type_field == ConfigCass2JsonAdapter::list_prop_prefix);
931 142605 : bool is_propm = (type_field == ConfigCass2JsonAdapter::map_prop_prefix);
932 142606 : bool is_prop = (type_field == ConfigCass2JsonAdapter::prop_prefix);
933 142599 : if (is_prop) {
934 37144 : string prop_name = adapter->key.substr(from_front_pos+1);
935 : //
936 : // properties like perms2 has no importance to control-node/dns
937 : // This property is present on each config object. Hence skipping such
938 : // properties gives performance improvement
939 : //
940 37144 : if (ConfigClientManager::skip_properties.find(prop_name) !=
941 74286 : ConfigClientManager::skip_properties.end()) {
942 611 : if ((prop_name.compare("draft_mode_state") == 0) &&
943 6 : !adapter->value.empty()) {
944 4 : context.ignore_object = true;
945 : }
946 606 : return false;
947 : }
948 37143 : }
949 :
950 141992 : string prop_name = "";
951 141999 : if (is_ref || is_parent) {
952 2846 : string ref_uuid = adapter->key.substr(from_back_pos+1);
953 :
954 2846 : string ref_name = client()->UUIDToFQName(ref_uuid).second;
955 2846 : if (ref_name == "ERROR") {
956 14 : context.parent_or_ref_fq_name_unknown = true;
957 14 : CONFIG_CLIENT_DEBUG(ConfigClientReadRetry,
958 : "Out of order parent or ref", uuid + ":" + adapter->key);
959 14 : return false;
960 : }
961 2832 : if (is_ref) {
962 1201 : adapter->ref_fq_name = ref_name;
963 : }
964 144845 : } else if (is_propl || is_propm) {
965 444 : prop_name = adapter->key.substr(0, from_back_pos);
966 :
967 444 : context.list_map_properties.insert(make_pair(prop_name, *adapter));
968 : }
969 :
970 141985 : if (adapter->key.compare("type") == 0) {
971 34212 : if (context.obj_type.empty()) {
972 34212 : context.obj_type = adapter->value;
973 68426 : context.obj_type.erase(remove(context.obj_type.begin(),
974 102638 : context.obj_type.end(), '\"'), context.obj_type.end());
975 : }
976 107770 : } else if (adapter->key.compare("fq_name") == 0) {
977 34210 : context.fq_name_present = true;
978 34210 : if (context.fq_name.empty()) {
979 34210 : context.fq_name = adapter->value.substr(1, adapter->value.size()-2);
980 68422 : context.fq_name.erase(remove(context.fq_name.begin(),
981 68422 : context.fq_name.end(), '\"'), context.fq_name.end());
982 68422 : context.fq_name.erase(remove(context.fq_name.begin(),
983 68422 : context.fq_name.end(), ' '), context.fq_name.end());
984 34211 : replace(context.fq_name.begin(), context.fq_name.end(), ',', ':');
985 : }
986 : }
987 : FieldDetailMap::iterator field_iter =
988 141988 : uuid_iter->second->GetFieldDetailMap().find(*adapter);
989 141988 : if (field_iter == uuid_iter->second->GetFieldDetailMap().end()) {
990 : // seeing field for first time
991 : FieldTimeStampInfo field_ts_info;
992 99116 : field_ts_info.refreshed = true;
993 99116 : field_ts_info.time_stamp = timestamp;
994 198241 : uuid_iter->second->GetFieldDetailMap().insert(make_pair
995 198244 : (*adapter, field_ts_info));
996 : } else {
997 42868 : field_iter->second.refreshed = true;
998 43981 : if (client()->SkipTimeStampCheckForTypeAndFQName() &&
999 1115 : ((adapter->key.compare("type") == 0) ||
1000 483 : (adapter->key.compare("fq_name") == 0))) {
1001 296 : return true;
1002 : }
1003 42570 : if (timestamp && field_iter->second.time_stamp == timestamp) {
1004 116 : if (is_propl || is_propm) {
1005 44 : context.candidate_list_map_properties.insert(prop_name);
1006 : }
1007 116 : return false;
1008 : }
1009 42454 : field_iter->second.time_stamp = timestamp;
1010 : }
1011 141579 : if (is_propl || is_propm) {
1012 401 : context.updated_list_map_properties.insert(prop_name);
1013 400 : return false;
1014 : } else {
1015 141178 : return true;
1016 : }
1017 142609 : }
1018 :
1019 : ConfigCassandraPartition::ObjCacheEntry *
1020 34231 : ConfigCassandraPartition::MarkCacheDirty(const string &uuid) {
1021 34231 : ObjectCacheMap::iterator uuid_iter = object_cache_map_.find(uuid);
1022 34231 : if (uuid_iter == object_cache_map_.end()) {
1023 : ObjCacheEntry *obj;
1024 23520 : string tmp_uuid = uuid;
1025 23520 : obj = new ObjCacheEntry(this, UTCTimestampUsec());
1026 : pair<ObjectCacheMap::iterator, bool> ret_uuid =
1027 23520 : object_cache_map_.insert(tmp_uuid, obj);
1028 23520 : assert(ret_uuid.second);
1029 23520 : uuid_iter = ret_uuid.first;
1030 23520 : } else {
1031 10711 : uuid_iter->second->SetLastReadTimeStamp(UTCTimestampUsec());
1032 : }
1033 34231 : for (FieldDetailMap::iterator it =
1034 34231 : uuid_iter->second->GetFieldDetailMap().begin();
1035 77193 : it != uuid_iter->second->GetFieldDetailMap().end(); it++) {
1036 42963 : it->second.refreshed = false;
1037 : }
1038 34231 : return uuid_iter->second;
1039 : }
1040 :
1041 36 : void ConfigCassandraPartition::FillUUIDToObjCacheInfo(const string &uuid,
1042 : ObjectCacheMap::const_iterator uuid_iter,
1043 : ConfigDBUUIDCacheEntry *entry) const {
1044 36 : entry->set_uuid(uuid);
1045 36 : entry->set_timestamp(
1046 72 : UTCUsecToString(uuid_iter->second->GetLastReadTimeStamp()));
1047 36 : entry->set_retry_count(uuid_iter->second->GetRetryCount());
1048 36 : entry->set_fq_name(uuid_iter->second->GetFQName());
1049 36 : entry->set_obj_type(uuid_iter->second->GetObjType());
1050 36 : entry->set_timer_running(uuid_iter->second->IsRetryTimerRunning());
1051 36 : entry->set_timer_created(uuid_iter->second->IsRetryTimerCreated());
1052 36 : vector<ConfigDBUUIDCacheData> fields;
1053 36 : for (FieldDetailMap::const_iterator it =
1054 36 : uuid_iter->second->GetFieldDetailMap().begin();
1055 158 : it != uuid_iter->second->GetFieldDetailMap().end(); it++) {
1056 122 : ConfigDBUUIDCacheData each_field;
1057 122 : each_field.set_refresh(it->second.refreshed);
1058 122 : each_field.set_field_name(it->first.key);
1059 122 : each_field.set_timestamp(UTCUsecToString(it->second.time_stamp));
1060 122 : fields.push_back(each_field);
1061 122 : }
1062 36 : entry->set_field_list(fields);
1063 36 : }
1064 :
1065 208 : bool ConfigCassandraPartition::UUIDToObjCacheShow(
1066 : const string &search_string, const string &last_uuid, uint32_t num_entries,
1067 : vector<ConfigDBUUIDCacheEntry> *entries) const {
1068 208 : uint32_t count = 0;
1069 208 : regex search_expr(search_string);
1070 208 : for (ObjectCacheMap::const_iterator it =
1071 208 : object_cache_map_.upper_bound(last_uuid);
1072 272 : count < num_entries && it != object_cache_map_.end(); it++) {
1073 64 : if (regex_search(it->first, search_expr) ||
1074 96 : regex_search(it->second->GetObjType(), search_expr) ||
1075 96 : regex_search(it->second->GetFQName(), search_expr)) {
1076 36 : count++;
1077 36 : ConfigDBUUIDCacheEntry entry;
1078 36 : FillUUIDToObjCacheInfo(it->first, it, &entry);
1079 36 : entries->push_back(entry);
1080 36 : }
1081 : }
1082 208 : return true;
1083 208 : }
|
