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 "query.h"
7 :
8 : #include "boost/uuid/uuid_io.hpp"
9 : #include "boost/algorithm/string.hpp"
10 : #include <string>
11 : #include "base/regex.h"
12 : #include "base/util.h"
13 : #include "rapidjson/document.h"
14 : #include <analytics/viz_types.h>
15 : #include <analytics/viz_constants.h>
16 : #include <contrail-collector/vizd_table_desc.h>
17 : #include "utils.h"
18 : #include "stats_query.h"
19 : #include "stats_select.h"
20 :
21 : using contrail::regex;
22 : using contrail::regex_match;
23 : using contrail::regex_search;
24 : using std::string;
25 : using namespace boost::algorithm;
26 :
27 : std::vector<std::string> session_json_fields = boost::assign::list_of
28 : ("remote_ip")
29 : ("client_port")
30 : ("other_vrouter_ip")
31 : ("underlay_proto")
32 : ("forward_flow_uuid")
33 : ("forward_setup_time")
34 : ("forward_teardown_time")
35 : ("forward_action")
36 : ("forward_sg_rule_uuid")
37 : ("forward_nw_ave_uuid")
38 : ("forward_underlay_source_port")
39 : ("forward_drop_reason")
40 : ("forward_teardown_bytes")
41 : ("forward_teardown_pkts")
42 : ("reverse_flow_uuid")
43 : ("reverse_setup_time")
44 : ("reverse_teardown_time")
45 : ("reverse_action")
46 : ("reverse_sg_rule_uuid")
47 : ("reverse_nw_ave_uuid")
48 : ("reverse_underlay_source_port")
49 : ("reverse_drop_reason")
50 : ("reverse_teardown_bytes")
51 : ("reverse_teardown_pkts")
52 : ("sourceip")
53 : ("destip")
54 : ("dport")
55 : ("sport")
56 : ("UuidKey")
57 : ("setup_time")
58 : ("teardown_time")
59 : ("agg-bytes")
60 : ("agg-pkts")
61 : ("action")
62 : ("sg_rule_uuid")
63 : ("nw_ace_uuid")
64 : ("underlay_source_port")
65 : ("drop_reason");
66 :
67 1992 : SelectQuery::SelectQuery(QueryUnit *main_query,
68 1992 : const std::map<std::string, std::string>& json_api_data):
69 : QueryUnit(main_query, main_query),
70 1992 : provide_timeseries(false),
71 1992 : granularity(0),
72 1992 : unroll_needed(false),
73 1992 : fs_query_type_(SelectQuery::FS_SELECT_INVALID) {
74 :
75 1992 : AnalyticsQuery *m_query = (AnalyticsQuery *)main_query;
76 1992 : result_.reset(new BufT);
77 1992 : mresult_.reset(new MapBufT);
78 :
79 : // initialize Cassandra related fields
80 3447 : if (m_query->is_message_table_query() ||
81 3447 : m_query->is_object_table_query(m_query->table())) {
82 930 : cfname = g_viz_constants.COLLECTOR_GLOBAL_TABLE;
83 3180 : } else if ((m_query->table() == (g_viz_constants.FLOW_TABLE)) ||
84 2120 : (m_query->table() == (g_viz_constants.OBJECT_VALUE_TABLE))) {
85 0 : cfname = m_query->table();
86 : } else {
87 : // this is a flow series query or stats query
88 : }
89 :
90 1990 : QE_TRACE(DEBUG, "cfname :" << cfname);
91 :
92 : // Do JSON parsing of main SELECT fields
93 1992 : std::map<std::string, std::string>::const_iterator iter;
94 1992 : iter = json_api_data.find(QUERY_SELECT);
95 3018 : QE_PARSE_ERROR(iter != json_api_data.end());
96 :
97 1992 : contrail_rapidjson::Document d;
98 0 : std::string json_string = "{ \"select\" : " +
99 1992 : iter->second + " }";
100 1992 : json_string_ = json_string;
101 1992 : QE_TRACE(DEBUG, "parsing through rapidjson: " << json_string);
102 1992 : d.Parse<0>(const_cast<char *>(json_string.c_str()));
103 1990 : const contrail_rapidjson::Value& json_select_fields = d["select"];
104 1991 : QE_PARSE_ERROR(json_select_fields.IsArray());
105 1991 : QE_TRACE(DEBUG, "# of select fields is " << json_select_fields.Size());
106 :
107 : // set direction
108 1992 : int direction = INGRESS;
109 1992 : iter = json_api_data.find(QUERY_FLOW_DIR);
110 1992 : if (iter != json_api_data.end()) {
111 0 : std::istringstream(iter->second) >> direction;
112 : }
113 :
114 : // whether uuid key was provided in select field
115 1992 : bool uuid_key_selected = false;
116 1992 : bool reverse_uuid_key_selected = false;
117 1992 : bool session_class_selected = false;
118 1992 : bool session_count_selected = false;
119 :
120 1991 : if (m_query->is_stat_table_query(m_query->table())
121 1992 : || m_query->is_session_query(m_query->table())) {
122 4943 : for (contrail_rapidjson::SizeType i = 0; i < json_select_fields.Size(); i++) {
123 3921 : std::string field(json_select_fields[i].GetString());
124 3921 : if (field == SELECT_SESSION_CLASS_ID) {
125 0 : session_class_selected = true;
126 0 : QE_INVALIDARG_ERROR(m_query->table() ==
127 : g_viz_constants.SESSION_SERIES_TABLE);
128 3932 : } else if (field == SELECT_SAMPLE_COUNT) {
129 36 : session_count_selected = true;
130 36 : QE_INVALIDARG_ERROR(m_query->table() ==
131 : g_viz_constants.SESSION_SERIES_TABLE);
132 : } else {
133 3896 : if (field == "forward_flow_uuid") {
134 144 : uuid_key_selected = true;
135 : }
136 3895 : if (field == "reverse_flow_uuid") {
137 144 : reverse_uuid_key_selected = true;
138 : }
139 7775 : if (field == g_viz_constants.SESSION_FWD_TEARDOWN_BYTES ||
140 7762 : field == g_viz_constants.SESSION_FWD_TEARDOWN_PKTS ||
141 11655 : field == g_viz_constants.SESSION_REV_TEARDOWN_BYTES ||
142 3869 : field == g_viz_constants.SESSION_REV_TEARDOWN_PKTS) {
143 24 : select_column_fields.push_back("SUM(" + field + ")");
144 : } else {
145 3869 : select_column_fields.push_back(field);
146 : }
147 : }
148 : // validate select fields for session query
149 3924 : if (m_query->is_session_query(m_query->table())) {
150 2535 : if (field == g_viz_constants.STAT_TIME_FIELD ||
151 2535 : field.substr(0, g_viz_constants.STAT_TIMEBIN_FIELD.size()) ==
152 : g_viz_constants.STAT_TIMEBIN_FIELD) {
153 95 : QE_INVALIDARG_ERROR(m_query->table() == g_viz_constants.SESSION_SERIES_TABLE);
154 : } else {
155 1201 : QE_INVALIDARG_ERROR(is_valid_select_field(field));
156 : }
157 : }
158 : std::vector<std::string>::iterator it =
159 3921 : std::find(session_json_fields.begin(),
160 : session_json_fields.end(),
161 : field);
162 3922 : unroll_needed |= (it != session_json_fields.end());
163 3922 : }
164 :
165 1020 : if (session_class_selected) {
166 0 : if (select_column_fields.empty()) {
167 0 : QE_INVALIDARG_ERROR(false &&
168 : "session_class_id is not supposed to be queried alone");
169 : }
170 0 : select_column_fields.push_back("CLASS(" +
171 0 : select_column_fields[0] + ")");
172 : }
173 1020 : if (session_count_selected) {
174 36 : if (select_column_fields.empty()) {
175 0 : QE_INVALIDARG_ERROR(false &&
176 : "sample_count is not supposed to be queried alone");
177 : }
178 108 : select_column_fields.push_back("COUNT(" +
179 108 : select_column_fields[0] + ")");
180 : }
181 :
182 1020 : if (m_query->table() == g_viz_constants.SESSION_RECORD_TABLE) {
183 156 : if (!uuid_key_selected) {
184 12 : select_column_fields.push_back(g_viz_constants.FORWARD_FLOW_UUID);
185 : }
186 156 : if (!reverse_uuid_key_selected) {
187 12 : select_column_fields.push_back(g_viz_constants.REVERSE_FLOW_UUID);
188 : }
189 156 : unroll_needed = true;
190 : }
191 1020 : stats_.reset(new StatsSelect(m_query, select_column_fields));
192 1021 : QE_INVALIDARG_ERROR(stats_->Status());
193 1021 : return;
194 : }
195 :
196 970 : bool flow_class_selected = false;
197 2440 : for (contrail_rapidjson::SizeType i = 0; i < json_select_fields.Size(); i++)
198 : {
199 1471 : QE_PARSE_ERROR(json_select_fields[i].IsString());
200 :
201 : // processing "T" or "T=" field
202 1470 : if (json_select_fields[i].GetString() ==
203 2941 : std::string(TIMESTAMP_FIELD))
204 : {
205 1 : provide_timeseries = true;
206 1 : select_column_fields.push_back(TIMESTAMP_FIELD);
207 1 : QE_INVALIDARG_ERROR(m_query->is_flow_query(m_query->table()) ||
208 : m_query->is_stat_table_query(m_query->table()));
209 1469 : } else if (boost::starts_with(json_select_fields[i].GetString(),
210 : TIMESTAMP_GRANULARITY))
211 : {
212 2 : provide_timeseries = true;
213 2 : std::string timestamp_str = json_select_fields[i].GetString();
214 2 : granularity = atoi(timestamp_str.c_str() +
215 : + sizeof(TIMESTAMP_GRANULARITY)-1);
216 2 : granularity = granularity * kMicrosecInSec;
217 2 : select_column_fields.push_back(json_select_fields[i].GetString());
218 2 : QE_INVALIDARG_ERROR(
219 : m_query->table() == g_viz_constants.FLOW_SERIES_TABLE ||
220 : m_query->is_stat_table_query(m_query->table()));
221 2 : }
222 1465 : else if (json_select_fields[i].GetString() ==
223 2933 : std::string(SELECT_PACKETS)) {
224 3 : agg_stats_t agg_stats_entry = {RAW, PKT_STATS};
225 3 : agg_stats.push_back(agg_stats_entry);
226 3 : select_column_fields.push_back(SELECT_PACKETS);
227 3 : QE_INVALIDARG_ERROR(m_query->is_flow_query(m_query->table()));
228 : }
229 1462 : else if (json_select_fields[i].GetString() ==
230 2927 : std::string(SELECT_BYTES)) {
231 1 : agg_stats_t agg_stats_entry = {RAW, BYTE_STATS};
232 1 : agg_stats.push_back(agg_stats_entry);
233 1 : select_column_fields.push_back(SELECT_BYTES);
234 1 : QE_INVALIDARG_ERROR(m_query->is_flow_query(m_query->table()));
235 : }
236 1461 : else if (json_select_fields[i].GetString() ==
237 2927 : std::string(SELECT_SUM_PACKETS)) {
238 0 : agg_stats_t agg_stats_entry = {SUM, PKT_STATS};
239 0 : agg_stats.push_back(agg_stats_entry);
240 0 : select_column_fields.push_back(SELECT_SUM_PACKETS);
241 :
242 0 : QE_INVALIDARG_ERROR(
243 : m_query->table() == g_viz_constants.FLOW_SERIES_TABLE);
244 : }
245 1463 : else if (json_select_fields[i].GetString() ==
246 2927 : std::string(SELECT_SUM_BYTES)) {
247 14 : agg_stats_t agg_stats_entry = {SUM, BYTE_STATS};
248 14 : agg_stats.push_back(agg_stats_entry);
249 15 : select_column_fields.push_back(SELECT_SUM_BYTES);
250 15 : QE_INVALIDARG_ERROR(
251 : m_query->table() == g_viz_constants.FLOW_SERIES_TABLE);
252 : }
253 1449 : else if (json_select_fields[i].GetString() ==
254 2899 : std::string(SELECT_FLOW_CLASS_ID)) {
255 0 : flow_class_selected = true;
256 0 : QE_INVALIDARG_ERROR(
257 : m_query->table() == g_viz_constants.FLOW_SERIES_TABLE);
258 : }
259 1448 : else if (json_select_fields[i].GetString() ==
260 2900 : std::string(SELECT_FLOW_COUNT)) {
261 0 : QE_INVALIDARG_ERROR(0);
262 : QE_INVALIDARG_ERROR(
263 : m_query->table() == g_viz_constants.FLOW_SERIES_TABLE);
264 : }
265 : // processing other select fields
266 : else {
267 1453 : QE_INVALIDARG_ERROR(is_valid_select_field(
268 : json_select_fields[i].GetString()));
269 2893 : if (json_select_fields[i].GetString() == g_viz_constants.FLOW_TABLE_AGG_PKTS ||
270 1447 : json_select_fields[i].GetString() == g_viz_constants.FLOW_TABLE_AGG_BYTES) {
271 0 : std::string field(json_select_fields[i].GetString());
272 0 : select_column_fields.push_back("SUM("+ field +")");
273 0 : } else {
274 1447 : select_column_fields.push_back(
275 2892 : get_column_name(json_select_fields[i].GetString()));
276 : }
277 1446 : if (json_select_fields[i].GetString() == g_viz_constants.UUID_KEY)
278 0 : uuid_key_selected = true;
279 : std::vector<std::string>::iterator it =
280 1446 : std::find(session_json_fields.begin(),
281 : session_json_fields.end(),
282 1446 : json_select_fields[i].GetString());
283 1444 : std::string field = json_select_fields[i].GetString();
284 1444 : if ((direction == 1 && field == "sourceip")
285 2894 : || (direction == 0 && field == "destip")) {
286 24 : unroll_needed |= false;
287 : } else {
288 1426 : unroll_needed |= (it != session_json_fields.end());
289 : }
290 1448 : }
291 : }
292 :
293 968 : if (m_query->table() == g_viz_constants.FLOW_SERIES_TABLE) {
294 40 : evaluate_fs_query_type();
295 40 : if (fs_query_type_ == SelectQuery::FS_SELECT_INVALID)
296 12 : QE_INVALIDARG_ERROR(false);
297 : }
298 :
299 963 : if (flow_class_selected) {
300 : try {
301 0 : select_column_fields.push_back("CLASS(" +
302 0 : select_column_fields[0] + ")");
303 0 : } catch (std::out_of_range& e) {
304 0 : QE_INVALIDARG_ERROR(0);
305 0 : }
306 : }
307 :
308 963 : if ((m_query->table() == g_viz_constants.FLOW_TABLE) && !uuid_key_selected) {
309 0 : select_column_fields.push_back(g_viz_constants.UUID_KEY);
310 0 : unroll_needed = true;
311 : }
312 :
313 963 : if (m_query->is_flow_query(m_query->table())) {
314 36 : if (!m_query->wherequery_->additional_select_.empty()) {
315 : // add additional select fields based on filters
316 0 : select_column_fields.insert(select_column_fields.end(),
317 0 : m_query->wherequery_->additional_select_.begin(),
318 0 : m_query->wherequery_->additional_select_.end());
319 : }
320 36 : stats_.reset(new StatsSelect(m_query, select_column_fields));
321 36 : QE_INVALIDARG_ERROR(stats_->Status());
322 : }
323 :
324 3016 : }
325 :
326 2650 : bool SelectQuery::is_valid_select_field(const std::string& select_field) const {
327 2650 : AnalyticsQuery *mquery = (AnalyticsQuery*)main_query;
328 2650 : const std::string& table = mquery->table();
329 :
330 11501 : for(size_t i = 0; i < g_viz_constants._TABLES.size(); i++) {
331 10951 : if (g_viz_constants._TABLES[i].name == table) {
332 40009 : for (size_t j = 0;
333 40009 : j < g_viz_constants._TABLES[i].schema.columns.size(); j++) {
334 39990 : if (g_viz_constants._TABLES[i].schema.columns[j].name ==
335 : select_field)
336 2100 : return true;
337 : }
338 1 : return false;
339 : }
340 : }
341 :
342 547 : return true;
343 2648 : }
344 :
345 40 : bool SelectQuery::is_flow_tuple_specified() {
346 40 : for (std::vector<std::string>::const_iterator it =
347 47 : select_column_fields.begin(); it != select_column_fields.end(); ++it) {
348 45 : std::string qstring(get_query_string(*it));
349 45 : if (qstring == g_viz_constants.FlowRecordNames[
350 90 : FlowRecordFields::FLOWREC_VROUTER] ||
351 45 : qstring == g_viz_constants.FlowRecordNames[
352 65 : FlowRecordFields::FLOWREC_SOURCEVN] ||
353 20 : qstring == g_viz_constants.FlowRecordNames[
354 40 : FlowRecordFields::FLOWREC_SOURCEIP] ||
355 20 : qstring == g_viz_constants.FlowRecordNames[
356 27 : FlowRecordFields::FLOWREC_DESTVN] ||
357 7 : qstring == g_viz_constants.FlowRecordNames[
358 14 : FlowRecordFields::FLOWREC_DESTIP] ||
359 7 : qstring == g_viz_constants.FlowRecordNames[
360 14 : FlowRecordFields::FLOWREC_PROTOCOL] ||
361 7 : qstring == g_viz_constants.FlowRecordNames[
362 14 : FlowRecordFields::FLOWREC_SPORT] ||
363 7 : qstring == g_viz_constants.FlowRecordNames[
364 97 : FlowRecordFields::FLOWREC_DPORT] ||
365 7 : qstring == g_viz_constants.FlowRecordNames[
366 7 : FlowRecordFields::FLOWREC_DIRECTION_ING]) {
367 38 : return true;
368 : }
369 45 : }
370 2 : return false;
371 : }
372 :
373 40 : void SelectQuery::evaluate_fs_query_type() {
374 40 : if (provide_timeseries) {
375 3 : if (granularity) {
376 2 : fs_query_type_ |= SelectQuery::FS_SELECT_TS;
377 : } else {
378 1 : fs_query_type_ |= SelectQuery::FS_SELECT_T;
379 : }
380 : }
381 40 : if (is_flow_tuple_specified()) {
382 38 : fs_query_type_ |= SelectQuery::FS_SELECT_FLOW_TUPLE;
383 : }
384 40 : if (agg_stats.size()) {
385 16 : fs_query_type_ |= SelectQuery::FS_SELECT_STATS;
386 16 : agg_op_t stats_op = AGG_OP_INVALID;
387 16 : std::vector<agg_stats_t>::const_iterator it;
388 29 : for (it = agg_stats.begin(); it != agg_stats.end(); ++it) {
389 17 : if ((*it).agg_op == RAW) {
390 3 : if ((fs_query_type_ & SelectQuery::FS_SELECT_TS) ||
391 : (stats_op == SUM)) {
392 3 : fs_query_type_ = SelectQuery::FS_SELECT_INVALID;
393 4 : return;
394 : }
395 0 : stats_op = RAW;
396 0 : fs_query_type_ |= SelectQuery::FS_SELECT_T;
397 14 : } else if ((*it).agg_op == SUM) {
398 14 : if ((fs_query_type_ & SelectQuery::FS_SELECT_T) ||
399 : (stats_op == RAW)) {
400 1 : fs_query_type_ = SelectQuery::FS_SELECT_INVALID;
401 1 : return;
402 : }
403 13 : stats_op = SUM;
404 : }
405 : }
406 : }
407 : }
408 :
409 : class SessionTableAttributeConverter : public boost::static_visitor<> {
410 : public:
411 234 : SessionTableAttributeConverter(std::vector<StatsSelect::StatEntry> *attribs):
412 234 : attribs_(attribs),
413 468 : cass_column2column_name_map_((g_viz_constants._VIZD_SESSION_TABLE_SCHEMA
414 234 : .find(g_viz_constants.SESSION_TABLE))->second
415 234 : .column_to_query_column) {
416 234 : }
417 0 : void operator()(const boost::blank &tblank, const int idx) const {
418 0 : QE_ASSERT(false && "Null Value in Query Result");
419 : }
420 4446 : void operator()(const std::string &tstring, const int idx) const {
421 4446 : if (tstring.empty()) {
422 0 : return;
423 : }
424 4446 : StatsSelect::StatEntry se;
425 4446 : std::string cname = "column" + integerToString(idx);
426 4446 : std::map<std::string, std::string>::const_iterator itr;
427 4446 : itr = (cass_column2column_name_map_.find(cname));
428 4445 : QE_ASSERT(itr != cass_column2column_name_map_.end());
429 4445 : se.name = itr->second;
430 : GenDb::ColIndexType::type index_type =
431 4445 : g_viz_constants._VIZD_SESSION_TABLE_SCHEMA.find(
432 4445 : g_viz_constants.SESSION_TABLE)->second.columns[idx].index_type;
433 4445 : if (index_type) {
434 : std::string value(regex_replace(tstring,
435 : SessionTableAttributeConverter::t2_expr_, "",
436 4211 : boost::match_default | boost::format_all));
437 4211 : se.value = value;
438 4209 : } else {
439 234 : se.value = tstring;
440 : }
441 4443 : attribs_->push_back(se);
442 4445 : }
443 0 : void operator()(const boost::uuids::uuid &tuuid, const int idx) const {
444 0 : StatsSelect::StatEntry se;
445 0 : std::string cname = "column" + integerToString(idx);
446 0 : std::map<std::string, std::string>::const_iterator itr;
447 0 : itr = (cass_column2column_name_map_.find(cname));
448 0 : QE_ASSERT(itr != cass_column2column_name_map_.end());
449 0 : se.name = itr->second;
450 0 : se.value = to_string(tuuid);
451 0 : attribs_->push_back(se);
452 0 : }
453 : template <typename IntegerType>
454 2016 : void operator()(const IntegerType &num, const int idx) const {
455 2016 : StatsSelect::StatEntry se;
456 2016 : std::string cname = "column" + integerToString(idx);
457 2016 : std::map<std::string, std::string>::const_iterator itr;
458 2016 : itr = (cass_column2column_name_map_.find(cname));
459 2016 : QE_ASSERT(itr != cass_column2column_name_map_.end());
460 2016 : se.name = itr->second;
461 2016 : se.value = (uint64_t)num;
462 2014 : attribs_->push_back(se);
463 2015 : }
464 0 : void operator()(const double &tdouble, const int idx) const {
465 0 : StatsSelect::StatEntry se;
466 0 : std::string cname = "column" + integerToString(idx);
467 0 : std::map<std::string, std::string>::const_iterator itr;
468 0 : itr = (cass_column2column_name_map_.find(cname));
469 0 : QE_ASSERT(itr != cass_column2column_name_map_.end());
470 0 : se.name = itr->second;
471 0 : se.value = tdouble;
472 0 : attribs_->push_back(se);
473 0 : }
474 234 : void operator()(const IpAddress &tipaddr, const int idx) const {
475 234 : StatsSelect::StatEntry se;
476 234 : std::string cname = "column" + integerToString(idx);
477 234 : std::map<std::string, std::string>::const_iterator itr;
478 234 : itr = (cass_column2column_name_map_.find(cname));
479 234 : QE_ASSERT(itr != cass_column2column_name_map_.end());
480 234 : se.name = itr->second;
481 234 : se.value = tipaddr.to_string();
482 234 : attribs_->push_back(se);
483 234 : }
484 0 : void operator()(const GenDb::Blob &tblob, const int idx) const {
485 0 : QE_ASSERT(0);
486 : }
487 : private:
488 : std::vector<StatsSelect::StatEntry> *attribs_;
489 : std::map<std::string, std::string> cass_column2column_name_map_;
490 : static regex t2_expr_;
491 : };
492 :
493 : regex SessionTableAttributeConverter::t2_expr_("^[\\d]+:");
494 :
495 4572 : void populate_attribs_from_json_member(
496 : const contrail_rapidjson::Value::ConstMemberIterator itr,
497 : std::vector<StatsSelect::StatEntry> *attribs_, const std::string& prefix) {
498 4572 : QE_ASSERT(itr->name.IsString());
499 :
500 4572 : std::string fvname(itr->name.GetString());
501 :
502 4573 : StatsSelect::StatEntry se;
503 4571 : se.name = prefix + fvname;
504 4573 : if (itr->value.IsString()) {
505 2375 : se.value = itr->value.GetString();
506 2201 : } else if (itr->value.IsUint()) {
507 2201 : se.value = (uint64_t)itr->value.GetUint();
508 0 : } else if (itr->value.IsUint64()){
509 0 : se.value = (uint64_t)itr->value.GetUint64();
510 0 : } else if (itr->value.IsDouble()) {
511 0 : se.value = (double) itr->value.GetDouble();
512 : } else {
513 0 : QE_ASSERT(0);
514 : }
515 4576 : attribs_->push_back(se);
516 4575 : }
517 :
518 297 : void parse_json(const contrail_rapidjson::Value &json_object,
519 : std::vector<StatsSelect::StatEntry> *attribs_) {
520 297 : for (contrail_rapidjson::Value::ConstMemberIterator itr =
521 1188 : json_object.MemberBegin(); itr != json_object.MemberEnd(); ++itr) {
522 594 : QE_ASSERT(itr->name.IsString());
523 594 : if (itr->value.IsObject()) {
524 594 : std::string prefix;
525 594 : if (itr->name == "forward_flow_info") {
526 297 : prefix = "forward_";
527 : } else {
528 297 : prefix = "reverse_";
529 : }
530 594 : for (contrail_rapidjson::Value::ConstMemberIterator itr2 =
531 5762 : itr->value.MemberBegin(); itr2 != itr->value.MemberEnd(); ++itr2) {
532 4573 : populate_attribs_from_json_member(itr2, attribs_, prefix);
533 : }
534 592 : } else {
535 0 : populate_attribs_from_json_member(itr, attribs_, "");
536 : }
537 : }
538 297 : }
539 :
540 33 : void map_session_to_flow(std::vector<StatsSelect::StatEntry> *attribs_,
541 : uint8_t is_client_session, uint8_t direction) {
542 33 : std::map<std::string, std::string> session2flow_map;
543 33 : if (direction) {
544 33 : if (is_client_session) {
545 21 : session2flow_map = g_viz_constants.session2flow_maps[0];
546 : } else {
547 12 : session2flow_map = g_viz_constants.session2flow_maps[2];
548 : }
549 : } else {
550 0 : if (is_client_session) {
551 0 : session2flow_map = g_viz_constants.session2flow_maps[1];
552 : } else {
553 0 : session2flow_map = g_viz_constants.session2flow_maps[3];
554 : }
555 : }
556 :
557 33 : std::vector<StatsSelect::StatEntry>::iterator it;
558 1122 : for (it = attribs_->begin(); it != attribs_->end(); it++) {
559 : std::map<std::string, std::string>::iterator itr =
560 1089 : session2flow_map.find(it->name);
561 1089 : if (itr != session2flow_map.end()) {
562 198 : it->name = itr->second;
563 : }
564 : }
565 33 : }
566 :
567 857 : query_status_t SelectQuery::process_query() {
568 :
569 857 : if (status_details != 0)
570 : {
571 0 : QE_TRACE(DEBUG,
572 : "No need to process query, as there were errors previously");
573 0 : return QUERY_FAILURE;
574 : }
575 :
576 : /*
577 : * various select queries
578 : * flow related queries
579 : * select T=5, sum(pkts)...
580 : * select <x-tuple>, sum(pkts)...
581 : * select T=5, <x-tuple>, sum(pkts)...
582 : *
583 : * select T, pkts...
584 : * select <x-tuple>, pkts...
585 : * select T, <x-tuple>, pkts...
586 : *
587 : * For all the above queries the output from select will be
588 : * a series of rows of
589 : * T, <x-tuple>, pkts...
590 : * it's expected that aggregation and binning will be done by
591 : * the next level
592 : *
593 : * message related queries
594 : */
595 857 : AnalyticsQuery *m_query = (AnalyticsQuery *)main_query;
596 857 : const std::vector<query_result_unit_t>& query_result =
597 : *m_query->where_info_;
598 857 : boost::shared_ptr<QueryResultMetaData> nullmetadata;
599 :
600 857 : if (m_query->is_session_query(m_query->table())
601 857 : || m_query->is_flow_query(m_query->table())) {
602 153 : QE_ASSERT(stats_.get());
603 : // can not handle query result of huge size
604 153 : if (query_result.size() > (size_t)query_result_size_limit) {
605 0 : QE_LOG(DEBUG,
606 : "Can not handle query result of size:" << query_result.size());
607 0 : QE_IO_ERROR_RETURN(0, QUERY_FAILURE);
608 : }
609 :
610 153 : if (!unroll_needed) {
611 80 : for (std::vector<query_result_unit_t>::const_iterator it = query_result.begin();
612 215 : it != query_result.end(); it++) {
613 : boost::uuids::uuid u;
614 135 : GenDb::DbDataValueVec::const_iterator itr;
615 135 : int idx = 2;
616 135 : uint8_t session_type = boost::get<uint8_t>(it->info.at(1));
617 135 : std::vector<StatsSelect::StatEntry> attribs;
618 135 : SessionTableAttributeConverter session_attribs_builder(&attribs);
619 4725 : for (itr = it->info.begin(); itr != it->info.end(); ++itr) {
620 4590 : if (idx == SessionRecordFields::SESSION_T1) {
621 135 : idx++;
622 135 : continue;
623 : }
624 4455 : if (idx == SessionRecordFields::SESSION_UUID) {
625 135 : u = boost::get<boost::uuids::uuid>(*itr);
626 135 : idx++;
627 135 : continue;
628 : }
629 4320 : const GenDb::DbDataValue &db_value(*itr);
630 4320 : boost::apply_visitor(boost::bind(session_attribs_builder, _1,
631 4320 : g_viz_constants.SessionCassTableColumns[idx]), db_value);
632 4320 : ++idx;
633 : }
634 135 : if (m_query->is_flow_query(m_query->table())) {
635 33 : StatsSelect::StatEntry direction;
636 33 : direction.name = "direction_ing";
637 33 : direction.value = (uint64_t)m_query->wherequery_->direction_ing;
638 33 : attribs.push_back(direction);
639 33 : map_session_to_flow(&attribs, session_type,
640 33 : m_query->wherequery_->direction_ing);
641 33 : }
642 135 : stats_->LoadRow(u, it->timestamp, attribs, *mresult_);
643 135 : }
644 : } else {
645 73 : uint64_t parset=0;
646 73 : uint64_t loadt=0;
647 73 : uint64_t jsont=0;
648 73 : for (std::vector<query_result_unit_t>::const_iterator it =
649 245 : query_result.begin(); it != query_result.end(); it++) {
650 : boost::uuids::uuid u;
651 99 : GenDb::DbDataValueVec::const_iterator itr;
652 99 : int idx = 2;
653 99 : uint8_t session_type = boost::get<uint8_t>(it->info.at(1));
654 99 : std::vector<StatsSelect::StatEntry> temp_attribs;
655 99 : SessionTableAttributeConverter session_attribs_builder(&temp_attribs);
656 2772 : for (itr = it->info.begin(); itr != (it->info.end() - 1); ++itr) {
657 2673 : if (g_viz_constants.SessionCassTableColumns[idx] ==
658 : SessionRecordFields::SESSION_T1) {
659 99 : idx++;
660 99 : continue;
661 : }
662 2574 : if (g_viz_constants.SessionCassTableColumns[idx] ==
663 : SessionRecordFields::SESSION_SAMPLED_FORWARD_BYTES) {
664 99 : idx += g_viz_constants.NUM_SESSION_STATS_FIELDS;
665 99 : itr += g_viz_constants.NUM_SESSION_STATS_FIELDS - 1;
666 99 : continue;
667 : }
668 2475 : if (g_viz_constants.SessionCassTableColumns[idx] ==
669 : SessionRecordFields::SESSION_UUID) {
670 99 : u = boost::get<boost::uuids::uuid>(*itr);
671 99 : idx++;
672 99 : continue;
673 : }
674 2376 : const GenDb::DbDataValue &db_value(*itr);
675 2376 : boost::apply_visitor(boost::bind(session_attribs_builder, _1,
676 2376 : g_viz_constants.SessionCassTableColumns[idx]), db_value);
677 2376 : ++idx;
678 : }
679 99 : std::string session_map(boost::get<std::string>(*itr));
680 99 : contrail_rapidjson::Document d;
681 99 : uint64_t thenj = UTCTimestampUsec();
682 198 : if (d.Parse<0>(const_cast<char *>(
683 198 : session_map.c_str())).HasParseError()) {
684 0 : QE_LOG(ERROR, "Error parsing json document: " <<
685 : d.GetParseError() << " - " << session_map);
686 0 : continue;
687 0 : }
688 99 : jsont += UTCTimestampUsec() - thenj;
689 99 : for (contrail_rapidjson::Value::ConstMemberIterator itr2 =
690 396 : d.MemberBegin(); itr2 != d.MemberEnd(); ++itr2) {
691 297 : QE_ASSERT(itr2->name.IsString());
692 297 : std::vector<StatsSelect::StatEntry> attribs;
693 297 : uint64_t thenp = UTCTimestampUsec();
694 297 : std::string ip_port(itr2->name.GetString());
695 297 : size_t delim_idx = ip_port.find(":");
696 : {
697 297 : StatsSelect::StatEntry se_client_port;
698 297 : se_client_port.name = "client_port";
699 : uint64_t cport;
700 297 : stringToInteger(ip_port.substr(0, delim_idx), cport);
701 297 : se_client_port.value = cport;
702 297 : attribs.push_back(se_client_port);
703 297 : }
704 : {
705 297 : StatsSelect::StatEntry se_remote_ip;
706 297 : se_remote_ip.name = "remote_ip";
707 297 : se_remote_ip.value = ip_port.substr(delim_idx + 1);
708 297 : attribs.push_back(se_remote_ip);
709 297 : }
710 297 : parse_json(itr2->value, &attribs);
711 297 : attribs.insert(attribs.end(), temp_attribs.begin(),
712 : temp_attribs.end());
713 297 : parset += UTCTimestampUsec() - thenp;
714 297 : uint64_t thenl = UTCTimestampUsec();
715 297 : if (m_query->is_flow_query(m_query->table())) {
716 0 : StatsSelect::StatEntry direction;
717 0 : direction.name = "direction_ing";
718 0 : direction.value = (uint64_t)m_query->wherequery_->direction_ing;
719 0 : attribs.push_back(direction);
720 0 : map_session_to_flow(&attribs, session_type,
721 0 : m_query->wherequery_->direction_ing);
722 0 : }
723 297 : stats_->LoadRow(u, it->timestamp, attribs, *mresult_);
724 297 : loadt += UTCTimestampUsec() - thenl;
725 297 : }
726 99 : }
727 73 : QE_TRACE(DEBUG, "Select ProcTime - Entries : " << query_result.size() <<
728 : " json : " << jsont << " parse : " << parset << " load : " << loadt);
729 :
730 : }
731 704 : } else if (m_query->is_stat_table_query(m_query->table())) {
732 305 : QE_ASSERT(stats_.get());
733 : // can not handle query result of huge size
734 305 : if (query_result.size() > (size_t)query_result_size_limit)
735 : {
736 0 : QE_LOG(DEBUG,
737 : "Can not handle query result of size:" << query_result.size());
738 0 : QE_IO_ERROR_RETURN(0, QUERY_FAILURE);
739 : }
740 :
741 : //uint64_t parset=0;
742 : //uint64_t loadt=0;
743 : //uint64_t jsont=0;
744 305 : for (std::vector<query_result_unit_t>::const_iterator it = query_result.begin();
745 408 : it != query_result.end(); it++) {
746 :
747 103 : string json_string;
748 103 : GenDb::DbDataValue value;
749 : boost::uuids::uuid u;
750 :
751 103 : it->get_stattable_info(json_string, u);
752 :
753 : //uint64_t thenj = UTCTimestampUsec();
754 103 : contrail_rapidjson::Document d;
755 206 : if (d.Parse<0>(const_cast<char *>(
756 206 : json_string.c_str())).HasParseError()) {
757 0 : QE_LOG(ERROR, "Error parsing json document: " <<
758 : d.GetParseError() << " - " << json_string);
759 0 : continue;
760 0 : }
761 : //jsont += UTCTimestampUsec() - thenj;
762 :
763 103 : std::vector<StatsSelect::StatEntry> attribs;
764 : {
765 103 : for (contrail_rapidjson::Value::ConstMemberIterator itr = d.MemberBegin();
766 1051 : itr != d.MemberEnd(); ++itr) {
767 948 : QE_ASSERT(itr->name.IsString());
768 :
769 : //uint64_t thenp = UTCTimestampUsec();
770 948 : std::string fvname(itr->name.GetString());
771 948 : char tname = fvname[fvname.length()-1];
772 :
773 948 : StatsSelect::StatEntry se;
774 948 : se.name = fvname.substr(0,fvname.length()-2);
775 948 : if (tname == 's') {
776 542 : se.value = itr->value.GetString();
777 542 : attribs.push_back(se);
778 406 : } else if (tname == 'n') {
779 251 : if (itr->value.IsUint()) {
780 250 : se.value = (uint64_t)itr->value.GetUint();
781 : } else {
782 1 : se.value = (uint64_t)itr->value.GetUint64();
783 : }
784 251 : attribs.push_back(se);
785 155 : } else if (tname == 'd') {
786 134 : se.value = (double) itr->value.GetDouble();
787 134 : attribs.push_back(se);
788 21 : } else if (tname == 'a') {
789 14 : std::ostringstream a_val;
790 14 : size_t i = 0;
791 : // handle list type
792 41 : for (contrail_rapidjson::Value::ConstValueIterator it =
793 41 : itr->value.Begin(); it != itr->value.End(); ++it) {
794 27 : if (i) {
795 13 : a_val << "; ";
796 : }
797 27 : QE_ASSERT(it->IsString());
798 27 : a_val << it->GetString();
799 27 : i++;
800 : }
801 14 : se.value = a_val.str();
802 14 : attribs.push_back(se);
803 21 : } else if (tname == 'm') {
804 : // handle map type
805 7 : std::ostringstream map_oss;
806 7 : map_oss << "{";
807 7 : size_t i = 0;
808 7 : for (contrail_rapidjson::Value::ConstMemberIterator it =
809 30 : itr->value.MemberBegin(); it != itr->value.MemberEnd();
810 16 : ++it) {
811 16 : QE_ASSERT(it->name.IsString() && it->value.IsString());
812 16 : StatsSelect::StatEntry entry;
813 16 : entry.name = se.name + "." + it->name.GetString();
814 16 : entry.value = it->value.GetString();
815 16 : if (i) {
816 9 : map_oss << ", ";
817 : }
818 16 : map_oss << "\"" << it->name.GetString() << "\"" << ":"
819 16 : << "\"" << it->value.GetString() << "\"";
820 16 : attribs.push_back(entry);
821 16 : i++;
822 16 : }
823 7 : map_oss << "}";
824 7 : se.value = map_oss.str();
825 7 : attribs.push_back(se);
826 7 : } else {
827 0 : QE_ASSERT(0);
828 : }
829 :
830 : //parset += UTCTimestampUsec() - thenp;
831 948 : }
832 : }
833 : //uint64_t thenl = UTCTimestampUsec();
834 103 : stats_->LoadRow(u, it->timestamp, attribs, *mresult_);
835 : //loadt += UTCTimestampUsec() - thenl;
836 103 : }
837 : //QE_TRACE(DEBUG, "Select ProcTime - Entries : " << query_result.size() <<
838 : // " json : " << jsont << " parse : " << parset << " load : " << loadt);
839 :
840 399 : } else if (m_query->table() == (g_viz_constants.OBJECT_VALUE_TABLE)) {
841 7 : uint32_t t2_start = m_query->from_time() >> g_viz_constants.RowTimeInBits;
842 7 : uint32_t t2_end = m_query->end_time() >> g_viz_constants.RowTimeInBits;
843 7 : uint32_t t1_start = m_query->from_time() & g_viz_constants.RowTimeInMask;
844 7 : uint32_t t1_end = m_query->end_time() & g_viz_constants.RowTimeInMask;
845 :
846 7 : std::vector<GenDb::DbDataValueVec> keys;
847 507 : for (uint32_t t2 = t2_start; t2 < t2_end; t2++) {
848 500 : GenDb::DbDataValueVec a_key;
849 500 : a_key.push_back(t2);
850 500 : a_key.push_back(m_query->object_value_key);
851 500 : keys.push_back(a_key);
852 500 : }
853 :
854 7 : GenDb::ColListVec mget_res;
855 7 : if (!m_query->dbif_->Db_GetMultiRow(&mget_res, g_viz_constants.OBJECT_VALUE_TABLE, keys)) {
856 0 : QE_IO_ERROR_RETURN(0, QUERY_FAILURE);
857 : }
858 :
859 7 : std::set<std::string> unique_values;
860 :
861 7 : GenDb::ColListVec::iterator first_it = mget_res.begin();
862 7 : GenDb::ColListVec::iterator last_it = mget_res.begin();
863 7 : if (mget_res.size() > 0)
864 7 : std::advance(last_it, mget_res.size()-1);
865 7 : for (GenDb::ColListVec::iterator it = mget_res.begin();
866 507 : it != mget_res.end(); it++) {
867 500 : for (GenDb::NewColVec::iterator jt = it->columns_.begin();
868 658 : jt != it->columns_.end(); jt++) {
869 161 : if (it == first_it) {
870 : uint32_t t1;
871 : try {
872 0 : t1 = boost::get<uint32_t>(jt->name->at(0));
873 0 : } catch (boost::bad_get& ex) {
874 0 : assert(0);
875 : }
876 0 : if (t1 < t1_start)
877 0 : continue;
878 : }
879 161 : if (it == last_it) {
880 : uint32_t t1;
881 : try {
882 4 : t1 = boost::get<uint32_t>(jt->name->at(0));
883 0 : } catch (boost::bad_get& ex) {
884 0 : assert(0);
885 : }
886 4 : if (t1 > t1_end)
887 3 : break;
888 : }
889 158 : std::string value;
890 : try {
891 158 : value = boost::get<std::string>(jt->value->at(0));
892 0 : } catch (boost::bad_get& ex) {
893 0 : assert(0);
894 : }
895 158 : unique_values.insert(value);
896 158 : }
897 : }
898 :
899 7 : for (std::set<std::string>::iterator it = unique_values.begin();
900 10 : it != unique_values.end(); it++) {
901 3 : std::map<std::string, std::string> cmap;
902 3 : cmap.insert(std::make_pair(g_viz_constants.OBJECT_ID, *it));
903 3 : result_->push_back(std::make_pair(cmap, nullmetadata));
904 3 : }
905 :
906 7 : } else {
907 392 : std::vector<GenDb::DbDataValueVec> keys;
908 : // can not handle query result of huge size
909 392 : if (query_result.size() > (size_t)query_result_size_limit)
910 : {
911 0 : QE_LOG(DEBUG,
912 : "Can not handle query result of size:" << query_result.size());
913 0 : QE_IO_ERROR_RETURN(0, QUERY_FAILURE);
914 : }
915 392 : QE_TRACE(DEBUG, "query_result.size():" << query_result.size());
916 :
917 392 : for (std::vector<query_result_unit_t>::const_iterator it = query_result.begin();
918 3936 : it != query_result.end(); it++) {
919 :
920 3544 : std::map<std::string, GenDb::DbDataValue> col_res_map;
921 3544 : std::string object_id = std::string();
922 81296 : for (unsigned int i = 0; i < it->info.size(); i++) {
923 :
924 77747 : std::string query_column;
925 77747 : GenDb::DbDataValue value;
926 77732 : get_query_column_value(it->info, i, &query_column, &value, &object_id);
927 : // timestamp has T2+T1 already calculated
928 77794 : if (query_column == g_viz_constants.TIMESTAMP) {
929 3544 : value = it->timestamp;
930 : }
931 77795 : col_res_map.insert(std::make_pair(query_column, value));
932 77762 : }
933 :
934 : // if select has object-id we need to make map of uuid->object-id
935 : // after T2:ObjectType: has been removed from object-id value.
936 : boost::uuids::uuid uuid_rkey;
937 3543 : std::map<boost::uuids::uuid, std::string> uuid_to_object_id;
938 3543 : if (m_query->is_object_table_query(m_query->table())) {
939 : std::map<std::string, GenDb::DbDataValue>::iterator uuid_it =
940 1998 : col_res_map.find(g_viz_constants.UUID_KEY);
941 1998 : boost::uuids::uuid u = boost::get<boost::uuids::uuid>(uuid_it->second);
942 1999 : uuid_rkey = boost::get<boost::uuids::uuid>(uuid_it->second);
943 1999 : if (object_id.empty()) {
944 : std::map<std::string, GenDb::DbDataValue>::iterator objectid_it =
945 0 : col_res_map.find(g_viz_constants.OBJECT_TYPE_NAME1);
946 0 : object_id = GenDb::DbDataValueToString(objectid_it->second);
947 : }
948 1999 : uuid_to_object_id.insert(std::make_pair(u, object_id));
949 : }
950 :
951 3542 : std::map<std::string, std::string> cmap;
952 3543 : std::vector<std::string>::iterator jt;
953 3543 : for (jt = select_column_fields.begin();
954 8134 : jt != select_column_fields.end(); jt++) {
955 4590 : std::map<std::string, GenDb::DbDataValue>::iterator kt = col_res_map.find(*jt);
956 4589 : if (kt == col_res_map.end()) {
957 2017 : if (m_query->is_object_table_query(m_query->table())) {
958 2017 : if (process_object_query_specific_select_params(
959 2017 : *jt, col_res_map, cmap, uuid_rkey,
960 2019 : uuid_to_object_id ) == false) {
961 : // Exit the loop. User is not interested
962 : // in this object log.
963 0 : break;
964 : }
965 : } else {
966 : // do not assert, append an empty string
967 0 : cmap.insert(std::make_pair(*jt, std::string("")));
968 : }
969 2571 : } else if (*jt == g_viz_constants.UUID_KEY) {
970 :
971 : boost::uuids::uuid u;
972 : try {
973 0 : u = boost::get<boost::uuids::uuid>(kt->second);
974 0 : } catch (boost::bad_get& ex) {
975 0 : QE_ASSERT(0);
976 : }
977 0 : std::string u_s(u.size(), 0);
978 0 : std::copy(u.begin(), u.end(), u_s.begin());
979 :
980 0 : cmap.insert(std::make_pair(kt->first, u_s));
981 0 : } else {
982 2571 : std::string vstr(GenDb::DbDataValueToString(kt->second));
983 2573 : cmap.insert(std::make_pair(kt->first, vstr));
984 2573 : }
985 : }
986 3544 : if (jt == select_column_fields.end()) {
987 3544 : result_->push_back(std::make_pair(cmap, nullmetadata));
988 : }
989 3543 : }
990 392 : }
991 : // Have the result ready and processing is done
992 857 : status_details = 0;
993 857 : parent_query->subquery_processed(this);
994 857 : return QUERY_SUCCESS;
995 857 : }
996 :
997 77721 : void SelectQuery::get_query_column_value(const GenDb::DbDataValueVec &info,
998 : unsigned int index,
999 : std::string *query_column,
1000 : GenDb::DbDataValue *value,
1001 : std::string *object_id) {
1002 :
1003 77721 : AnalyticsQuery *m_query = (AnalyticsQuery *)main_query;
1004 77721 : std::string columnN = MsgTableIndexToColumn(index);
1005 77605 : *query_column = MsgTableColumnToQueryColumn(columnN);
1006 77766 : *value = info.at(index);
1007 :
1008 77747 : if (value->which() != GenDb::DB_VALUE_BLANK) {
1009 : // if T2: was prepended then remove it now
1010 52838 : GenDb::ColIndexType::type index_type = MsgTableIndexToIndexType(index);
1011 52808 : if (index_type) {
1012 13974 : std::string value_str(GenDb::DbDataValueToString(*value));
1013 13979 : value_str.erase(value_str.begin(),
1014 13976 : value_str.begin() + value_str.find_first_of(":") + 1);
1015 :
1016 : // Remove "<object-type>:" from OBJECT_TYPE_NAME[1..6] values
1017 13979 : if (boost::starts_with(*query_column,
1018 : g_viz_constants.OBJECT_TYPE_NAME_PFX)) {
1019 3359 : if (boost::starts_with(value_str, m_query->table())) {
1020 1999 : value_str.erase(value_str.begin(),
1021 1999 : value_str.begin() + value_str.find_first_of(":") + 1);
1022 1999 : *object_id = value_str;
1023 5995 : QE_LOG(DEBUG, "Column " << *query_column << " Table " <<
1024 : m_query->table() << " object_id " << *object_id);
1025 : } else {
1026 1361 : value_str.erase(value_str.begin(),
1027 2722 : value_str.begin() + value_str.find_first_of(":") + 1);
1028 : }
1029 : }
1030 13979 : *value = value_str;
1031 13982 : }
1032 : }
1033 77701 : }
1034 :
1035 2018 : bool SelectQuery::process_object_query_specific_select_params(
1036 : const std::string& sel_field,
1037 : std::map<std::string, GenDb::DbDataValue>& col_res_map,
1038 : std::map<std::string, std::string>& cmap,
1039 : const boost::uuids::uuid& uuid,
1040 : std::map<boost::uuids::uuid, std::string>&
1041 : uuid_to_objectid) {
1042 2018 : std::map<std::string, GenDb::DbDataValue>::iterator cit;
1043 2018 : cit = col_res_map.find(g_viz_constants.SANDESH_TYPE);
1044 2019 : QE_ASSERT(cit != col_res_map.end());
1045 : uint32_t type_val;
1046 : try {
1047 2019 : type_val = boost::get<uint32_t>(cit->second);
1048 0 : } catch (boost::bad_get& ex) {
1049 0 : QE_ASSERT(0);
1050 : }
1051 :
1052 2019 : std::string sandesh_type;
1053 2019 : if (type_val == SandeshType::SYSTEM) {
1054 5 : sandesh_type = g_viz_constants.SYSTEM_LOG;
1055 2014 : } else if ((type_val == SandeshType::OBJECT) ||
1056 5 : (type_val == SandeshType::UVE) ||
1057 : (type_val == SandeshType::ALARM)) {
1058 2014 : sandesh_type = g_viz_constants.OBJECT_LOG;
1059 : } else {
1060 : // Ignore this message.
1061 0 : return false;
1062 : }
1063 :
1064 2019 : if (sel_field == sandesh_type) {
1065 1999 : std::map<std::string, GenDb::DbDataValue>::iterator xml_it;
1066 1999 : xml_it = col_res_map.find(g_viz_constants.DATA);
1067 1999 : QE_ASSERT(xml_it != col_res_map.end());
1068 1999 : std::string xml_data;
1069 : try {
1070 1999 : xml_data = boost::get<std::string>(xml_it->second);
1071 0 : } catch (boost::bad_get& ex) {
1072 0 : QE_ASSERT(0);
1073 : }
1074 1999 : cmap.insert(std::make_pair(sel_field, xml_data));
1075 2020 : } else if (sel_field == "ObjectId") {
1076 : // Look up the object_id corresponding to the uuid
1077 9 : std::map<boost::uuids::uuid, std::string>::iterator uuid_iter;
1078 9 : uuid_iter = uuid_to_objectid.find(uuid);
1079 9 : QE_ASSERT(uuid_iter != uuid_to_objectid.end());
1080 9 : std::string object_id_val(uuid_iter->second);
1081 9 : cmap.insert(std::make_pair(sel_field,
1082 : object_id_val));
1083 21 : } else if (is_present_in_select_column_fields(sandesh_type)) {
1084 12 : cmap.insert(std::make_pair(sel_field, std::string("")));
1085 : } else {
1086 0 : return false;
1087 : }
1088 :
1089 2020 : return true;
1090 2020 : }
1091 :
|
