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