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