Line data Source code
1 : /*
2 : * Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
3 : */
4 :
5 : #include <exception>
6 : #include <mutex>
7 : #include <string>
8 : #include <boost/bind/bind.hpp>
9 : #include <boost/foreach.hpp>
10 : #include <boost/assign/list_of.hpp>
11 : #include <boost/uuid/name_generator.hpp>
12 : #include <boost/system/error_code.hpp>
13 :
14 : #include <rapidjson/document.h>
15 : #include <rapidjson/stringbuffer.h>
16 : #include <rapidjson/writer.h>
17 :
18 : #include <base/string_util.h>
19 : #include <base/logging.h>
20 : #include <io/event_manager.h>
21 : #include <base/connection_info.h>
22 : #include <base/address_util.h>
23 : #include <sandesh/sandesh_message_builder.h>
24 : #include <sandesh/protocol/TXMLProtocol.h>
25 : #include <database/cassandra/cql/cql_if.h>
26 : #include <zookeeper/zookeeper_client.h>
27 :
28 : #include "sandesh/common/flow_constants.h"
29 : #include "sandesh/common/flow_types.h"
30 : #include "viz_constants.h"
31 : #include "vizd_table_desc.h"
32 : #include "viz_collector.h"
33 : #include "collector.h"
34 : #include "db_handler.h"
35 : #include "parser_util.h"
36 : #include "db_handler_impl.h"
37 : #include "viz_sandesh.h"
38 :
39 : using namespace boost::placeholders;
40 :
41 : #define DB_LOG(_Level, _Msg) \
42 : do { \
43 : if (LoggingDisabled()) break; \
44 : log4cplus::Logger _Xlogger = log4cplus::Logger::getRoot(); \
45 : if (_Xlogger.isEnabledFor(log4cplus::_Level##_LOG_LEVEL)) { \
46 : log4cplus::tostringstream _Xbuf; \
47 : _Xbuf << name_ << ": " << __func__ << ": " << _Msg; \
48 : _Xlogger.forcedLog(log4cplus::_Level##_LOG_LEVEL, \
49 : _Xbuf.str()); \
50 : } \
51 : } while (false)
52 :
53 : using std::pair;
54 : using std::string;
55 : using boost::system::error_code;
56 : using namespace pugi;
57 : using namespace contrail::sandesh::protocol;
58 : using process::ConnectionState;
59 : using process::ConnectionType;
60 : using process::ConnectionStatus;
61 : using namespace boost::system;
62 :
63 : uint32_t DbHandler::field_cache_index_ = 0;
64 : std::set<std::string> DbHandler::field_cache_set_;
65 : std::mutex DbHandler::fmutex_;
66 :
67 1 : DbHandler::DbHandler(EventManager *evm,
68 : GenDb::GenDbIf::DbErrorHandler err_handler,
69 : std::string name,
70 : const Options::Cassandra &cassandra_options,
71 : bool use_db_write_options,
72 : const DbWriteOptions &db_write_options,
73 1 : ConfigClientCollector *config_client) :
74 1 : dbif_(new cass::cql::CqlIf(evm, cassandra_options.cassandra_ips_,
75 1 : cassandra_options.cassandra_ports_[0],
76 1 : cassandra_options.user_, cassandra_options.password_,
77 1 : cassandra_options.use_ssl_, cassandra_options.ca_certs_,
78 1 : true)),
79 1 : name_(name),
80 1 : drop_level_(SandeshLevel::INVALID),
81 1 : ttl_map_(cassandra_options.ttlmap_),
82 1 : compaction_strategy_(cassandra_options.compaction_strategy_),
83 1 : flow_tables_compaction_strategy_(
84 1 : cassandra_options.flow_tables_compaction_strategy_),
85 1 : gen_partition_no_((uint8_t)g_viz_constants.PARTITION_MIN,
86 1 : (uint8_t)g_viz_constants.PARTITION_MAX),
87 1 : disable_all_writes_(cassandra_options.disable_all_db_writes_),
88 1 : disable_statistics_writes_(cassandra_options.disable_db_stats_writes_),
89 1 : disable_messages_writes_(cassandra_options.disable_db_messages_writes_),
90 1 : config_client_(config_client),
91 6 : use_db_write_options_(use_db_write_options) {
92 1 : udc_.reset(new UserDefinedCounters());
93 1 : if (config_client) {
94 0 : config_client->RegisterConfigReceive("udc",
95 : boost::bind(&DbHandler::ReceiveConfig, this, _1, _2));
96 : }
97 1 : error_code error;
98 1 : col_name_ = ResolveCanonicalName();
99 :
100 1 : if (cassandra_options.cluster_id_.empty()) {
101 1 : tablespace_ = g_viz_constants.COLLECTOR_KEYSPACE_CQL;
102 : } else {
103 0 : tablespace_ = g_viz_constants.COLLECTOR_KEYSPACE_CQL + '_' + cassandra_options.cluster_id_;
104 : }
105 :
106 1 : if (cassandra_options.replication_factor_.empty()) {
107 1 : replication_factor_ = "1";
108 2 : DB_LOG(WARN, "CASSANDRA.replication_factor not set; defaulting to 1");
109 : } else {
110 0 : replication_factor_ = cassandra_options.replication_factor_;
111 : }
112 :
113 1 : if (use_db_write_options_) {
114 : // Set disk-usage watermark defaults
115 0 : SetDiskUsagePercentageHighWaterMark(
116 : db_write_options.get_disk_usage_percentage_high_watermark0(),
117 : db_write_options.get_high_watermark0_message_severity_level());
118 0 : SetDiskUsagePercentageLowWaterMark(
119 : db_write_options.get_disk_usage_percentage_low_watermark0(),
120 : db_write_options.get_low_watermark0_message_severity_level());
121 0 : SetDiskUsagePercentageHighWaterMark(
122 : db_write_options.get_disk_usage_percentage_high_watermark1(),
123 : db_write_options.get_high_watermark1_message_severity_level());
124 0 : SetDiskUsagePercentageLowWaterMark(
125 : db_write_options.get_disk_usage_percentage_low_watermark1(),
126 : db_write_options.get_low_watermark1_message_severity_level());
127 0 : SetDiskUsagePercentageHighWaterMark(
128 : db_write_options.get_disk_usage_percentage_high_watermark2(),
129 : db_write_options.get_high_watermark2_message_severity_level());
130 0 : SetDiskUsagePercentageLowWaterMark(
131 : db_write_options.get_disk_usage_percentage_low_watermark2(),
132 : db_write_options.get_low_watermark2_message_severity_level());
133 :
134 : // Set cassandra pending tasks watermark defaults
135 0 : SetPendingCompactionTasksHighWaterMark(
136 : db_write_options.get_pending_compaction_tasks_high_watermark0(),
137 : db_write_options.get_high_watermark0_message_severity_level());
138 0 : SetPendingCompactionTasksLowWaterMark(
139 : db_write_options.get_pending_compaction_tasks_low_watermark0(),
140 : db_write_options.get_low_watermark0_message_severity_level());
141 0 : SetPendingCompactionTasksHighWaterMark(
142 : db_write_options.get_pending_compaction_tasks_high_watermark1(),
143 : db_write_options.get_high_watermark1_message_severity_level());
144 0 : SetPendingCompactionTasksLowWaterMark(
145 : db_write_options.get_pending_compaction_tasks_low_watermark1(),
146 : db_write_options.get_low_watermark1_message_severity_level());
147 0 : SetPendingCompactionTasksHighWaterMark(
148 : db_write_options.get_pending_compaction_tasks_high_watermark2(),
149 : db_write_options.get_high_watermark2_message_severity_level());
150 0 : SetPendingCompactionTasksLowWaterMark(
151 : db_write_options.get_pending_compaction_tasks_low_watermark2(),
152 : db_write_options.get_low_watermark2_message_severity_level());
153 :
154 : // Initialize drop-levels to lowest severity level.
155 0 : SetDiskUsagePercentageDropLevel(0,
156 : db_write_options.get_low_watermark2_message_severity_level());
157 0 : SetPendingCompactionTasksDropLevel(0,
158 : db_write_options.get_low_watermark2_message_severity_level());
159 : }
160 1 : session_table_db_stats_ = SessionTableDbStats();
161 1 : }
162 :
163 :
164 6 : DbHandler::DbHandler(GenDb::GenDbIf *dbif, const TtlMap& ttl_map) :
165 6 : dbif_(dbif),
166 6 : ttl_map_(ttl_map),
167 6 : gen_partition_no_((uint8_t)g_viz_constants.PARTITION_MIN,
168 6 : (uint8_t)g_viz_constants.PARTITION_MAX),
169 6 : disable_all_writes_(false),
170 6 : disable_statistics_writes_(false),
171 6 : disable_messages_writes_(false),
172 18 : use_db_write_options_(false) {
173 6 : udc_.reset(new UserDefinedCounters());
174 6 : }
175 :
176 13 : DbHandler::~DbHandler() {
177 13 : }
178 :
179 0 : uint64_t DbHandler::GetTtlInHourFromMap(const TtlMap& ttl_map,
180 : TtlType::type type) {
181 0 : TtlMap::const_iterator it = ttl_map.find(type);
182 0 : if (it != ttl_map.end()) {
183 0 : return it->second;
184 : } else {
185 0 : return 0;
186 : }
187 : }
188 :
189 8 : uint64_t DbHandler::GetTtlFromMap(const TtlMap& ttl_map,
190 : TtlType::type type) {
191 8 : TtlMap::const_iterator it = ttl_map.find(type);
192 8 : if (it != ttl_map.end()) {
193 8 : return it->second*3600;
194 : } else {
195 0 : return 0;
196 : }
197 : }
198 :
199 0 : std::string DbHandler::GetName() const {
200 0 : return name_;
201 : }
202 :
203 0 : std::vector<boost::asio::ip::tcp::endpoint> DbHandler::GetEndpoints() const {
204 0 : return dbif_->Db_GetEndpoints();
205 : }
206 :
207 0 : void DbHandler::SetDiskUsagePercentageDropLevel(size_t count,
208 : SandeshLevel::type drop_level) {
209 0 : disk_usage_percentage_drop_level_ = drop_level;
210 0 : }
211 :
212 0 : void DbHandler::SetDiskUsagePercentage(size_t disk_usage_percentage) {
213 0 : disk_usage_percentage_ = disk_usage_percentage;
214 0 : }
215 :
216 0 : void DbHandler::SetDiskUsagePercentageHighWaterMark(
217 : uint32_t disk_usage_percentage,
218 : SandeshLevel::type level) {
219 : WaterMarkInfo wm = WaterMarkInfo(disk_usage_percentage,
220 : boost::bind(&DbHandler::SetDiskUsagePercentageDropLevel,
221 0 : this, _1, level));
222 0 : disk_usage_percentage_watermark_tuple_.SetHighWaterMark(wm);
223 0 : }
224 :
225 0 : void DbHandler::SetDiskUsagePercentageLowWaterMark(
226 : uint32_t disk_usage_percentage,
227 : SandeshLevel::type level) {
228 : WaterMarkInfo wm = WaterMarkInfo(disk_usage_percentage,
229 : boost::bind(&DbHandler::SetDiskUsagePercentageDropLevel,
230 0 : this, _1, level));
231 0 : disk_usage_percentage_watermark_tuple_.SetLowWaterMark(wm);
232 0 : }
233 :
234 0 : void DbHandler::ProcessDiskUsagePercentage(uint32_t disk_usage_percentage) {
235 0 : std::scoped_lock lock(disk_usage_percentage_water_mutex_);
236 0 : disk_usage_percentage_watermark_tuple_.ProcessWaterMarks(
237 : disk_usage_percentage,
238 0 : DbHandler::disk_usage_percentage_);
239 0 : }
240 :
241 0 : void DbHandler::SetPendingCompactionTasksDropLevel(size_t count,
242 : SandeshLevel::type drop_level) {
243 0 : pending_compaction_tasks_drop_level_ = drop_level;
244 0 : }
245 :
246 0 : void DbHandler::SetPendingCompactionTasks(size_t pending_compaction_tasks) {
247 0 : pending_compaction_tasks_ = pending_compaction_tasks;
248 0 : }
249 :
250 0 : void DbHandler::SetPendingCompactionTasksHighWaterMark(
251 : uint32_t pending_compaction_tasks,
252 : SandeshLevel::type level) {
253 : WaterMarkInfo wm = WaterMarkInfo(pending_compaction_tasks,
254 : boost::bind(&DbHandler::SetPendingCompactionTasksDropLevel, this,
255 0 : _1, level));
256 0 : pending_compaction_tasks_watermark_tuple_.SetHighWaterMark(wm);
257 0 : }
258 :
259 0 : void DbHandler::SetPendingCompactionTasksLowWaterMark(
260 : uint32_t pending_compaction_tasks,
261 : SandeshLevel::type level) {
262 : WaterMarkInfo wm = WaterMarkInfo(pending_compaction_tasks,
263 : boost::bind(&DbHandler::SetPendingCompactionTasksDropLevel, this,
264 0 : _1, level));
265 0 : pending_compaction_tasks_watermark_tuple_.SetLowWaterMark(wm);
266 0 : }
267 :
268 0 : void DbHandler::ProcessPendingCompactionTasks(
269 : uint32_t pending_compaction_tasks) {
270 0 : std::scoped_lock lock(pending_compaction_tasks_water_mutex_);
271 0 : pending_compaction_tasks_watermark_tuple_.ProcessWaterMarks(
272 : pending_compaction_tasks,
273 0 : DbHandler::pending_compaction_tasks_);
274 0 : }
275 :
276 0 : bool DbHandler::DropMessage(const SandeshHeader &header,
277 : const VizMsg *vmsg) {
278 0 : SandeshType::type stype(header.get_Type());
279 : // If Flow message, drop it
280 0 : if (stype == SandeshType::FLOW) {
281 0 : std::scoped_lock lock(smutex_);
282 0 : dropped_msg_stats_.Update(vmsg);
283 0 : return true;
284 0 : }
285 0 : if (!(stype == SandeshType::SYSTEM || stype == SandeshType::OBJECT ||
286 : stype == SandeshType::UVE ||
287 : stype == SandeshType::SESSION)) {
288 0 : return false;
289 : }
290 : // First check again the queue watermark drop level
291 : SandeshLevel::type slevel(static_cast<SandeshLevel::type>(
292 0 : header.get_Level()));
293 0 : if (slevel >= drop_level_) {
294 0 : std::scoped_lock lock(smutex_);
295 0 : dropped_msg_stats_.Update(vmsg);
296 0 : return true;
297 0 : }
298 : // Next check against the disk usage and pending compaction tasks
299 : // drop levels
300 0 : bool disk_usage_percentage_drop = false;
301 0 : bool pending_compaction_tasks_drop = false;
302 0 : if (use_db_write_options_) {
303 : SandeshLevel::type disk_usage_percentage_drop_level =
304 0 : GetDiskUsagePercentageDropLevel();
305 : SandeshLevel::type pending_compaction_tasks_drop_level =
306 0 : GetPendingCompactionTasksDropLevel();
307 0 : if (slevel >= disk_usage_percentage_drop_level) {
308 0 : disk_usage_percentage_drop = true;
309 : }
310 0 : if (slevel >= pending_compaction_tasks_drop_level) {
311 0 : pending_compaction_tasks_drop = true;
312 : }
313 : }
314 :
315 0 : bool drop(disk_usage_percentage_drop || pending_compaction_tasks_drop);
316 0 : if (drop) {
317 0 : std::scoped_lock lock(smutex_);
318 0 : dropped_msg_stats_.Update(vmsg);
319 0 : }
320 0 : return drop;
321 : }
322 :
323 0 : void DbHandler::SetDropLevel(size_t queue_count, SandeshLevel::type level,
324 : boost::function<void (void)> cb) {
325 0 : if (drop_level_ != level) {
326 0 : DB_LOG(INFO, "DB DROP LEVEL: [" <<
327 : Sandesh::LevelToString(drop_level_) << "] -> [" <<
328 : Sandesh::LevelToString(level) << "], DB QUEUE COUNT: " <<
329 : queue_count);
330 0 : drop_level_ = level;
331 : }
332 : // Always invoke the callback
333 0 : if (!cb.empty()) {
334 0 : cb();
335 : }
336 0 : }
337 :
338 0 : bool DbHandler::CreateTables() {
339 0 : for (std::vector<GenDb::NewCf>::const_iterator it = vizd_tables.begin();
340 0 : it != vizd_tables.end(); it++) {
341 0 : if (!dbif_->Db_AddColumnfamily(*it, compaction_strategy_)) {
342 0 : DB_LOG(ERROR, it->cfname_ << " FAILED");
343 0 : return false;
344 : }
345 :
346 : // find schema for it->cfname_
347 : // find all columns with index_type field set
348 0 : table_schema cfschema;
349 0 : cfschema = g_viz_constants._VIZD_TABLE_SCHEMA.find(it->cfname_)->second;
350 0 : BOOST_FOREACH (schema_column column, cfschema.columns) {
351 0 : if (column.index_mode != GenDb::ColIndexMode::NONE) {
352 0 : if (!dbif_->Db_CreateIndex(it->cfname_, column.name, "",
353 : column.index_mode)) {
354 0 : DB_LOG(ERROR, it->cfname_ << ": CreateIndex FAILED for "
355 : << column.name);
356 0 : return false;
357 : }
358 : }
359 0 : }
360 0 : }
361 :
362 0 : for (std::vector<GenDb::NewCf>::const_iterator it = vizd_stat_tables.begin();
363 0 : it != vizd_stat_tables.end(); it++) {
364 0 : if (!dbif_->Db_AddColumnfamily(*it, compaction_strategy_)) {
365 0 : DB_LOG(ERROR, it->cfname_ << " FAILED");
366 0 : return false;
367 : }
368 0 : table_schema cfschema;
369 0 : cfschema = g_viz_constants._VIZD_STAT_TABLE_SCHEMA.find(it->cfname_)->second;
370 0 : BOOST_FOREACH (schema_column column, cfschema.columns) {
371 0 : if (column.index_mode != GenDb::ColIndexMode::NONE) {
372 0 : if (!dbif_->Db_CreateIndex(it->cfname_, column.name, "",
373 : column.index_mode)) {
374 0 : DB_LOG(ERROR, it->cfname_ << ": CreateIndex FAILED for "
375 : << column.name);
376 0 : return false;
377 : }
378 : }
379 0 : }
380 0 : }
381 :
382 0 : for (std::vector<GenDb::NewCf>::const_iterator it = vizd_session_tables.begin();
383 0 : it != vizd_session_tables.end(); it++) {
384 0 : if (!dbif_->Db_AddColumnfamily(*it, flow_tables_compaction_strategy_)) {
385 0 : DB_LOG(ERROR, it->cfname_ << " FAILED");
386 0 : return false;
387 : }
388 :
389 : // will be un-commented once we start using cassandra-2.10 for systemless tests
390 : table_schema cfschema = g_viz_constants._VIZD_SESSION_TABLE_SCHEMA
391 0 : .find(it->cfname_)->second;
392 0 : BOOST_FOREACH(schema_column column, cfschema.columns) {
393 0 : if (column.index_mode != GenDb::ColIndexMode::NONE) {
394 0 : if (!dbif_->Db_CreateIndex(it->cfname_, column.name, "",
395 : column.index_mode)) {
396 0 : DB_LOG(ERROR, it->cfname_ << ": CreateIndex FAILED for "
397 : << column.name);
398 0 : return false;
399 : }
400 : }
401 0 : }
402 0 : }
403 :
404 0 : if (!dbif_->Db_SetTablespace(tablespace_)) {
405 0 : DB_LOG(ERROR, "Set KEYSPACE: " << tablespace_ << " FAILED");
406 0 : return false;
407 : }
408 0 : GenDb::ColList col_list;
409 0 : std::string cfname = g_viz_constants.SYSTEM_OBJECT_TABLE;
410 0 : GenDb::DbDataValueVec key;
411 0 : key.push_back(g_viz_constants.SYSTEM_OBJECT_ANALYTICS);
412 :
413 0 : bool init_done = false;
414 0 : if (dbif_->Db_GetRow(&col_list, cfname, key,
415 : GenDb::DbConsistency::LOCAL_ONE)) {
416 0 : for (GenDb::NewColVec::iterator it = col_list.columns_.begin();
417 0 : it != col_list.columns_.end(); it++) {
418 0 : std::string col_name;
419 : try {
420 0 : col_name = boost::get<std::string>(it->name->at(0));
421 0 : } catch (boost::bad_get& ex) {
422 0 : DB_LOG(ERROR, cfname << ": Column Name Get FAILED");
423 0 : }
424 :
425 0 : if (col_name == g_viz_constants.SYSTEM_OBJECT_START_TIME) {
426 0 : init_done = true;
427 : }
428 0 : }
429 : }
430 :
431 0 : if (!init_done) {
432 0 : std::auto_ptr<GenDb::ColList> col_list(new GenDb::ColList);
433 0 : col_list->cfname_ = g_viz_constants.SYSTEM_OBJECT_TABLE;
434 : // Rowkey
435 0 : GenDb::DbDataValueVec& rowkey = col_list->rowkey_;
436 0 : rowkey.reserve(1);
437 0 : rowkey.push_back(g_viz_constants.SYSTEM_OBJECT_ANALYTICS);
438 : // Columns
439 0 : GenDb::NewColVec& columns = col_list->columns_;
440 0 : columns.reserve(4);
441 :
442 0 : uint64_t current_tm = UTCTimestampUsec();
443 :
444 : GenDb::NewCol *col(new GenDb::NewCol(
445 0 : g_viz_constants.SYSTEM_OBJECT_START_TIME, current_tm, 0));
446 0 : columns.push_back(col);
447 :
448 : GenDb::NewCol *flow_col(new GenDb::NewCol(
449 0 : g_viz_constants.SYSTEM_OBJECT_FLOW_START_TIME, current_tm, 0));
450 0 : columns.push_back(flow_col);
451 :
452 : GenDb::NewCol *msg_col(new GenDb::NewCol(
453 0 : g_viz_constants.SYSTEM_OBJECT_MSG_START_TIME, current_tm, 0));
454 0 : columns.push_back(msg_col);
455 :
456 : GenDb::NewCol *stat_col(new GenDb::NewCol(
457 0 : g_viz_constants.SYSTEM_OBJECT_STAT_START_TIME, current_tm, 0));
458 0 : columns.push_back(stat_col);
459 :
460 0 : if (!dbif_->Db_AddColumnSync(col_list,
461 : GenDb::DbConsistency::LOCAL_ONE)) {
462 0 : DB_LOG(ERROR, g_viz_constants.SYSTEM_OBJECT_TABLE <<
463 : ": Start Time Column Add FAILED");
464 0 : return false;
465 : }
466 0 : }
467 :
468 : /*
469 : * add ttls to cassandra to be retrieved by other daemons
470 : */
471 : {
472 0 : std::auto_ptr<GenDb::ColList> col_list(new GenDb::ColList);
473 0 : col_list->cfname_ = g_viz_constants.SYSTEM_OBJECT_TABLE;
474 : // Rowkey
475 0 : GenDb::DbDataValueVec& rowkey = col_list->rowkey_;
476 0 : rowkey.reserve(1);
477 0 : rowkey.push_back(g_viz_constants.SYSTEM_OBJECT_ANALYTICS);
478 : // Columns
479 0 : GenDb::NewColVec& columns = col_list->columns_;
480 0 : columns.reserve(4);
481 :
482 : GenDb::NewCol *col(new GenDb::NewCol(
483 0 : g_viz_constants.SYSTEM_OBJECT_FLOW_DATA_TTL, (uint64_t)DbHandler::GetTtlInHourFromMap(ttl_map_, TtlType::FLOWDATA_TTL), 0));
484 0 : columns.push_back(col);
485 :
486 : GenDb::NewCol *flow_col(new GenDb::NewCol(
487 0 : g_viz_constants.SYSTEM_OBJECT_STATS_DATA_TTL, (uint64_t)DbHandler::GetTtlInHourFromMap(ttl_map_, TtlType::STATSDATA_TTL), 0));
488 0 : columns.push_back(flow_col);
489 :
490 : GenDb::NewCol *msg_col(new GenDb::NewCol(
491 0 : g_viz_constants.SYSTEM_OBJECT_CONFIG_AUDIT_TTL, (uint64_t)DbHandler::GetTtlInHourFromMap(ttl_map_, TtlType::CONFIGAUDIT_TTL), 0));
492 0 : columns.push_back(msg_col);
493 :
494 : GenDb::NewCol *stat_col(new GenDb::NewCol(
495 0 : g_viz_constants.SYSTEM_OBJECT_GLOBAL_DATA_TTL, (uint64_t)DbHandler::GetTtlInHourFromMap(ttl_map_, TtlType::GLOBAL_TTL), 0));
496 0 : columns.push_back(stat_col);
497 :
498 0 : if (!dbif_->Db_AddColumnSync(col_list,
499 : GenDb::DbConsistency::LOCAL_ONE)) {
500 0 : DB_LOG(ERROR, g_viz_constants.SYSTEM_OBJECT_TABLE <<
501 : ": TTL Column Add FAILED");
502 0 : return false;
503 : }
504 0 : }
505 :
506 0 : return true;
507 0 : }
508 :
509 0 : void DbHandler::UnInit() {
510 0 : dbif_->Db_Uninit();
511 0 : dbif_->Db_SetInitDone(false);
512 0 : }
513 :
514 0 : bool DbHandler::Init(bool initial) {
515 0 : SetDropLevel(0, SandeshLevel::INVALID, NULL);
516 0 : if (initial) {
517 0 : return Initialize();
518 : } else {
519 0 : return Setup();
520 : }
521 : }
522 :
523 0 : bool DbHandler::Initialize() {
524 0 : DB_LOG(DEBUG, "Initializing..");
525 :
526 : /* init of vizd table structures */
527 0 : init_vizd_tables();
528 :
529 0 : if (!dbif_->Db_Init()) {
530 0 : DB_LOG(ERROR, "Connection to DB FAILED");
531 0 : return false;
532 : }
533 :
534 0 : if (!dbif_->Db_AddSetTablespace(tablespace_, replication_factor_)) {
535 0 : DB_LOG(ERROR, "Create/Set KEYSPACE: " << tablespace_ << " FAILED");
536 0 : return false;
537 : }
538 :
539 0 : if (!CreateTables()) {
540 0 : DB_LOG(ERROR, "CreateTables FAILED");
541 0 : return false;
542 : }
543 :
544 0 : dbif_->Db_SetInitDone(true);
545 0 : DB_LOG(DEBUG, "Initializing Done");
546 :
547 0 : return true;
548 : }
549 :
550 0 : bool DbHandler::Setup() {
551 0 : DB_LOG(DEBUG, "Setup..");
552 0 : if (!dbif_->Db_Init()) {
553 0 : DB_LOG(ERROR, "Connection to DB FAILED");
554 0 : return false;
555 : }
556 0 : if (!dbif_->Db_SetTablespace(tablespace_)) {
557 0 : DB_LOG(ERROR, "Set KEYSPACE: " << tablespace_ << " FAILED");
558 0 : return false;
559 : }
560 0 : for (std::vector<GenDb::NewCf>::const_iterator it = vizd_tables.begin();
561 0 : it != vizd_tables.end(); it++) {
562 0 : if (!dbif_->Db_UseColumnfamily(*it)) {
563 0 : DB_LOG(ERROR, it->cfname_ <<
564 : ": Db_UseColumnfamily FAILED");
565 0 : return false;
566 : }
567 : }
568 0 : for (std::vector<GenDb::NewCf>::const_iterator it = vizd_stat_tables.begin();
569 0 : it != vizd_stat_tables.end(); it++) {
570 0 : if (!dbif_->Db_UseColumnfamily(*it)) {
571 0 : DB_LOG(ERROR, it->cfname_ << ": Db_UseColumnfamily FAILED");
572 0 : return false;
573 : }
574 : }
575 0 : dbif_->Db_SetInitDone(true);
576 0 : DB_LOG(DEBUG, "Setup Done");
577 0 : return true;
578 : }
579 :
580 19 : bool DbHandler::IsAllWritesDisabled() const {
581 19 : return disable_all_writes_;
582 : }
583 :
584 0 : bool DbHandler::IsStatisticsWritesDisabled() const {
585 0 : return disable_statistics_writes_;
586 : }
587 :
588 2 : bool DbHandler::IsMessagesWritesDisabled() const {
589 2 : return disable_messages_writes_;
590 : }
591 :
592 0 : void DbHandler::DisableAllWrites(bool disable) {
593 0 : disable_all_writes_ = disable;
594 0 : }
595 :
596 0 : void DbHandler::DisableStatisticsWrites(bool disable) {
597 0 : disable_statistics_writes_ = disable;
598 0 : }
599 :
600 0 : void DbHandler::DisableMessagesWrites(bool disable) {
601 0 : disable_messages_writes_ = disable;
602 0 : }
603 :
604 0 : void DbHandler::SetDbQueueWaterMarkInfo(Sandesh::QueueWaterMarkInfo &wm,
605 : boost::function<void (void)> defer_undefer_cb) {
606 0 : dbif_->Db_SetQueueWaterMark(boost::get<2>(wm),
607 0 : boost::get<0>(wm),
608 0 : boost::bind(&DbHandler::SetDropLevel, this, _1, boost::get<1>(wm),
609 : defer_undefer_cb));
610 0 : }
611 :
612 0 : void DbHandler::ResetDbQueueWaterMarkInfo() {
613 0 : dbif_->Db_ResetQueueWaterMarks();
614 0 : }
615 :
616 0 : void DbHandler::GetSandeshStats(std::string *drop_level,
617 : std::vector<SandeshStats> *vdropmstats) const {
618 0 : *drop_level = Sandesh::LevelToString(drop_level_);
619 0 : if (vdropmstats) {
620 0 : std::scoped_lock lock(smutex_);
621 0 : dropped_msg_stats_.Get(vdropmstats);
622 0 : }
623 0 : }
624 :
625 0 : bool DbHandler::GetStats(uint64_t *queue_count, uint64_t *enqueues) const {
626 0 : return dbif_->Db_GetQueueStats(queue_count, enqueues);
627 : }
628 :
629 0 : bool DbHandler::GetStats(std::vector<GenDb::DbTableInfo> *vdbti,
630 : GenDb::DbErrors *dbe, std::vector<GenDb::DbTableInfo> *vstats_dbti) {
631 : {
632 0 : std::scoped_lock lock(smutex_);
633 0 : stable_stats_.GetDiffs(vstats_dbti);
634 0 : }
635 0 : return dbif_->Db_GetStats(vdbti, dbe);
636 : }
637 :
638 0 : bool DbHandler::GetCumulativeStats(std::vector<GenDb::DbTableInfo> *vdbti,
639 : GenDb::DbErrors *dbe, std::vector<GenDb::DbTableInfo> *vstats_dbti) const {
640 : {
641 0 : std::scoped_lock lock(smutex_);
642 0 : stable_stats_.GetCumulative(vstats_dbti);
643 0 : }
644 0 : return dbif_->Db_GetCumulativeStats(vdbti, dbe);
645 : }
646 :
647 0 : bool DbHandler::GetCqlMetrics(cass::cql::Metrics *metrics) const {
648 0 : cass::cql::CqlIf *cql_if(dynamic_cast<cass::cql::CqlIf *>(dbif_.get()));
649 0 : if (cql_if == NULL) {
650 0 : return false;
651 : }
652 0 : return cql_if->Db_GetCqlMetrics(metrics);
653 : }
654 :
655 0 : bool DbHandler::GetCqlStats(cass::cql::DbStats *stats) const {
656 0 : cass::cql::CqlIf *cql_if(dynamic_cast<cass::cql::CqlIf *>(dbif_.get()));
657 0 : if (cql_if == NULL) {
658 0 : return false;
659 : }
660 0 : return cql_if->Db_GetCqlStats(stats);
661 : }
662 :
663 17 : bool DbHandler::InsertIntoDb(std::auto_ptr<GenDb::ColList> col_list,
664 : GenDb::DbConsistency::type dconsistency,
665 : GenDb::GenDbIf::DbAddColumnCb db_cb) {
666 17 : if (IsAllWritesDisabled()) {
667 0 : return true;
668 : }
669 17 : return dbif_->Db_AddColumn(col_list, dconsistency, db_cb);
670 : }
671 :
672 1 : bool DbHandler::AllowMessageTableInsert(const SandeshHeader &header) {
673 3 : return !IsMessagesWritesDisabled() && !IsAllWritesDisabled() &&
674 3 : (header.get_Type() != SandeshType::FLOW) &&
675 2 : (header.get_Type() != SandeshType::SESSION);
676 : }
677 :
678 3 : void DbHandler::MessageTableOnlyInsert(const VizMsg *vmsgp,
679 : const DbHandler::ObjectNamesVec &object_names,
680 : GenDb::GenDbIf::DbAddColumnCb db_cb) {
681 3 : const SandeshHeader &header(vmsgp->msg->GetHeader());
682 3 : const std::string &message_type(vmsgp->msg->GetMessageType());
683 : uint64_t timestamp;
684 : int ttl;
685 3 : if (message_type == "VncApiConfigLog") {
686 1 : ttl = GetTtl(TtlType::CONFIGAUDIT_TTL);
687 : } else {
688 2 : ttl = GetTtl(TtlType::GLOBAL_TTL);
689 : }
690 3 : std::auto_ptr<GenDb::ColList> col_list(new GenDb::ColList);
691 3 : col_list->cfname_ = g_viz_constants.COLLECTOR_GLOBAL_TABLE;
692 3 : timestamp = header.get_Timestamp();
693 3 : uint32_t T2(timestamp >> g_viz_constants.RowTimeInBits);
694 3 : uint32_t T1(timestamp & g_viz_constants.RowTimeInMask);
695 :
696 : // Rowkey
697 3 : GenDb::DbDataValueVec& rowkey = col_list->rowkey_;
698 3 : rowkey.reserve(2);
699 3 : rowkey.push_back(T2);
700 3 : rowkey.push_back(gen_partition_no_());
701 :
702 : // Columns
703 3 : GenDb::DbDataValueVec *col_name(new GenDb::DbDataValueVec());
704 3 : col_name->reserve(20);
705 3 : col_name->push_back(T1);
706 3 : col_name->push_back(vmsgp->unm);
707 :
708 : // Prepend T2: to secondary index columns
709 3 : col_name->push_back(PrependT2(T2, header.get_Source()));
710 3 : col_name->push_back(PrependT2(T2, message_type));
711 3 : col_name->push_back(PrependT2(T2, header.get_Module()));
712 :
713 : // if number of entries in object_names > MSG_TABLE_MAX_OBJECTS_PER_MSG
714 : // - print error
715 : // - increment counter + export on introspect - TODO
716 : // - let first 6 entries go through
717 3 : unsigned int count = 0;
718 15 : BOOST_FOREACH(const std::string &object_name, object_names) {
719 6 : count++;
720 6 : if (count > g_viz_constants.MSG_TABLE_MAX_OBJECTS_PER_MSG) {
721 0 : DB_LOG(ERROR, "Number of object_names in message > " <<
722 : g_viz_constants.MSG_TABLE_MAX_OBJECTS_PER_MSG <<
723 : ". Ignoring extra object_names");
724 0 : break;
725 : }
726 6 : col_name->push_back(PrependT2(T2, object_name));
727 : }
728 : // Set the value as BLANK for remaining entries if
729 : // object_names.size() < MSG_TABLE_MAX_OBJECTS_PER_MSG
730 3 : for (int i = object_names.size();
731 15 : i < g_viz_constants.MSG_TABLE_MAX_OBJECTS_PER_MSG; i++) {
732 12 : col_name->push_back(GenDb::DbDataValue());
733 : }
734 3 : if (header.__isset.IPAddress) {
735 0 : boost::system::error_code ec;
736 0 : IpAddress ipaddr(IpAddress::from_string(header.get_IPAddress(), ec));
737 0 : if (ec ) {
738 0 : LOG(ERROR, "MessageTable: INVALID IP address:"
739 : << header.get_IPAddress());
740 0 : col_name->push_back(GenDb::DbDataValue());
741 : } else {
742 0 : col_name->push_back(ipaddr);
743 : }
744 : } else {
745 3 : col_name->push_back(GenDb::DbDataValue());
746 : }
747 3 : if (header.__isset.Pid) {
748 0 : col_name->push_back((uint32_t)header.get_Pid());
749 : } else {
750 3 : col_name->push_back(GenDb::DbDataValue());
751 : }
752 3 : col_name->push_back(header.get_Category());
753 3 : col_name->push_back((uint32_t)header.get_Level());
754 3 : col_name->push_back(header.get_NodeType());
755 3 : col_name->push_back(header.get_InstanceId());
756 3 : col_name->push_back((uint32_t)header.get_SequenceNum());
757 3 : col_name->push_back((uint8_t)header.get_Type());
758 : GenDb::DbDataValueVec *col_value(new GenDb::DbDataValueVec(1,
759 3 : vmsgp->msg->ExtractMessage()));
760 3 : GenDb::NewCol *col(new GenDb::NewCol(col_name, col_value, ttl));
761 3 : GenDb::NewColVec& columns = col_list->columns_;
762 3 : columns.reserve(1);
763 3 : columns.push_back(col);
764 3 : if (!InsertIntoDb(col_list, GenDb::DbConsistency::LOCAL_ONE, db_cb)) {
765 0 : DB_LOG(ERROR, "Addition of message: " << message_type <<
766 : ", message UUID: " << vmsgp->unm << " COLUMN FAILED");
767 0 : return;
768 : }
769 3 : }
770 :
771 1 : void DbHandler::MessageTableInsert(const VizMsg *vmsgp,
772 : const DbHandler::ObjectNamesVec &object_names,
773 : GenDb::GenDbIf::DbAddColumnCb db_cb) {
774 1 : const SandeshHeader &header(vmsgp->msg->GetHeader());
775 1 : const std::string &message_type(vmsgp->msg->GetMessageType());
776 :
777 1 : if (!AllowMessageTableInsert(header))
778 0 : return;
779 :
780 1 : MessageTableOnlyInsert(vmsgp, object_names, db_cb);
781 :
782 :
783 1 : const SandeshType::type &stype(header.get_Type());
784 :
785 : /*
786 : * Insert the message types,module_id in the stat table
787 : * Construct the atttributes,attrib_tags beofore inserting
788 : * to the StatTableInsert
789 : */
790 1 : if ((stype == SandeshType::SYSLOG) || (stype == SandeshType::SYSTEM)) {
791 : //Insert only if sandesh type is a SYSTEM LOG or SYSLOG
792 : //Insert into the FieldNames stats table entries for Messagetype and Module ID
793 1 : int ttl = GetTtl(TtlType::GLOBAL_TTL);
794 1 : FieldNamesTableInsert(header.get_Timestamp(),
795 : g_viz_constants.MESSAGE_TABLE,
796 : ":Messagetype", message_type, ttl, db_cb);
797 1 : FieldNamesTableInsert(header.get_Timestamp(),
798 : g_viz_constants.MESSAGE_TABLE,
799 : ":ModuleId", header.get_Module(), ttl, db_cb);
800 1 : FieldNamesTableInsert(header.get_Timestamp(),
801 : g_viz_constants.MESSAGE_TABLE,
802 : ":Source", header.get_Source(), ttl, db_cb);
803 1 : if (!header.get_Category().empty()) {
804 0 : FieldNamesTableInsert(header.get_Timestamp(),
805 : g_viz_constants.MESSAGE_TABLE,
806 : ":Category", header.get_Category(), ttl, db_cb);
807 : }
808 : }
809 : }
810 :
811 : /*
812 : * This function takes field name and field value as arguments and inserts
813 : * into the FieldNames stats table
814 : */
815 11 : void DbHandler::FieldNamesTableInsert(uint64_t timestamp,
816 : const std::string& table_prefix,
817 : const std::string& field_name, const std::string& field_val, int ttl,
818 : GenDb::GenDbIf::DbAddColumnCb db_cb) {
819 : /*
820 : * Insert the message types in the stat table
821 : * Construct the atttributes,attrib_tags before inserting
822 : * to the StatTableInsert
823 : */
824 11 : uint32_t temp_u32 = timestamp >> g_viz_constants.RowTimeInBits;
825 11 : std::string table_name(table_prefix);
826 11 : table_name.append(field_name);
827 :
828 : /* Check if fieldname and value were already seen in this T2;
829 : 2 caches are mainted one for last T2 and T2-1.
830 : We only need to record them if they have NOT been seen yet */
831 11 : bool record = false;
832 11 : std::string fc_entry(table_name);
833 11 : fc_entry.append(":");
834 11 : fc_entry.append(field_val);
835 : {
836 11 : std::scoped_lock lock(fmutex_);
837 11 : record = CanRecordDataForT2(temp_u32, fc_entry);
838 11 : }
839 :
840 11 : if (!record) return;
841 :
842 11 : DbHandler::TagMap tmap;
843 11 : DbHandler::AttribMap amap;
844 11 : DbHandler::Var pv;
845 11 : DbHandler::AttribMap attribs;
846 11 : pv = table_name;
847 11 : tmap.insert(make_pair("name", make_pair(pv, amap)));
848 11 : attribs.insert(make_pair(string("name"), pv));
849 11 : string sattrname("fields.value");
850 11 : pv = string(field_val);
851 11 : attribs.insert(make_pair(sattrname,pv));
852 :
853 : //pv = string(header.get_Source());
854 : // Put the name of the collector, not the message source.
855 : // Using the message source will make queries slower
856 11 : pv = string(col_name_);
857 11 : tmap.insert(make_pair("Source",make_pair(pv,amap)));
858 11 : attribs.insert(make_pair(string("Source"),pv));
859 :
860 11 : StatTableInsertTtl(timestamp, "FieldNames","fields", tmap, attribs, ttl,
861 : db_cb);
862 11 : }
863 :
864 : /*
865 : * This function checks if the data can be recorded or not
866 : * for the given t2. If t2 corresponding to the data is
867 : * older than field_cache_old_t2_ and field_cache_t2_
868 : * it is ignored
869 : */
870 17 : bool DbHandler::CanRecordDataForT2(uint32_t temp_u32, std::string fc_entry) {
871 17 : bool record = false;
872 :
873 17 : uint32_t cacheindex = temp_u32 >> g_viz_constants.CacheTimeInAdditionalBits;
874 17 : if (cacheindex > field_cache_index_) {
875 2 : field_cache_index_ = cacheindex;
876 2 : field_cache_set_.clear();
877 2 : field_cache_set_.insert(fc_entry);
878 2 : record = true;
879 15 : } else if (cacheindex == field_cache_index_) {
880 15 : if (field_cache_set_.find(fc_entry) ==
881 30 : field_cache_set_.end()) {
882 13 : field_cache_set_.insert(fc_entry);
883 13 : record = true;
884 : }
885 : }
886 17 : return record;
887 : }
888 :
889 0 : void DbHandler::GetRuleMap(RuleMap& rulemap) {
890 0 : }
891 :
892 : /*
893 : * insert an entry into an ObjectTrace table
894 : * key is T2
895 : * column is
896 : * name: <key>:T1 (value in timestamp)
897 : * value: uuid (of the corresponding global message)
898 : */
899 1 : void DbHandler::ObjectTableInsert(const std::string &table, const std::string &objectkey_str,
900 : uint64_t ×tamp, const boost::uuids::uuid& unm, const VizMsg *vmsgp,
901 : GenDb::GenDbIf::DbAddColumnCb db_cb) {
902 1 : if (IsMessagesWritesDisabled() || IsAllWritesDisabled()) {
903 0 : return;
904 : }
905 1 : uint32_t T2(timestamp >> g_viz_constants.RowTimeInBits);
906 1 : uint32_t T1(timestamp & g_viz_constants.RowTimeInMask);
907 1 : const std::string &message_type(vmsgp->msg->GetMessageType());
908 : int ttl;
909 1 : if (message_type == "VncApiConfigLog") {
910 0 : ttl = GetTtl(TtlType::CONFIGAUDIT_TTL);
911 : } else {
912 1 : ttl = GetTtl(TtlType::GLOBAL_TTL);
913 : }
914 :
915 : {
916 1 : std::auto_ptr<GenDb::ColList> col_list(new GenDb::ColList);
917 1 : col_list->cfname_ = g_viz_constants.OBJECT_VALUE_TABLE;
918 1 : GenDb::DbDataValueVec& rowkey = col_list->rowkey_;
919 1 : rowkey.reserve(2);
920 1 : rowkey.push_back(T2);
921 1 : rowkey.push_back(table);
922 1 : GenDb::DbDataValueVec *col_name(new GenDb::DbDataValueVec(1, T1));
923 1 : GenDb::DbDataValueVec *col_value(new GenDb::DbDataValueVec(1, objectkey_str));
924 1 : GenDb::NewCol *col(new GenDb::NewCol(col_name, col_value, ttl));
925 1 : GenDb::NewColVec& columns = col_list->columns_;
926 1 : columns.reserve(1);
927 1 : columns.push_back(col);
928 1 : if (!InsertIntoDb(col_list, GenDb::DbConsistency::LOCAL_ONE, db_cb)) {
929 0 : DB_LOG(ERROR, "Addition of " << objectkey_str <<
930 : ", message UUID " << unm << " " << table << " into table "
931 : << g_viz_constants.OBJECT_VALUE_TABLE << " FAILED");
932 0 : return;
933 : }
934 :
935 : /*
936 : * Inserting into the stat table
937 : */
938 1 : const SandeshHeader &header(vmsgp->msg->GetHeader());
939 1 : const std::string &message_type(vmsgp->msg->GetMessageType());
940 : //Insert into the FieldNames stats table entries for Messagetype and Module ID
941 1 : FieldNamesTableInsert(timestamp,
942 : table, ":ObjectId", objectkey_str, ttl, db_cb);
943 1 : FieldNamesTableInsert(timestamp,
944 : table, ":Messagetype", message_type, ttl, db_cb);
945 1 : FieldNamesTableInsert(timestamp,
946 : table, ":ModuleId", header.get_Module(), ttl, db_cb);
947 1 : FieldNamesTableInsert(timestamp,
948 : table, ":Source", header.get_Source(), ttl, db_cb);
949 :
950 1 : FieldNamesTableInsert(timestamp,
951 : "OBJECT:", table, table, ttl, db_cb);
952 1 : }
953 : }
954 :
955 11 : bool DbHandler::StatTableWrite(uint32_t t2, const std::string& statName,
956 : const std::string& statAttr, const std::string& source, const std::string& name,
957 : const std::string& key, const std::string& proxy,
958 : const std::vector<std::vector<std::string> >& tags,
959 : uint32_t t1, const boost::uuids::uuid& unm,
960 : const std::string& jsonline, int ttl,
961 : GenDb::GenDbIf::DbAddColumnCb db_cb) {
962 :
963 11 : uint8_t part = 0;
964 11 : std::auto_ptr<GenDb::ColList> col_list(new GenDb::ColList);
965 11 : col_list->cfname_ = g_viz_constants.STATS_TABLE;
966 :
967 11 : GenDb::DbDataValueVec& rowkey = col_list->rowkey_;
968 11 : rowkey.reserve(4);
969 11 : rowkey.push_back(t2);
970 11 : rowkey.push_back(part);
971 11 : rowkey.push_back(statName);
972 11 : rowkey.push_back(statAttr);
973 :
974 11 : GenDb::DbDataValueVec *col_name(new GenDb::DbDataValueVec);
975 11 : col_name->reserve(6);
976 11 : col_name->push_back(name);
977 11 : col_name->push_back(t1);
978 11 : col_name->push_back(unm);
979 11 : col_name->push_back(PrependT2(t2, source));
980 11 : col_name->push_back(PrependT2(t2, key));
981 11 : col_name->push_back(PrependT2(t2, proxy));
982 11 : col_name->push_back(PrependT2(t2, boost::algorithm::join(tags[0], ";")));
983 11 : col_name->push_back(PrependT2(t2, boost::algorithm::join(tags[1], ";")));
984 11 : col_name->push_back(PrependT2(t2, boost::algorithm::join(tags[2], ";")));
985 11 : col_name->push_back(PrependT2(t2, boost::algorithm::join(tags[3], ";")));
986 :
987 11 : GenDb::DbDataValueVec *col_value(new GenDb::DbDataValueVec(1, jsonline));
988 11 : GenDb::NewCol *col(new GenDb::NewCol(col_name, col_value, ttl));
989 11 : GenDb::NewColVec& columns = col_list->columns_;
990 11 : columns.push_back(col);
991 :
992 11 : if (!InsertIntoDb(col_list, GenDb::DbConsistency::LOCAL_ONE, db_cb)) {
993 0 : DB_LOG(ERROR, "Addition of " << statName <<
994 : ", " << statAttr << " into table " <<
995 : g_viz_constants.STATS_TABLE <<" FAILED");
996 0 : std::scoped_lock lock(smutex_);
997 0 : stable_stats_.Update(statName + ":" + statAttr, true, true, false, 1);
998 0 : return false;
999 0 : } else {
1000 11 : std::scoped_lock lock(smutex_);
1001 11 : stable_stats_.Update(statName + ":" + statAttr, true, false, false, 1);
1002 11 : return true;
1003 11 : }
1004 11 : }
1005 :
1006 : void
1007 0 : DbHandler::StatTableInsert(uint64_t ts,
1008 : const std::string& statName,
1009 : const std::string& statAttr,
1010 : const TagMap & attribs_tag,
1011 : const AttribMap & attribs,
1012 : GenDb::GenDbIf::DbAddColumnCb db_cb) {
1013 0 : if (IsAllWritesDisabled() || IsStatisticsWritesDisabled()) {
1014 0 : return;
1015 : }
1016 0 : int ttl = GetTtl(TtlType::STATSDATA_TTL);
1017 0 : StatTableInsertTtl(ts, statName, statAttr, attribs_tag, attribs, ttl,
1018 : db_cb);
1019 : }
1020 :
1021 0 : static inline unsigned int djb_hash (const char *str, size_t len) {
1022 0 : unsigned int hash = 5381;
1023 0 : for (size_t i = 0 ; i < len ; i++)
1024 0 : hash = ((hash << 5) + hash) + str[i];
1025 0 : return hash;
1026 : }
1027 :
1028 : typedef std::pair<std::string, std::string> MapElem;
1029 :
1030 : // This function writes Stats samples to the DB.
1031 : void
1032 11 : DbHandler::StatTableInsertTtl(uint64_t ts,
1033 : const std::string& statName,
1034 : const std::string& statAttr,
1035 : const TagMap & attribs_tag,
1036 : const AttribMap & attribs, int ttl,
1037 : GenDb::GenDbIf::DbAddColumnCb db_cb) {
1038 :
1039 11 : uint64_t temp_u64 = ts;
1040 11 : uint32_t temp_u32 = temp_u64 >> g_viz_constants.RowTimeInBits;
1041 : boost::uuids::uuid unm;
1042 11 : if (statName.compare("FieldNames") != 0) {
1043 0 : unm = umn_gen_();
1044 : }
1045 :
1046 : // This is very primitive JSON encoding.
1047 : // Should replace with rapidJson at some point.
1048 :
1049 : // Encoding of all attribs
1050 :
1051 11 : contrail_rapidjson::Document dd;
1052 11 : dd.SetObject();
1053 :
1054 11 : AttribMap attribs_buf;
1055 11 : for (AttribMap::const_iterator it = attribs.begin();
1056 44 : it != attribs.end(); it++) {
1057 33 : switch (it->second.type) {
1058 33 : case STRING: {
1059 33 : contrail_rapidjson::Value val(contrail_rapidjson::kStringType);
1060 66 : std::string nm = it->first + std::string("|s");
1061 : pair<AttribMap::iterator,bool> rt =
1062 33 : attribs_buf.insert(make_pair(nm, it->second));
1063 33 : val.SetString(it->second.str.c_str(), dd.GetAllocator());
1064 33 : contrail_rapidjson::Value vk;
1065 33 : dd.AddMember(vk.SetString(rt.first->first.c_str(),
1066 : dd.GetAllocator()), val, dd.GetAllocator());
1067 33 : string field_name = it->first;
1068 33 : if (field_name.compare("fields.value") == 0) {
1069 11 : if (statName.compare("FieldNames") == 0) {
1070 : //Make uuid a fn of the field.values
1071 11 : boost::uuids::name_generator gen(DbHandler::seed_uuid);
1072 11 : unm = gen(it->second.str.c_str());
1073 : }
1074 : }
1075 33 : }
1076 33 : break;
1077 0 : case UINT64: {
1078 0 : contrail_rapidjson::Value val(contrail_rapidjson::kNumberType);
1079 0 : std::string nm = it->first + std::string("|n");
1080 : pair<AttribMap::iterator,bool> rt =
1081 0 : attribs_buf.insert(make_pair(nm, it->second));
1082 0 : val.SetUint64(it->second.num);
1083 0 : contrail_rapidjson::Value vk;
1084 0 : dd.AddMember(vk.SetString(rt.first->first.c_str(),
1085 : dd.GetAllocator()), val, dd.GetAllocator());
1086 0 : }
1087 0 : break;
1088 0 : case DOUBLE: {
1089 0 : contrail_rapidjson::Value val(contrail_rapidjson::kNumberType);
1090 0 : std::string nm = it->first + std::string("|d");
1091 : pair<AttribMap::iterator,bool> rt =
1092 0 : attribs_buf.insert(make_pair(nm, it->second));
1093 0 : val.SetDouble(it->second.dbl);
1094 0 : contrail_rapidjson::Value vk;
1095 0 : dd.AddMember(vk.SetString(rt.first->first.c_str(),
1096 : dd.GetAllocator()), val, dd.GetAllocator());
1097 0 : }
1098 0 : break;
1099 0 : case LIST: {
1100 0 : contrail_rapidjson::Value val_array(contrail_rapidjson::kArrayType);
1101 0 : std::string nm = it->first + std::string("|a");
1102 0 : BOOST_FOREACH(const std::string& elem, it->second.vec) {
1103 0 : contrail_rapidjson::Value val(contrail_rapidjson::kStringType);
1104 0 : val.SetString(elem.c_str(), dd.GetAllocator());
1105 0 : val_array.PushBack(val, dd.GetAllocator());
1106 0 : }
1107 : pair<AttribMap::iterator,bool> rt =
1108 0 : attribs_buf.insert(make_pair(nm, it->second));
1109 0 : contrail_rapidjson::Value vk;
1110 0 : dd.AddMember(vk.SetString(rt.first->first.c_str(),
1111 : dd.GetAllocator()), val_array, dd.GetAllocator());
1112 :
1113 0 : }
1114 0 : break;
1115 0 : case MAP: {
1116 0 : contrail_rapidjson::Value val_obj(contrail_rapidjson::kObjectType);
1117 0 : std::string nm = it->first + std::string("|m");
1118 0 : BOOST_FOREACH(const MapElem& pair, it->second.map) {
1119 0 : contrail_rapidjson::Value val(contrail_rapidjson::kStringType);
1120 0 : val.SetString(pair.second.c_str(), dd.GetAllocator());
1121 0 : contrail_rapidjson::Value val_key;
1122 0 : val_obj.AddMember(val_key.SetString(pair.first.c_str(),
1123 : dd.GetAllocator()), val, dd.GetAllocator());
1124 0 : }
1125 : pair<AttribMap::iterator,bool> rt =
1126 0 : attribs_buf.insert(make_pair(nm, it->second));
1127 0 : contrail_rapidjson::Value vk;
1128 0 : dd.AddMember(vk.SetString(rt.first->first.c_str(),
1129 : dd.GetAllocator()), val_obj, dd.GetAllocator());
1130 :
1131 0 : }
1132 0 : break;
1133 0 : default:
1134 0 : continue;
1135 0 : }
1136 : }
1137 :
1138 22 : contrail_rapidjson::StringBuffer sb;
1139 22 : contrail_rapidjson::Writer<contrail_rapidjson::StringBuffer> writer(sb);
1140 11 : dd.Accept(writer);
1141 22 : string jsonline(sb.GetString());
1142 :
1143 : uint32_t t1;
1144 11 : t1 = (uint32_t)(temp_u64& g_viz_constants.RowTimeInMask);
1145 :
1146 11 : if ( statName.compare("FieldNames") != 0) {
1147 0 : std::string tablename(std::string("StatTable.") + statName + "." + statAttr);
1148 0 : FieldNamesTableInsert(ts,
1149 : "STAT:", tablename, tablename, ttl, db_cb);
1150 0 : }
1151 :
1152 22 : std::vector<std::vector<std::string> > tags(4);
1153 22 : std::string name, source, key, proxy;
1154 11 : for (TagMap::const_iterator it = attribs_tag.begin();
1155 33 : it != attribs_tag.end(); it++) {
1156 :
1157 22 : pair<string,DbHandler::Var> ptag;
1158 22 : ptag.first = it->first;
1159 22 : ptag.second = it->second.first;
1160 :
1161 : /* Record in the fieldNames table if we have a string tag,
1162 : and if we are not recording a fieldNames stats entry itself */
1163 44 : if ((ptag.second.type == DbHandler::STRING) &&
1164 22 : (statName.compare("FieldNames") != 0)) {
1165 0 : FieldNamesTableInsert(ts, std::string("StatTable.") +
1166 0 : statName + "." + statAttr,
1167 0 : std::string(":") + ptag.first, ptag.second.str, ttl,
1168 : db_cb);
1169 : }
1170 :
1171 22 : if (ptag.first == g_viz_constants.STATS_NAME_FIELD) {
1172 11 : name = ptag.second.str;
1173 11 : } else if (ptag.first == g_viz_constants.STATS_SOURCE_FIELD) {
1174 11 : source = ptag.second.str;
1175 0 : } else if (boost::algorithm::ends_with(ptag.first, g_viz_constants.STATS_KEY_FIELD)) {
1176 0 : key = ptag.second.str;
1177 0 : } else if (boost::algorithm::ends_with(ptag.first, g_viz_constants.STATS_PROXY_FIELD)) {
1178 0 : proxy = ptag.second.str;
1179 : } else {
1180 0 : switch (ptag.second.type) {
1181 0 : case STRING:
1182 : case UINT64:
1183 : case DOUBLE: {
1184 0 : std::ostringstream tag_oss;
1185 0 : tag_oss << ptag.first << "=" << ptag.second;
1186 0 : size_t idx = djb_hash(ptag.first.c_str(), ptag.first.length())
1187 0 : % g_viz_constants.NUM_STATS_TAGS_FIELD;
1188 0 : tags[idx].push_back(tag_oss.str());
1189 0 : }
1190 0 : break;
1191 0 : case LIST: {
1192 0 : BOOST_FOREACH(const std::string& elem, ptag.second.vec) {
1193 0 : std::ostringstream tag_oss;
1194 0 : tag_oss << ptag.first << "=" << elem;
1195 0 : size_t idx = djb_hash(ptag.first.c_str(), ptag.first.length())
1196 0 : % g_viz_constants.NUM_STATS_TAGS_FIELD;
1197 0 : tags[idx].push_back(tag_oss.str());
1198 0 : }
1199 : }
1200 0 : break;
1201 0 : case MAP: {
1202 0 : BOOST_FOREACH(const MapElem& pair, ptag.second.map) {
1203 0 : std::ostringstream tag_oss;
1204 0 : std::string nm = ptag.first + "." + pair.first;
1205 0 : tag_oss << nm << "=" << pair.second;
1206 0 : size_t idx = djb_hash(nm.c_str(), nm.length())
1207 0 : % g_viz_constants.NUM_STATS_TAGS_FIELD;
1208 0 : tags[idx].push_back(tag_oss.str());
1209 0 : }
1210 : }
1211 0 : break;
1212 0 : default: {
1213 0 : continue;
1214 : }
1215 0 : }
1216 : }
1217 :
1218 22 : if (!it->second.second.empty()) {
1219 0 : for (AttribMap::const_iterator jt = it->second.second.begin();
1220 0 : jt != it->second.second.end(); jt++) {
1221 0 : if (jt->first == g_viz_constants.STATS_NAME_FIELD) {
1222 0 : name = jt->second.str;
1223 0 : } else if (jt->first == g_viz_constants.STATS_SOURCE_FIELD) {
1224 0 : source = jt->second.str;
1225 0 : } else if (boost::algorithm::ends_with(jt->first, g_viz_constants.STATS_KEY_FIELD)) {
1226 0 : key = jt->second.str;
1227 0 : } else if (boost::algorithm::ends_with(jt->first, g_viz_constants.STATS_PROXY_FIELD)) {
1228 0 : proxy = jt->second.str;
1229 : } else {
1230 0 : std::ostringstream tag_oss;
1231 0 : tag_oss << jt->first << "=" << jt->second;
1232 0 : size_t idx = djb_hash(jt->first.c_str(), jt->first.length())
1233 0 : % g_viz_constants.NUM_STATS_TAGS_FIELD;
1234 0 : tags[idx].push_back(tag_oss.str());
1235 0 : switch (jt->second.type) {
1236 0 : case STRING:
1237 : case UINT64:
1238 : case DOUBLE: {
1239 0 : std::ostringstream tag_oss;
1240 0 : tag_oss << jt->first << "=" << jt->second;
1241 0 : size_t idx = djb_hash(jt->first.c_str(), jt->first.length())
1242 0 : % g_viz_constants.NUM_STATS_TAGS_FIELD;
1243 0 : tags[idx].push_back(tag_oss.str());
1244 0 : }
1245 0 : break;
1246 0 : case LIST: {
1247 0 : BOOST_FOREACH(const std::string& elem, jt->second.vec) {
1248 0 : std::ostringstream tag_oss;
1249 0 : tag_oss << jt->first << "=" << elem;
1250 0 : size_t idx = djb_hash(jt->first.c_str(), jt->first.length())
1251 0 : % g_viz_constants.NUM_STATS_TAGS_FIELD;
1252 0 : tags[idx].push_back(tag_oss.str());
1253 0 : }
1254 : }
1255 0 : break;
1256 0 : case MAP: {
1257 0 : BOOST_FOREACH(const MapElem& pair, jt->second.map) {
1258 0 : std::ostringstream tag_oss;
1259 0 : std::string nm = jt->first + "." + pair.first;
1260 0 : tag_oss << nm << "=" << pair.second;
1261 0 : size_t idx = djb_hash(nm.c_str(), nm.length())
1262 0 : % g_viz_constants.NUM_STATS_TAGS_FIELD;
1263 0 : tags[idx].push_back(tag_oss.str());
1264 0 : }
1265 : }
1266 0 : break;
1267 0 : default: {
1268 0 : continue;
1269 : }
1270 0 : }
1271 0 : }
1272 : }
1273 : }
1274 22 : }
1275 11 : StatTableWrite(temp_u32, statName, statAttr, source, name, key, proxy, tags, t1,
1276 : unm, jsonline, ttl, db_cb);
1277 11 : }
1278 :
1279 : boost::uuids::uuid DbHandler::seed_uuid = StringToUuid(std::string("ffffffff-ffff-ffff-ffff-ffffffffffff"));
1280 :
1281 : SessionValueArray default_col_values = boost::assign::list_of
1282 : (GenDb::DbDataValue((uint32_t)0))
1283 : (GenDb::DbDataValue((uint8_t)0))
1284 : (GenDb::DbDataValue((uint8_t)0))
1285 : (GenDb::DbDataValue((uint8_t)0))
1286 : (GenDb::DbDataValue((uint16_t)0))
1287 : (GenDb::DbDataValue((uint16_t)0))
1288 : (GenDb::DbDataValue((uint32_t)0))
1289 : (GenDb::DbDataValue(boost::uuids::nil_uuid()))
1290 : (GenDb::DbDataValue("__UNKNOWN__"))
1291 : (GenDb::DbDataValue("__UNKNOWN__"))
1292 : (GenDb::DbDataValue("__UNKNOWN__"))
1293 : (GenDb::DbDataValue("__UNKNOWN__"))
1294 : (GenDb::DbDataValue("__UNKNOWN__"))
1295 : (GenDb::DbDataValue("__UNKNOWN__"))
1296 : (GenDb::DbDataValue("__UNKNOWN__"))
1297 : (GenDb::DbDataValue("__UNKNOWN__"))
1298 : (GenDb::DbDataValue("__UNKNOWN__"))
1299 : (GenDb::DbDataValue("__UNKNOWN__"))
1300 : (GenDb::DbDataValue("__UNKNOWN__"))
1301 : (GenDb::DbDataValue("__UNKNOWN__"))
1302 : (GenDb::DbDataValue("__UNKNOWN__"))
1303 : (GenDb::DbDataValue("__UNKNOWN__"))
1304 : (GenDb::DbDataValue("__UNKNOWN__"))
1305 : (GenDb::DbDataValue("__UNKNOWN__"))
1306 : (GenDb::DbDataValue("__UNKNOWN__"))
1307 : (GenDb::DbDataValue("__UNKNOWN__"))
1308 : (GenDb::DbDataValue(""))
1309 : (GenDb::DbDataValue(IpAddress()))
1310 : (GenDb::DbDataValue((uint64_t)0))
1311 : (GenDb::DbDataValue((uint64_t)0))
1312 : (GenDb::DbDataValue((uint64_t)0))
1313 : (GenDb::DbDataValue((uint64_t)0))
1314 : (GenDb::DbDataValue((uint64_t)0))
1315 : (GenDb::DbDataValue((uint64_t)0))
1316 : (GenDb::DbDataValue((uint64_t)0))
1317 : (GenDb::DbDataValue((uint64_t)0))
1318 : (GenDb::DbDataValue(""));
1319 :
1320 2 : static bool PopulateSessionTable(uint32_t t2, SessionValueArray& svalues,
1321 : DbInsertCb db_insert_cb, TtlMap& ttl_map) {
1322 :
1323 2 : std::auto_ptr<GenDb::ColList> colList(new GenDb::ColList);
1324 : // RowKey
1325 2 : colList->rowkey_.reserve(3);
1326 2 : colList->rowkey_.push_back(svalues[SessionRecordFields::SESSION_T2]);
1327 2 : colList->rowkey_.push_back(svalues[SessionRecordFields::SESSION_PARTITION_NO]);
1328 2 : colList->rowkey_.push_back(svalues[SessionRecordFields::SESSION_IS_SI]);
1329 4 : colList->rowkey_.push_back(svalues[
1330 2 : SessionRecordFields::SESSION_IS_CLIENT_SESSION]);
1331 :
1332 : // Column Names
1333 2 : GenDb::DbDataValueVec* cnames(new GenDb::DbDataValueVec);
1334 66 : for (int sfield = g_viz_constants.SESSION_MIN;
1335 66 : sfield != g_viz_constants.SESSION_MAX - 1; sfield++) {
1336 64 : if (svalues[sfield].which() == GenDb::DB_VALUE_BLANK) {
1337 18 : if (sfield >= g_viz_constants.SESSION_INDEX_MIN &&
1338 18 : sfield <= g_viz_constants.SESSION_INDEX_MAX) {
1339 10 : cnames->push_back(integerToString(t2) + ":" + g_viz_constants.UNKNOWN);
1340 : } else {
1341 8 : cnames->push_back(default_col_values[sfield]);
1342 : }
1343 : }
1344 : else {
1345 46 : cnames->push_back(svalues[sfield]);
1346 : }
1347 : }
1348 :
1349 : // Column Values
1350 2 : GenDb::DbDataValueVec* cvalue(new GenDb::DbDataValueVec);
1351 2 : cvalue->reserve(1);
1352 2 : cvalue->push_back(svalues[SessionRecordFields::SESSION_MAP]);
1353 :
1354 2 : int ttl = DbHandler::GetTtlFromMap(ttl_map, TtlType::FLOWDATA_TTL);
1355 :
1356 2 : colList->cfname_ = g_viz_constants.SESSION_TABLE;
1357 2 : colList->columns_.push_back(new GenDb::NewCol(cnames, cvalue, ttl));
1358 2 : if (!db_insert_cb(colList)) {
1359 0 : LOG(ERROR, "Populating SessionTable FAILED");
1360 : }
1361 :
1362 2 : return true;
1363 2 : }
1364 :
1365 2 : void JsonifySessionMap(const pugi::xml_node& root, std::string *json_string) {
1366 :
1367 2 : contrail_rapidjson::Document session_map;
1368 2 : session_map.SetObject();
1369 : contrail_rapidjson::Document::AllocatorType& allocator =
1370 2 : session_map.GetAllocator();
1371 :
1372 5 : for (pugi::xml_node ip_port = root.first_child(); ip_port; ip_port =
1373 3 : ip_port.next_sibling().next_sibling()) {
1374 3 : std::ostringstream ip_port_ss;
1375 3 : ip_port_ss << ip_port.child(g_flow_constants.PORT.c_str()).child_value() << ":"
1376 3 : << ip_port.child(g_flow_constants.IP.c_str()).child_value();
1377 3 : contrail_rapidjson::Value session_val(contrail_rapidjson::kObjectType);
1378 3 : pugi::xml_node session(ip_port.next_sibling());
1379 9 : for (pugi::xml_node field = session.first_child(); field;
1380 6 : field = field.next_sibling()) {
1381 6 : contrail_rapidjson::Value fk(contrail_rapidjson::kStringType);
1382 6 : std::string fname(field.name());
1383 : uint64_t val;
1384 6 : if (fname == "forward_flow_info" || fname == "reverse_flow_info") {
1385 6 : contrail_rapidjson::Value flow_info(contrail_rapidjson::kObjectType);
1386 6 : for (pugi::xml_node finfo = field.child("SessionFlowInfo").first_child();
1387 18 : finfo; finfo = finfo.next_sibling()) {
1388 12 : std::string name(finfo.name());
1389 12 : std::string value(finfo.child_value());
1390 : std::map<std::string, bool>::const_iterator it =
1391 12 : g_flow_constants.SessionFlowInfoField2Type.find(name);
1392 12 : assert(it != g_flow_constants.SessionFlowInfoField2Type.end());
1393 12 : if (it->second) {
1394 12 : stringToInteger(value, val);
1395 12 : contrail_rapidjson::Value fv(contrail_rapidjson::kNumberType);
1396 12 : flow_info.AddMember(fk.SetString(name.c_str(), allocator),
1397 : fv.SetUint64(val), allocator);
1398 12 : } else {
1399 0 : contrail_rapidjson::Value fv(contrail_rapidjson::kStringType);
1400 0 : flow_info.AddMember(fk.SetString(name.c_str(), allocator),
1401 : fv.SetString(value.c_str(), allocator), allocator);
1402 0 : }
1403 12 : }
1404 6 : session_val.AddMember(fk.SetString(fname.c_str(), allocator),
1405 : flow_info, allocator);
1406 6 : } else {
1407 0 : std::string fvalue(field.child_value());
1408 0 : if (stringToInteger(fvalue, val)) {
1409 0 : contrail_rapidjson::Value fv(contrail_rapidjson::kNumberType);
1410 0 : session_val.AddMember(fk.SetString(fname.c_str(), allocator),
1411 : fv.SetUint64(val), allocator);
1412 0 : } else {
1413 0 : contrail_rapidjson::Value fv(contrail_rapidjson::kStringType);
1414 0 : session_val.AddMember(fk.SetString(fname.c_str(), allocator),
1415 : fv.SetString(fvalue.c_str(), allocator), allocator);
1416 0 : }
1417 0 : }
1418 6 : }
1419 3 : contrail_rapidjson::Value vk(contrail_rapidjson::kStringType);
1420 3 : session_map.AddMember(vk.SetString(ip_port_ss.str().c_str(),
1421 : allocator), session_val, allocator);
1422 3 : }
1423 :
1424 2 : contrail_rapidjson::StringBuffer sb;
1425 2 : contrail_rapidjson::Writer<contrail_rapidjson::StringBuffer> writer(sb);
1426 2 : session_map.Accept(writer);
1427 2 : *json_string = sb.GetString();
1428 2 : }
1429 :
1430 : /*
1431 : * process the session sample and insert into the appropriate table
1432 : */
1433 1 : bool DbHandler::SessionSampleAdd(const pugi::xml_node& session_sample,
1434 : const SandeshHeader& header,
1435 : GenDb::GenDbIf::DbAddColumnCb db_cb) {
1436 1 : SessionValueArray session_entry_values;
1437 1 : pugi::xml_node &mnode = const_cast<pugi::xml_node &>(session_sample);
1438 :
1439 : // Set T1 and T2 from timestamp
1440 1 : uint64_t timestamp(header.get_Timestamp());
1441 1 : uint32_t T2(timestamp >> g_viz_constants.RowTimeInBits);
1442 1 : uint32_t T1(timestamp & g_viz_constants.RowTimeInMask);
1443 1 : session_entry_values[SessionRecordFields::SESSION_T2] = T2;
1444 1 : session_entry_values[SessionRecordFields::SESSION_T1] = T1;
1445 : // vrouter
1446 1 : session_entry_values[SessionRecordFields::SESSION_VROUTER] = header.get_Source();
1447 :
1448 1 : pugi::xml_node session_agg_info_node;
1449 : // Populate session_entry_values from message
1450 17 : for (pugi::xml_node sfield = mnode.first_child(); sfield;
1451 16 : sfield = sfield.next_sibling()) {
1452 :
1453 16 : std::string col_type(sfield.attribute("type").value());
1454 16 : std::string col_name(sfield.name());
1455 16 : SessionTypeMap::const_iterator it = session_msg2type_map.find(col_name);
1456 16 : if (it != session_msg2type_map.end()) {
1457 15 : const SessionTypeInfo &stinfo(it->second);
1458 :
1459 15 : if (col_type == "set") {
1460 1 : pugi::xml_node set = sfield.child("set");
1461 1 : std::ostringstream set_value;
1462 1 : set_value << T2 << ":";
1463 1 : int i = 0;
1464 4 : for (pugi::xml_node set_elem = set.first_child(); set_elem;
1465 3 : set_elem = set_elem.next_sibling()) {
1466 3 : if (i) {
1467 2 : set_value << ";";
1468 : }
1469 3 : std::string val = set_elem.child_value();
1470 3 : TXMLProtocol::unescapeXMLControlChars(val);
1471 3 : set_value << val;
1472 3 : i++;
1473 3 : }
1474 1 : session_entry_values[stinfo.get<0>()] = set_value.str();
1475 1 : continue;
1476 1 : }
1477 :
1478 14 : switch(stinfo.get<1>()) {
1479 2 : case GenDb::DbDataType::Unsigned8Type:
1480 : {
1481 : uint8_t val;
1482 2 : stringToInteger(sfield.child_value(), val);
1483 2 : session_entry_values[stinfo.get<0>()] =
1484 4 : static_cast<uint8_t>(val);
1485 2 : break;
1486 : }
1487 0 : case GenDb::DbDataType::Unsigned16Type:
1488 : {
1489 : uint16_t val;
1490 0 : stringToInteger(sfield.child_value(), val);
1491 0 : session_entry_values[stinfo.get<0>()] =
1492 0 : static_cast<uint16_t>(val);
1493 0 : break;
1494 : }
1495 0 : case GenDb::DbDataType::Unsigned32Type:
1496 : {
1497 : uint32_t val;
1498 0 : stringToInteger(sfield.child_value(), val);
1499 0 : session_entry_values[stinfo.get<0>()] =
1500 0 : static_cast<uint32_t>(val);
1501 0 : break;
1502 : }
1503 0 : case GenDb::DbDataType::Unsigned64Type:
1504 : {
1505 : uint64_t val;
1506 0 : stringToInteger(sfield.child_value(), val);
1507 0 : session_entry_values[stinfo.get<0>()] =
1508 0 : static_cast<uint64_t>(val);
1509 0 : break;
1510 : }
1511 0 : case GenDb::DbDataType::LexicalUUIDType:
1512 : case GenDb::DbDataType::TimeUUIDType:
1513 : {
1514 0 : std::stringstream ss;
1515 0 : ss << sfield.child_value();
1516 : boost::uuids::uuid u;
1517 0 : if (!ss.str().empty()) {
1518 0 : ss >> u;
1519 0 : if (ss.fail()) {
1520 0 : LOG(ERROR, "SessionTable: " << col_name << ": (" <<
1521 : sfield.child_value() << ") INVALID");
1522 : }
1523 : }
1524 0 : session_entry_values[stinfo.get<0>()] = u;
1525 0 : break;
1526 0 : }
1527 11 : case GenDb::DbDataType::AsciiType:
1528 : case GenDb::DbDataType::UTF8Type:
1529 : {
1530 11 : std::string val = sfield.child_value();
1531 11 : TXMLProtocol::unescapeXMLControlChars(val);
1532 11 : switch(stinfo.get<0>()) {
1533 11 : case SessionRecordFields::SESSION_DEPLOYMENT:
1534 : case SessionRecordFields::SESSION_TIER:
1535 : case SessionRecordFields::SESSION_APPLICATION:
1536 : case SessionRecordFields::SESSION_SITE:
1537 : case SessionRecordFields::SESSION_REMOTE_DEPLOYMENT:
1538 : case SessionRecordFields::SESSION_REMOTE_TIER:
1539 : case SessionRecordFields::SESSION_REMOTE_APPLICATION:
1540 : case SessionRecordFields::SESSION_REMOTE_SITE:
1541 : case SessionRecordFields::SESSION_REMOTE_PREFIX:
1542 : case SessionRecordFields::SESSION_SECURITY_POLICY_RULE:
1543 : case SessionRecordFields::SESSION_VMI:
1544 : case SessionRecordFields::SESSION_VN:
1545 : case SessionRecordFields::SESSION_REMOTE_VN:
1546 : {
1547 11 : std::ostringstream v;
1548 11 : v << T2 << ":" << val;
1549 11 : session_entry_values[stinfo.get<0>()] = v.str();
1550 11 : break;
1551 11 : }
1552 0 : default:
1553 : {
1554 0 : session_entry_values[stinfo.get<0>()] = val;
1555 0 : break;
1556 : }
1557 : }
1558 11 : break;
1559 11 : }
1560 1 : case GenDb::DbDataType::InetType:
1561 : {
1562 1 : boost::system::error_code ec;
1563 : IpAddress ipaddr(IpAddress::from_string(
1564 1 : sfield.child_value(), ec));
1565 1 : if (ec) {
1566 0 : LOG(ERROR, "SessionRecordTable: " << col_name << ": ("
1567 : << sfield.child_value() << ") INVALID");
1568 : }
1569 1 : session_entry_values[stinfo.get<0>()] = ipaddr;
1570 1 : break;
1571 : }
1572 0 : default:
1573 : {
1574 0 : VIZD_ASSERT(0);
1575 : break;
1576 : }
1577 : }
1578 1 : } else if (col_type == "map" && col_name == "sess_agg_info") {
1579 1 : session_agg_info_node = sfield.child("map");
1580 1 : continue;
1581 : }
1582 18 : }
1583 :
1584 1 : for (pugi::xml_node ip_port_proto = session_agg_info_node.first_child();
1585 3 : ip_port_proto; ip_port_proto = ip_port_proto.next_sibling().next_sibling()) {
1586 : uint16_t val;
1587 2 : stringToInteger(ip_port_proto.child(g_flow_constants.SERVICE_PORT.c_str()).child_value(), val);
1588 2 : session_entry_values[SessionRecordFields::SESSION_SPORT] = val;
1589 2 : stringToInteger(ip_port_proto.child(g_flow_constants.PROTOCOL.c_str()).child_value(), val);
1590 2 : session_entry_values[SessionRecordFields::SESSION_PROTOCOL] = val;
1591 2 : session_entry_values[SessionRecordFields::SESSION_UUID] = umn_gen_();
1592 : // Partition No
1593 2 : uint8_t partition_no = gen_partition_no_();
1594 2 : session_entry_values[SessionRecordFields::SESSION_PARTITION_NO] = partition_no;
1595 2 : std::ostringstream oss;
1596 2 : oss << T2 << ":" << ip_port_proto.child(g_flow_constants.LOCAL_IP.c_str()).child_value();
1597 2 : session_entry_values[SessionRecordFields::SESSION_IP] = oss.str();
1598 2 : pugi::xml_node sess_agg_info = ip_port_proto.next_sibling();
1599 2 : for (pugi::xml_node agg_info = sess_agg_info.first_child();
1600 12 : agg_info; agg_info = agg_info.next_sibling()) {
1601 10 : if (strcmp(agg_info.attribute("type").value(), "map") == 0) {
1602 : int16_t samples;
1603 2 : stringToInteger(agg_info.child("map").attribute("size").value(), samples);
1604 2 : session_table_db_stats_.num_samples += samples;
1605 2 : std::string session_map;
1606 2 : JsonifySessionMap(agg_info.child("map"), &session_map);
1607 2 : session_table_db_stats_.curr_json_size += session_map.size();
1608 2 : session_entry_values[SessionRecordFields::SESSION_MAP]
1609 2 : = session_map;
1610 2 : continue;
1611 2 : }
1612 8 : std::string field_name(agg_info.name());
1613 : SessionTypeMap::const_iterator it =
1614 8 : session_msg2type_map.find(field_name);
1615 8 : if (it != session_msg2type_map.end()) {
1616 8 : const SessionTypeInfo &stinfo(it->second);
1617 : uint64_t val;
1618 8 : stringToInteger(agg_info.child_value(), val);
1619 8 : session_entry_values[stinfo.get<0>()]
1620 16 : = static_cast<uint64_t>(val);
1621 : }
1622 8 : }
1623 : DbInsertCb db_insert_cb =
1624 4 : boost::bind(&DbHandler::InsertIntoDb, this, _1,
1625 2 : GenDb::DbConsistency::LOCAL_ONE, db_cb);
1626 2 : if (!PopulateSessionTable(T2, session_entry_values,
1627 2 : db_insert_cb, ttl_map_)) {
1628 0 : DB_LOG(ERROR, "Populating SessionRecordTable FAILED");
1629 : }
1630 2 : session_table_db_stats_.num_writes++;
1631 2 : }
1632 :
1633 1 : int ttl = DbHandler::GetTtlFromMap(ttl_map_, TtlType::FLOWDATA_TTL);
1634 : // insert into FieldNames table
1635 2 : FieldNamesTableInsert(timestamp, g_viz_constants.SESSION_TABLE, ":vrouter",
1636 1 : boost::get<std::string>(session_entry_values[
1637 1 : SessionRecordFields::SESSION_VROUTER]), ttl, db_cb);
1638 2 : FieldNamesTableInsert(timestamp, g_viz_constants.SESSION_TABLE, ":vn",
1639 1 : boost::get<std::string>(session_entry_values[
1640 1 : SessionRecordFields::SESSION_VN]), ttl, db_cb);
1641 2 : FieldNamesTableInsert(timestamp, g_viz_constants.SESSION_TABLE, ":remote_vn",
1642 1 : boost::get<std::string>(session_entry_values[
1643 1 : SessionRecordFields::SESSION_REMOTE_VN]), ttl, db_cb);
1644 :
1645 1 : session_table_db_stats_.num_messages++;
1646 1 : return true;
1647 1 : }
1648 :
1649 : /*
1650 : * process the session sandesh message
1651 : */
1652 :
1653 1 : bool DbHandler::SessionTableInsert(const pugi::xml_node &parent,
1654 : const SandeshHeader& header, GenDb::GenDbIf::DbAddColumnCb db_cb) {
1655 1 : pugi::xml_node session_data(parent.child("session_data"));
1656 1 : if (!session_data) {
1657 0 : return true;
1658 : }
1659 : // Session sandesh message may contain a list of session samples or
1660 : // a single session sample
1661 1 : if (strcmp(session_data.attribute("type").value(), "list") == 0) {
1662 1 : pugi::xml_node session_list = session_data.child("list");
1663 2 : for (pugi::xml_node ssample = session_list.first_child(); ssample;
1664 1 : ssample = ssample.next_sibling()) {
1665 1 : SessionSampleAdd(ssample, header, db_cb);
1666 : }
1667 : } else {
1668 0 : SessionSampleAdd(session_data.first_child(), header, db_cb);
1669 : }
1670 1 : return true;
1671 : }
1672 :
1673 0 : bool DbHandler::GetSessionTableDbInfo(SessionTableDbInfo *session_table_info) {
1674 : {
1675 0 : std::scoped_lock lock(smutex_);
1676 0 : if (session_table_db_stats_.num_messages == 0) {
1677 0 : return true;
1678 : }
1679 0 : double writes_per_message = (double)session_table_db_stats_.num_writes /
1680 0 : session_table_db_stats_.num_messages;
1681 0 : double session_per_db_write = (double)session_table_db_stats_.num_samples /
1682 0 : session_table_db_stats_.num_writes;
1683 0 : double json_size_per_write = (double)session_table_db_stats_.curr_json_size /
1684 0 : session_table_db_stats_.num_writes;
1685 0 : session_table_info->set_writes_per_message(writes_per_message);
1686 0 : session_table_info->set_sessions_per_db_record(session_per_db_write);
1687 0 : session_table_info->set_json_size_per_write(json_size_per_write);
1688 0 : session_table_info->set_num_messages(session_table_db_stats_.num_messages);
1689 0 : session_table_db_stats_.num_writes = 0;
1690 0 : session_table_db_stats_.num_samples = 0;
1691 0 : session_table_db_stats_.num_messages = 0;
1692 0 : session_table_db_stats_.curr_json_size = 0;
1693 0 : }
1694 0 : return true;
1695 : }
1696 :
1697 0 : bool DbHandler::UnderlayFlowSampleInsert(const UFlowData& flow_data,
1698 : uint64_t timestamp,
1699 : GenDb::GenDbIf::DbAddColumnCb db_cb) {
1700 0 : const std::vector<UFlowSample>& flow = flow_data.get_flow();
1701 0 : for (std::vector<UFlowSample>::const_iterator it = flow.begin();
1702 0 : it != flow.end(); ++it) {
1703 : // Add all attributes
1704 0 : DbHandler::AttribMap amap;
1705 0 : DbHandler::Var name(flow_data.get_name());
1706 0 : amap.insert(std::make_pair("name", name));
1707 0 : DbHandler::Var pifindex = it->get_pifindex();
1708 0 : amap.insert(std::make_pair("flow.pifindex", pifindex));
1709 0 : DbHandler::Var sip = it->get_sip();
1710 0 : amap.insert(std::make_pair("flow.sip", sip));
1711 0 : DbHandler::Var dip = it->get_dip();
1712 0 : amap.insert(std::make_pair("flow.dip", dip));
1713 0 : DbHandler::Var sport = static_cast<uint64_t>(it->get_sport());
1714 0 : amap.insert(std::make_pair("flow.sport", sport));
1715 0 : DbHandler::Var dport = static_cast<uint64_t>(it->get_dport());
1716 0 : amap.insert(std::make_pair("flow.dport", dport));
1717 0 : DbHandler::Var protocol = static_cast<uint64_t>(it->get_protocol());
1718 0 : amap.insert(std::make_pair("flow.protocol", protocol));
1719 0 : DbHandler::Var ft = it->get_flowtype();
1720 0 : amap.insert(std::make_pair("flow.flowtype", ft));
1721 :
1722 0 : DbHandler::TagMap tmap;
1723 : // Add tag -> name:.pifindex
1724 0 : DbHandler::AttribMap amap_name_pifindex;
1725 0 : amap_name_pifindex.insert(std::make_pair("flow.pifindex", pifindex));
1726 0 : tmap.insert(std::make_pair("name", std::make_pair(name,
1727 : amap_name_pifindex)));
1728 : // Add tag -> .sip
1729 0 : DbHandler::AttribMap amap_sip;
1730 0 : tmap.insert(std::make_pair("flow.sip", std::make_pair(sip, amap_sip)));
1731 : // Add tag -> .dip
1732 0 : DbHandler::AttribMap amap_dip;
1733 0 : tmap.insert(std::make_pair("flow.dip", std::make_pair(dip, amap_dip)));
1734 : // Add tag -> .protocol:.sport
1735 0 : DbHandler::AttribMap amap_protocol_sport;
1736 0 : amap_protocol_sport.insert(std::make_pair("flow.sport", sport));
1737 0 : tmap.insert(std::make_pair("flow.protocol",
1738 0 : std::make_pair(protocol, amap_protocol_sport)));
1739 : // Add tag -> .protocol:.dport
1740 0 : DbHandler::AttribMap amap_protocol_dport;
1741 0 : amap_protocol_dport.insert(std::make_pair("flow.dport", dport));
1742 0 : tmap.insert(std::make_pair("flow.protocol",
1743 0 : std::make_pair(protocol, amap_protocol_dport)));
1744 0 : StatTableInsert(timestamp, "UFlowData", "flow", tmap, amap, db_cb);
1745 0 : }
1746 0 : return true;
1747 : }
1748 :
1749 : using namespace zookeeper::client;
1750 :
1751 0 : DbHandlerInitializer::DbHandlerInitializer(EventManager *evm,
1752 : const std::string &db_name, const std::string &timer_task_name,
1753 : DbHandlerInitializer::InitializeDoneCb callback,
1754 : const Options::Cassandra &cassandra_options,
1755 : const std::string &zookeeper_server_list,
1756 : bool use_zookeeper,
1757 : const DbWriteOptions &db_write_options,
1758 0 : ConfigClientCollector *config_client) :
1759 0 : db_name_(db_name),
1760 0 : db_handler_(new DbHandler(evm,
1761 : boost::bind(&DbHandlerInitializer::ScheduleInit, this),
1762 : db_name, cassandra_options,
1763 0 : true, db_write_options, config_client)),
1764 0 : callback_(callback),
1765 0 : db_init_timer_(TimerManager::CreateTimer(*evm->io_service(),
1766 0 : db_name + " Db Init Timer",
1767 : TaskScheduler::GetInstance()->GetTaskId(timer_task_name))),
1768 0 : zookeeper_server_list_(zookeeper_server_list),
1769 0 : use_zookeeper_(use_zookeeper),
1770 0 : zoo_locked_(false) {
1771 0 : if (use_zookeeper_) {
1772 0 : zoo_client_.reset(new ZookeeperClient(db_name_.c_str(),
1773 0 : zookeeper_server_list_.c_str()));
1774 0 : zoo_mutex_.reset(new ZookeeperLock(zoo_client_.get(), "/collector"));
1775 : }
1776 0 : }
1777 :
1778 0 : DbHandlerInitializer::DbHandlerInitializer(EventManager *evm,
1779 : const std::string &db_name, const std::string &timer_task_name,
1780 : DbHandlerInitializer::InitializeDoneCb callback,
1781 0 : DbHandlerPtr db_handler) :
1782 0 : db_name_(db_name),
1783 0 : db_handler_(db_handler),
1784 0 : callback_(callback),
1785 0 : db_init_timer_(TimerManager::CreateTimer(*evm->io_service(),
1786 0 : db_name + " Db Init Timer",
1787 0 : TaskScheduler::GetInstance()->GetTaskId(timer_task_name))) {
1788 0 : }
1789 :
1790 0 : DbHandlerInitializer::~DbHandlerInitializer() {
1791 0 : }
1792 :
1793 0 : bool DbHandlerInitializer::Initialize() {
1794 : // Synchronize creation across nodes using zookeeper
1795 0 : if (use_zookeeper_ && !zoo_locked_) {
1796 0 : assert(zoo_mutex_->Lock());
1797 0 : zoo_locked_ = true;
1798 : }
1799 0 : if (!db_handler_->Init(true)) {
1800 0 : if (use_zookeeper_ && zoo_locked_) {
1801 0 : assert(zoo_mutex_->Release());
1802 0 : zoo_locked_ = false;
1803 : }
1804 : // Update connection info
1805 0 : ConnectionState::GetInstance()->Update(ConnectionType::DATABASE,
1806 0 : db_name_, ConnectionStatus::DOWN, db_handler_->GetEndpoints(),
1807 0 : std::string());
1808 0 : LOG(DEBUG, db_name_ << ": Db Initialization FAILED");
1809 0 : ScheduleInit();
1810 0 : return false;
1811 : }
1812 0 : if (use_zookeeper_ && zoo_locked_) {
1813 0 : assert(zoo_mutex_->Release());
1814 0 : zoo_locked_ = false;
1815 : }
1816 : // Update connection info
1817 0 : ConnectionState::GetInstance()->Update(ConnectionType::DATABASE,
1818 0 : db_name_, ConnectionStatus::UP, db_handler_->GetEndpoints(),
1819 0 : std::string());
1820 :
1821 0 : if (callback_) {
1822 0 : callback_();
1823 : }
1824 :
1825 0 : LOG(DEBUG, db_name_ << ": Db Initialization DONE");
1826 0 : return true;
1827 : }
1828 :
1829 0 : DbHandlerPtr DbHandlerInitializer::GetDbHandler() const {
1830 0 : return db_handler_;
1831 : }
1832 :
1833 0 : void DbHandlerInitializer::Shutdown() {
1834 0 : TimerManager::DeleteTimer(db_init_timer_);
1835 0 : db_init_timer_ = NULL;
1836 0 : db_handler_->UnInit();
1837 0 : }
1838 :
1839 0 : bool DbHandlerInitializer::InitTimerExpired() {
1840 : // Start the timer again if initialization is not done
1841 0 : bool done = Initialize();
1842 0 : return !done;
1843 : }
1844 :
1845 0 : void DbHandlerInitializer::InitTimerErrorHandler(string error_name,
1846 : string error_message) {
1847 0 : LOG(ERROR, db_name_ << ": " << error_name << " " << error_message);
1848 0 : }
1849 :
1850 0 : void DbHandlerInitializer::StartInitTimer() {
1851 0 : db_init_timer_->Start(kInitRetryInterval,
1852 : boost::bind(&DbHandlerInitializer::InitTimerExpired, this),
1853 : boost::bind(&DbHandlerInitializer::InitTimerErrorHandler, this,
1854 : _1, _2));
1855 0 : }
1856 :
1857 0 : void DbHandlerInitializer::ScheduleInit() {
1858 0 : db_handler_->UnInit();
1859 0 : StartInitTimer();
1860 0 : }
1861 :
1862 : // Prepend T2 in decimal since T2 timestamp column is in decimal
1863 107 : std::string PrependT2(uint32_t T2, const std::string &str) {
1864 107 : std::string tempstr = integerToString(T2);
1865 107 : tempstr.append(":");
1866 107 : tempstr.append(str);
1867 107 : return tempstr;
1868 0 : }
|
