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