Line data Source code
1 : /*
2 : * Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
3 : */
4 :
5 : /*
6 : * This file will not contain actual query processing code but instead only
7 : * the code to
8 : * a) Interact with external interfaces like REDIS etc.
9 : * b) Parse JSON strings passed to populate the query structures
10 : */
11 :
12 : #include "rapidjson/document.h"
13 : #include "base/logging.h"
14 : #include "query.h"
15 : #include <boost/assign/list_of.hpp>
16 : #include <boost/foreach.hpp>
17 : #include <cerrno>
18 : #include <contrail-collector/vizd_table_desc.h>
19 : #include "stats_select.h"
20 : #include "stats_query.h"
21 : #include "base/regex.h"
22 : #include "base/connection_info.h"
23 : #include "utils.h"
24 : #include <database/cassandra/cql/cql_if.h>
25 : #include <boost/make_shared.hpp>
26 : #include "qe_sandesh.h"
27 : #include <algorithm>
28 :
29 : using std::map;
30 : using std::string;
31 : using std::vector;
32 : using boost::assign::map_list_of;
33 : using boost::system::error_code;
34 : using contrail::regex;
35 : using contrail::regex_match;
36 : using contrail::regex_search;
37 : using process::ConnectionState;
38 : using process::ConnectionType;
39 : using process::ConnectionStatus;
40 :
41 : int QueryEngine::max_slice_ = 100;
42 :
43 : bool oldDataExists;
44 :
45 : typedef std::vector< std::pair<std::string, std::string> > spair_vector;
46 : static spair_vector query_string_to_column_name(0);
47 :
48 3717 : std::string get_column_name(std::string query_string)
49 : {
50 3717 : spair_vector::iterator iter;
51 :
52 3717 : for (iter = query_string_to_column_name.begin();
53 3717 : iter != query_string_to_column_name.end();
54 0 : iter++)
55 : {
56 0 : if (iter->first == query_string)
57 0 : return iter->second;
58 : }
59 :
60 3716 : return query_string;
61 : }
62 :
63 45 : std::string get_query_string(std::string column_name)
64 : {
65 45 : spair_vector::iterator iter;
66 :
67 45 : for (iter = query_string_to_column_name.begin();
68 45 : iter != query_string_to_column_name.end();
69 0 : iter++)
70 : {
71 0 : if (iter->second == column_name)
72 0 : return iter->first;
73 : }
74 :
75 45 : return column_name;
76 : }
77 :
78 0 : QueryResultMetaData::~QueryResultMetaData() {
79 0 : }
80 :
81 2005 : PostProcessingQuery::PostProcessingQuery(
82 : const std::map<std::string, std::string>& json_api_data,
83 2005 : QueryUnit *main_query) : QueryUnit(main_query, main_query),
84 2005 : sorted(false), limit(0) {
85 2005 : AnalyticsQuery *m_query = (AnalyticsQuery *)main_query;
86 2005 : std::map<std::string, std::string>::const_iterator iter;
87 :
88 2005 : QE_TRACE(DEBUG, __func__ );
89 :
90 2005 : json_string_ = "";
91 :
92 19087 : for (iter = json_api_data.begin(); iter != json_api_data.end(); iter++)
93 : {
94 17081 : if (iter->first == QUERY_SORT_OP)
95 : {
96 126 : sorted = true;
97 : int tmp;
98 126 : std::istringstream(iter->second) >> tmp;
99 126 : sorting_type = (sort_op)tmp;
100 126 : m_query->merge_needed = true;
101 126 : QE_TRACE(DEBUG, "sorting_type :" << sorting_type);
102 126 : json_string_ += iter->second;
103 126 : json_string_ += " ";
104 : }
105 :
106 17081 : if (iter->first == QUERY_LIMIT)
107 : {
108 85 : std::istringstream(iter->second) >> limit;
109 85 : m_query->merge_needed = true;
110 85 : QE_TRACE(DEBUG, "limit :"<< limit);
111 85 : json_string_ += iter->second;
112 85 : json_string_ += " ";
113 : }
114 :
115 17083 : if (iter->first == QUERY_SORT_FIELDS)
116 : {
117 126 : contrail_rapidjson::Document d;
118 0 : std::string json_string = "{ \"sort_fields\" : " +
119 126 : iter->second + " }";
120 126 : json_string_ += json_string;
121 126 : json_string_ += " ";
122 :
123 126 : d.Parse<0>(const_cast<char *>(json_string.c_str()));
124 : const contrail_rapidjson::Value& json_sort_fields =
125 126 : d["sort_fields"];
126 126 : QE_PARSE_ERROR(json_sort_fields.IsArray());
127 126 : QE_TRACE(DEBUG, "# of sort fields:"<< json_sort_fields.Size());
128 252 : for (contrail_rapidjson::SizeType i = 0; i<json_sort_fields.Size(); i++)
129 : {
130 126 : QE_PARSE_ERROR(json_sort_fields[i].IsString());
131 126 : std::string sort_str(json_sort_fields[i].GetString());
132 126 : QE_TRACE(DEBUG, "sort field:" << sort_str);
133 126 : std::string datatype(m_query->get_column_field_datatype(sort_str));
134 126 : if (m_query->is_stat_table_query(m_query->table()) &&
135 0 : (m_query->stats().is_stat_table_static())) {
136 : // This is a static StatTable. We can check the schema
137 0 : std::string sfield;
138 :
139 : // If this is an agg field, check underlying data type
140 0 : if (StatsQuery::ParseAgg(sort_str, sfield) !=
141 : QEOpServerProxy::INVALID) {
142 0 : std::string dtype2(m_query->get_column_field_datatype(sfield));
143 0 : QE_INVALIDARG_ERROR(dtype2 != std::string(""));
144 0 : } else {
145 0 : QE_INVALIDARG_ERROR(datatype != std::string(""));
146 : }
147 0 : }
148 126 : QE_INVALIDARG_ERROR(
149 : m_query->is_valid_sort_field(sort_str) != false);
150 252 : sort_field_t sort_field(get_column_name(sort_str), datatype);
151 126 : sort_fields.push_back(sort_field);
152 126 : }
153 126 : }
154 :
155 : /*
156 : * old filter style is just list of expr ANDed
157 : * new filter are list of ANDs over OR
158 : * both modes are supported with the below code
159 : */
160 17082 : if (iter->first == QUERY_FILTER) {
161 407 : contrail_rapidjson::Document d;
162 0 : std::string json_string = "{ \"filter\" : " +
163 407 : iter->second + " }";
164 407 : json_string_ += json_string;
165 407 : json_string_ += " ";
166 :
167 407 : d.Parse<0>(const_cast<char *>(json_string.c_str()));
168 : const contrail_rapidjson::Value& json_filters =
169 408 : d["filter"];
170 407 : QE_PARSE_ERROR(json_filters.IsArray());
171 407 : QE_TRACE(DEBUG, "# of filters:"<< json_filters.Size());
172 408 : bool single_list = false;
173 408 : if (json_filters.Size()) {
174 408 : contrail_rapidjson::SizeType zeroth = 0;
175 408 : const contrail_rapidjson::Value& json_filters_0 = json_filters[zeroth];
176 408 : if (!json_filters_0.IsArray()) {
177 374 : single_list = true;
178 : }
179 : }
180 :
181 408 : if (single_list) {
182 : //parse the old format
183 374 : std::vector<filter_match_t> filter_and;
184 748 : for (contrail_rapidjson::SizeType j = 0; j<json_filters.Size(); j++)
185 : {
186 374 : filter_match_t filter;
187 374 : QE_PARSE_ERROR((json_filters[j].HasMember(WHERE_MATCH_NAME)
188 : && json_filters[j].HasMember(WHERE_MATCH_VALUE)
189 : && json_filters[j].HasMember(WHERE_MATCH_OP)));
190 : const contrail_rapidjson::Value& name_value =
191 374 : json_filters[j][WHERE_MATCH_NAME];
192 : const contrail_rapidjson::Value& value_value =
193 374 : json_filters[j][WHERE_MATCH_VALUE];
194 : const contrail_rapidjson::Value& op_value =
195 374 : json_filters[j][WHERE_MATCH_OP];
196 :
197 : // do some validation checks
198 374 : QE_INVALIDARG_ERROR(name_value.IsString());
199 374 : QE_INVALIDARG_ERROR
200 : ((value_value.IsString() || value_value.IsNumber() ||
201 : value_value.IsDouble()));
202 374 : QE_INVALIDARG_ERROR(op_value.IsNumber());
203 :
204 374 : filter.name = name_value.GetString();
205 374 : filter.op = (match_op)op_value.GetInt();
206 :
207 : // extract value after type conversion
208 : {
209 374 : if (value_value.IsString())
210 : {
211 136 : filter.value = value_value.GetString();
212 238 : } else if (value_value.IsInt()){
213 : int int_value;
214 220 : std::ostringstream convert;
215 220 : int_value = value_value.GetInt();
216 220 : convert << int_value;
217 220 : filter.value = convert.str();
218 238 : } else if (value_value.IsUint()) {
219 : uint32_t uint_value;
220 0 : std::ostringstream convert;
221 0 : uint_value = value_value.GetUint();
222 0 : convert << uint_value;
223 0 : filter.value = convert.str();
224 18 : } else if (value_value.IsDouble()) {
225 : double dbl_value;
226 18 : std::ostringstream convert;
227 18 : dbl_value = value_value.GetDouble();
228 18 : convert << dbl_value;
229 18 : filter.value = convert.str();
230 18 : }
231 : }
232 :
233 374 : if (filter.op == REGEX_MATCH)
234 : {
235 : // compile regex beforehand
236 302 : filter.match_e = regex(filter.value);
237 : }
238 :
239 374 : filter_and.push_back(filter);
240 373 : }
241 374 : filter_list.push_back(filter_and);
242 374 : } else {
243 : //new OR of ANDs
244 85 : for (contrail_rapidjson::SizeType j = 0; j<json_filters.Size(); j++) {
245 51 : std::vector<filter_match_t> filter_and;
246 51 : const contrail_rapidjson::Value& json_filter_and = json_filters[j];
247 51 : QE_PARSE_ERROR(json_filter_and.IsArray());
248 :
249 102 : for (contrail_rapidjson::SizeType k = 0; k<json_filter_and.Size(); k++) {
250 51 : filter_match_t filter;
251 51 : QE_PARSE_ERROR((
252 : json_filter_and[k].HasMember(WHERE_MATCH_NAME)
253 : && json_filter_and[k].HasMember(WHERE_MATCH_VALUE)
254 : && json_filter_and[k].HasMember(WHERE_MATCH_OP)));
255 : const contrail_rapidjson::Value& name_value =
256 51 : json_filter_and[k][WHERE_MATCH_NAME];
257 : const contrail_rapidjson::Value& value_value =
258 51 : json_filter_and[k][WHERE_MATCH_VALUE];
259 : const contrail_rapidjson::Value& op_value =
260 51 : json_filter_and[k][WHERE_MATCH_OP];
261 :
262 : // do some validation checks
263 51 : QE_INVALIDARG_ERROR(name_value.IsString());
264 51 : QE_INVALIDARG_ERROR
265 : ((value_value.IsString() || value_value.IsNumber()));
266 51 : QE_INVALIDARG_ERROR(op_value.IsNumber());
267 :
268 51 : filter.name = name_value.GetString();
269 51 : filter.op = (match_op)op_value.GetInt();
270 :
271 : // extract value after type conversion
272 51 : if (value_value.IsString()) {
273 51 : filter.value = value_value.GetString();
274 0 : } else if (value_value.IsInt()) {
275 : int int_value;
276 0 : std::ostringstream convert;
277 0 : int_value = value_value.GetInt();
278 0 : convert << int_value;
279 0 : filter.value = convert.str();
280 0 : } else if (value_value.IsUint()) {
281 : uint32_t uint_value;
282 0 : std::ostringstream convert;
283 0 : uint_value = value_value.GetUint();
284 0 : convert << uint_value;
285 0 : filter.value = convert.str();
286 0 : }
287 :
288 51 : if (filter.op == REGEX_MATCH) {
289 : // compile regex beforehand
290 0 : filter.match_e = regex(filter.value);
291 : }
292 :
293 51 : filter_and.push_back(filter);
294 51 : }
295 51 : filter_list.push_back(filter_and);
296 51 : }
297 : }
298 408 : }
299 : }
300 :
301 2005 : if (!m_query->wherequery_->filter_list_.empty()) {
302 0 : if (filter_list.empty()) {
303 0 : filter_list = m_query->wherequery_->filter_list_;
304 : } else {
305 0 : BOOST_FOREACH(std::vector<filter_match_t> &filter_and, filter_list) {
306 0 : BOOST_FOREACH(const std::vector<filter_match_t> &where_filter_and,
307 : m_query->wherequery_->filter_list_) {
308 0 : filter_and.insert(filter_and.end(), where_filter_and.begin(),
309 : where_filter_and.end());
310 : }
311 : }
312 : }
313 : }
314 :
315 : // add filter to filter query engine logs if requested
316 3830 : if (((AnalyticsQuery *)main_query)->filter_qe_logs &&
317 1825 : ((AnalyticsQuery *)main_query)->is_message_table_query()) {
318 355 : QE_TRACE(DEBUG, " Adding filter for QE logs");
319 355 : filter_match_t filter;
320 355 : filter.name = g_viz_constants.MODULE;
321 : filter.value =
322 355 : ((AnalyticsQuery *)main_query)->sandesh_moduleid;
323 355 : filter.op = NOT_EQUAL;
324 355 : filter.ignore_col_absence = true;
325 355 : if (!filter_list.size()) {
326 177 : std::vector<filter_match_t> filter_and;
327 177 : filter_and.push_back(filter);
328 177 : filter_list.push_back(filter_and);
329 177 : } else {
330 373 : for (unsigned int i = 0; i < filter_list.size(); i++) {
331 195 : filter_list[i].push_back(filter);
332 : }
333 : }
334 355 : }
335 :
336 : // If the user has specified the sorting field and not the sorting order,
337 : // then sort the result in ascending order.
338 2005 : if (sort_fields.size() && sorted == false) {
339 0 : sorted = true;
340 0 : sorting_type = ASCENDING;
341 : }
342 0 : }
343 :
344 32 : bool AnalyticsQuery::merge_processing(
345 : const QEOpServerProxy::BufferT& input,
346 : QEOpServerProxy::BufferT& output) {
347 :
348 32 : if (status_details != 0)
349 : {
350 0 : QE_TRACE(DEBUG,
351 : "No need to process query, as there were errors previously");
352 0 : return false;
353 : }
354 :
355 : // Have the result ready and processing is done
356 32 : status_details = 0;
357 32 : return postprocess_->merge_processing(input, output);
358 : }
359 :
360 4 : bool AnalyticsQuery::final_merge_processing(
361 : const std::vector<boost::shared_ptr<QEOpServerProxy::BufferT> >& inputs,
362 : QEOpServerProxy::BufferT& output) {
363 :
364 4 : if (status_details != 0)
365 : {
366 0 : QE_TRACE(DEBUG,
367 : "No need to process query, as there were errors previously");
368 0 : return false;
369 : }
370 :
371 : // Have the result ready and processing is done
372 4 : status_details = 0;
373 4 : return postprocess_->final_merge_processing(inputs, output);
374 : }
375 :
376 : // this is to get parallelization details once the query is parsed
377 127 : void AnalyticsQuery::get_query_details(bool& is_merge_needed, bool& is_map_output,
378 : std::vector<uint64_t>& chunk_sizes,
379 : std::string& where, uint32_t& wterms,
380 : std::string& select,
381 : std::string& post,
382 : uint64_t& time_period,
383 : int& parse_status)
384 : {
385 127 : QE_TRACE(DEBUG, "time_slice is " << time_slice);
386 127 : if (status_details == 0)
387 : {
388 127 : for (uint64_t chunk_start = original_from_time;
389 991 : chunk_start < original_end_time; chunk_start += time_slice)
390 : {
391 864 : if ((chunk_start+time_slice) <= original_end_time) {
392 744 : chunk_sizes.push_back(time_slice);
393 : } else {
394 120 : chunk_sizes.push_back((original_end_time - chunk_start));
395 : }
396 : }
397 : } else {
398 0 : chunk_sizes.push_back(0); // just return some dummy value
399 : }
400 :
401 127 : time_period = (end_time_ - from_time_) / 1000000;
402 :
403 127 : parse_status = status_details;
404 127 : if (parse_status != 0) return;
405 :
406 127 : if (is_stat_table_query(table_)
407 88 : || is_session_query(table_)
408 215 : || is_flow_query(table_)) {
409 71 : is_merge_needed = selectquery_->stats_->IsMergeNeeded();
410 : } else {
411 56 : is_merge_needed = merge_needed;
412 : }
413 :
414 127 : where = wherequery_->json_string_;
415 127 : wterms = wherequery_->wterms_;
416 127 : select = selectquery_->json_string_;
417 127 : post = postprocess_->json_string_;
418 254 : is_map_output = is_stat_table_query(table_)
419 88 : || is_session_query(table_)
420 215 : || is_flow_query(table_);
421 : }
422 :
423 2005 : bool AnalyticsQuery::can_parallelize_query() {
424 2005 : parallelize_query_ = true;
425 2005 : if (table_ == g_viz_constants.OBJECT_VALUE_TABLE) {
426 21 : parallelize_query_ = false;
427 : }
428 2005 : return parallelize_query_;
429 : }
430 :
431 : /* parse the stat name and attribute which can be used
432 : * to collect stats. Table name is of the format,
433 : * StatTable.TableName.AttrName
434 : * The functions sets the stat_name_attr member, which is later
435 : * used to collect stats information
436 : */
437 710 : void AnalyticsQuery::ParseStatName(std::string& stat_table_name) {
438 710 : string stat_table("StatTable.");
439 709 : stat_name_attr = stat_table_name.substr(stat_table.length());
440 710 : std::replace(stat_name_attr.begin(), stat_name_attr.end(), '.', ':');
441 710 : }
442 :
443 2019 : void AnalyticsQuery::Init(const std::string& qid,
444 : const std::map<std::string, std::string>& json_api_data,
445 : int32_t or_number)
446 : {
447 2019 : std::map<std::string, std::string>::const_iterator iter;
448 :
449 2019 : QE_TRACE(DEBUG, __func__);
450 :
451 : // populate fields
452 2021 : query_id = qid;
453 :
454 : sandesh_moduleid =
455 2021 : g_vns_constants.ModuleNames.find(Module::QUERY_ENGINE)->second;
456 :
457 : {
458 2021 : std::stringstream json_string; json_string << " { ";
459 2021 : for (std::map<std::string,
460 2021 : std::string>::iterator it = json_api_data_.begin();
461 19083 : it != json_api_data_.end(); it++) {
462 : json_string <<
463 34123 : ((it != json_api_data_.begin())? " , " : "") <<
464 17061 : it->first << ": " << it->second;
465 : }
466 2020 : json_string << " } ";
467 6465 : QE_LOG_GLOBAL(DEBUG, "json query is: " << json_string.str());
468 2020 : }
469 :
470 : // parse JSON query
471 : // FROM field
472 : {
473 2020 : iter = json_api_data.find(QUERY_TABLE);
474 2052 : QE_PARSE_ERROR(iter != json_api_data.end());
475 :
476 : //strip " from the passed string
477 2004 : table_ = iter->second.substr(1, iter->second.size()-2);
478 :
479 : // boost::to_upper(table);
480 2004 : QE_TRACE(DEBUG, " table is " << table_);
481 2005 : if (is_stat_table_query(table_)) {
482 710 : stats_.reset(new StatsQuery(table_));
483 710 : ParseStatName(table_);
484 : }
485 : }
486 :
487 : uint64_t ttl;
488 2005 : uint64_t min_start_time = UTCTimestampUsec();
489 2005 : uint64_t max_end_time = min_start_time;
490 :
491 2005 : if (is_stat_table_query(table_)) {
492 710 : ttl = ttlmap_.find(TtlType::STATSDATA_TTL)->second;
493 1295 : } else if (is_flow_query(table_) || is_session_query(table_)) {
494 365 : ttl = ttlmap_.find(TtlType::FLOWDATA_TTL)->second;
495 930 : } else if (is_object_table_query(table_)) {
496 395 : ttl = ttlmap_.find(TtlType::CONFIGAUDIT_TTL)->second;
497 : } else {
498 535 : ttl = ttlmap_.find(TtlType::GLOBAL_TTL)->second;
499 : }
500 2004 : min_start_time = min_start_time-ttl*60*60*1000000;
501 :
502 : // Start time
503 : {
504 2004 : iter = json_api_data.find(QUERY_START_TIME);
505 2003 : QE_PARSE_ERROR(iter != json_api_data.end());
506 2003 : QE_PARSE_ERROR(parse_time(iter->second, &req_from_time_));
507 2005 : QE_TRACE(DEBUG, " from_time is " << req_from_time_);
508 2005 : if (req_from_time_ < min_start_time)
509 : {
510 1 : from_time_ = min_start_time;
511 1 : QE_TRACE(DEBUG, "updated start_time to:" << from_time_);
512 : } else {
513 2004 : from_time_ = req_from_time_;
514 : }
515 : }
516 :
517 : // End time
518 : {
519 2005 : iter = json_api_data.find(QUERY_END_TIME);
520 2005 : QE_PARSE_ERROR(iter != json_api_data.end());
521 2005 : QE_PARSE_ERROR(parse_time(iter->second, &req_end_time_));
522 2005 : QE_TRACE(DEBUG, " end_time is " << req_end_time_);
523 :
524 2005 : if (req_end_time_ > max_end_time) {
525 70 : end_time_ = max_end_time;
526 70 : QE_TRACE(DEBUG, "updated end_time to:" << end_time_);
527 : } else {
528 1935 : end_time_ = req_end_time_;
529 : }
530 : }
531 :
532 2005 : if (is_stat_fieldnames_table_query(table_)) {
533 37 : uint64_t time_period = (end_time_ - from_time_); /* in usec */
534 37 : uint64_t cache_time = (1 << (g_viz_constants.RowTimeInBits +
535 37 : g_viz_constants.CacheTimeInAdditionalBits));
536 37 : if (time_period < cache_time) {
537 19 : uint64_t diff_time_usec = (cache_time - time_period);
538 19 : from_time_ = from_time_ - diff_time_usec;
539 19 : if (from_time_ < min_start_time) {
540 0 : from_time_ = min_start_time;
541 : }
542 : }
543 : }
544 :
545 : // Initialize SELECT/WHERE/Post-Processing components of query
546 : // for input validation
547 :
548 : // where processing initialization
549 2005 : std::string where_json_string;
550 : {
551 2005 : int direction = INGRESS;
552 2005 : iter = json_api_data.find(QUERY_FLOW_DIR);
553 2004 : if (iter != json_api_data.end()) {
554 0 : std::istringstream(iter->second) >> direction;
555 0 : QE_TRACE(DEBUG, "set flow direction to:" << direction);
556 : }
557 :
558 2004 : int is_si = 0;
559 2004 : iter = json_api_data.find(QUERY_SESSION_IS_SI);
560 2003 : if (iter != json_api_data.end()) {
561 329 : std::istringstream(iter->second) >> is_si;
562 329 : QE_TRACE(DEBUG, "set session is_si to:" << is_si);
563 : }
564 :
565 2003 : int session_type = 0;
566 2003 : iter = json_api_data.find(QUERY_SESSION_TYPE);
567 2003 : if (iter != json_api_data.end()) {
568 329 : if (iter->second == "\"client\"") {
569 306 : session_type = 1;
570 23 : } else if (iter->second == "\"server\"") {
571 23 : session_type = 0;
572 : } else {
573 0 : QE_INVALIDARG_ERROR(false && "session_type_invalid");
574 : }
575 329 : QE_TRACE(DEBUG, "set session is_si to:" << session_type);
576 : }
577 1674 : else if (is_session_query(table_)) {
578 0 : QE_LOG_GLOBAL(ERROR, "session_type is required for session queries");
579 0 : this->status_details = -1;
580 0 : return;
581 : }
582 :
583 2003 : iter = json_api_data.find(QUERY_WHERE);
584 2003 : if (iter == json_api_data.end())
585 : {
586 349 : QE_TRACE(DEBUG, "Where * query");
587 349 : where_json_string = std::string("");
588 : } else {
589 1654 : where_json_string = iter->second;
590 : }
591 :
592 2004 : QE_TRACE(DEBUG, " Initializing Where Query");
593 2005 : wherequery_ = new WhereQuery(where_json_string, session_type,
594 2005 : is_si, direction, or_number, this);
595 2005 : this->status_details = wherequery_->status_details;
596 2005 : if (this->status_details != 0 )
597 : {
598 0 : QE_LOG_GLOBAL(DEBUG, "Error in WHERE parsing");
599 0 : return;
600 : }
601 : }
602 :
603 : // select processing initialization
604 : {
605 2005 : QE_TRACE(DEBUG, " Initializing Select Query");
606 2005 : selectquery_ = new SelectQuery(this, json_api_data);
607 2005 : this->status_details = selectquery_->status_details;
608 2005 : if (this->status_details != 0 )
609 : {
610 0 : QE_LOG_GLOBAL(DEBUG, "Error in SELECT parsing");
611 0 : return;
612 : }
613 : /*
614 : * ObjectId queries are special, they are requested from Object* tables,
615 : * but the values are extrated from g_viz_constants.OBJECT_VALUE_TABLE
616 : */
617 2005 : if (is_object_table_query(table_)) {
618 395 : if (selectquery_->ObjectIdQuery()) {
619 21 : object_value_key = table_;
620 21 : table_ = g_viz_constants.OBJECT_VALUE_TABLE;
621 : }
622 : }
623 : }
624 :
625 : // post processing initialization
626 2005 : QE_TRACE(DEBUG, " Initializing PostProcessing Query");
627 2005 : postprocess_ = new PostProcessingQuery(json_api_data, this);
628 2005 : this->status_details = postprocess_->status_details;
629 2005 : if (this->status_details != 0 )
630 : {
631 0 : QE_LOG_GLOBAL(DEBUG, "Error in PostProcess parsing");
632 0 : return;
633 : }
634 :
635 2005 : if (is_stat_table_query(table_)
636 1295 : || is_session_query(table_)
637 3300 : || is_flow_query(table_)) {
638 1075 : selectquery_->stats_->SetSortOrder(postprocess_->sort_fields);
639 : }
640 :
641 : // just to take care of issues with Analytics start time
642 2005 : if (from_time_ > end_time_)
643 1 : from_time_ = end_time_ - 1;
644 :
645 : // Get the right job slice for parallelization
646 2005 : original_from_time = from_time_;
647 2005 : original_end_time = end_time_;
648 :
649 2005 : if (can_parallelize_query()) {
650 1984 : uint64_t smax = pow(2,g_viz_constants.RowTimeInBits) * \
651 1983 : QueryEngine::max_slice_;
652 :
653 1983 : time_slice = ((end_time_ - from_time_)/total_parallel_batches) + 1;
654 :
655 1983 : if (time_slice < (uint64_t)pow(2,g_viz_constants.RowTimeInBits)) {
656 417 : time_slice = pow(2,g_viz_constants.RowTimeInBits);
657 : }
658 1984 : if (time_slice > smax) {
659 0 : time_slice = smax;
660 : }
661 1984 : QE_TRACE(DEBUG, "time_slice:" << time_slice << " , # of parallel "
662 : "batches:" << total_parallel_batches);
663 :
664 : } else {
665 : // No parallelization
666 84 : QE_LOG_GLOBAL(DEBUG, "No parallelization for this query");
667 21 : merge_needed = false;
668 21 : parallelize_query_ = false;
669 21 : time_slice = end_time_ - from_time_;
670 : }
671 :
672 2005 : from_time_ =
673 2005 : original_from_time + time_slice*parallel_batch_num;
674 2005 : end_time_ = from_time_ + time_slice;
675 2005 : if (from_time_ >= original_end_time)
676 : {
677 0 : processing_needed = false;
678 2005 : } else if (end_time_ > original_end_time) {
679 248 : end_time_ = original_end_time;
680 : }
681 :
682 2005 : if (processing_needed)
683 : {
684 : // change it to trace later TBD
685 2005 : QE_TRACE(DEBUG, "For batch:" << parallel_batch_num << " from_time:" << from_time_ << " end_time:" << end_time_ << " time slice:" << time_slice);
686 : } else {
687 0 : QE_TRACE(DEBUG, "No processing needed for batch:" << parallel_batch_num);
688 : }
689 :
690 2005 : }
691 12151 : QueryUnit::QueryUnit(QueryUnit *p_query, QueryUnit *m_query):
692 12147 : parent_query(p_query), main_query(m_query), pending_subqueries(0),
693 12151 : query_status(QUERY_PROCESSING_NOT_STARTED), status_details(0)
694 : {
695 12143 : if (p_query)
696 10129 : p_query->sub_queries.push_back(this);
697 12137 : };
698 :
699 12159 : QueryUnit::~QueryUnit()
700 : {
701 12159 : int num_sub_queries = sub_queries.size();
702 22297 : for(int i = 0; i<num_sub_queries; i++)
703 10137 : delete sub_queries[i];
704 12160 : }
705 :
706 :
707 : // Get UUID from the info field
708 0 : void query_result_unit_t::get_uuid(boost::uuids::uuid& u) const
709 : {
710 : try {
711 0 : u = boost::get<boost::uuids::uuid>(info.at(0));
712 0 : } catch (boost::bad_get& ex) {
713 0 : QE_ASSERT(0);
714 : }
715 0 : }
716 :
717 220 : void query_result_unit_t::set_stattable_info(
718 : const std::string& attribstr,
719 : const boost::uuids::uuid& uuid) {
720 220 : info.push_back(attribstr);
721 220 : info.push_back(uuid);
722 220 : }
723 :
724 1 : void query_result_unit_t::get_objectid(std::string& object_id) const {
725 : try {
726 1 : object_id = boost::get<std::string>(info.at(1));
727 0 : } catch (boost::bad_get& ex) {
728 0 : QE_ASSERT(0);
729 0 : } catch (const std::out_of_range& oor) {
730 0 : QE_ASSERT(0);
731 : }
732 1 : }
733 :
734 103 : void query_result_unit_t::get_stattable_info(
735 : std::string& attribstr,
736 : boost::uuids::uuid& uuid) const {
737 :
738 103 : int index = 0;
739 :
740 : try {
741 103 : attribstr = boost::get<std::string>(info.at(index++));
742 0 : } catch (boost::bad_get& ex) {
743 0 : QE_ASSERT(0);
744 0 : } catch (const std::out_of_range& oor) {
745 0 : QE_ASSERT(0);
746 : }
747 :
748 : try {
749 103 : uuid = boost::get<boost::uuids::uuid>(info.at(index++));
750 0 : } catch (boost::bad_get& ex) {
751 0 : QE_ASSERT(0);
752 0 : } catch (const std::out_of_range& oor) {
753 0 : QE_ASSERT(0);
754 : }
755 :
756 103 : }
757 :
758 865 : query_status_t AnalyticsQuery::process_query()
759 : {
760 865 : if (status_details != 0)
761 : {
762 0 : QE_TRACE(DEBUG,
763 : "No need to process query, as there were errors previously");
764 0 : return QUERY_FAILURE;
765 : }
766 :
767 865 : QE_TRACE(DEBUG, "Start Select Processing");
768 865 : select_start_ = UTCTimestampUsec();
769 865 : query_status = selectquery_->process_query();
770 865 : status_details = selectquery_->status_details;
771 865 : qperf_.chunk_select_time =
772 865 : static_cast<uint32_t>((UTCTimestampUsec() - select_start_)/1000);
773 :
774 865 : if (query_status != QUERY_SUCCESS)
775 : {
776 0 : QE_LOG(DEBUG,
777 : "select processing failed with error:"<< query_status);
778 0 : return query_status;
779 : }
780 865 : QE_TRACE(DEBUG, "End Select Processing. row #s:" <<
781 : selectquery_->result_->size());
782 865 : QE_TRACE(DEBUG, "Start PostProcessing");
783 865 : postproc_start_ = UTCTimestampUsec();
784 865 : query_status = postprocess_->process_query();
785 865 : status_details = postprocess_->status_details;
786 865 : qperf_.chunk_postproc_time =
787 865 : static_cast<uint32_t>((UTCTimestampUsec() - postproc_start_)/1000);
788 :
789 865 : final_result = std::move(postprocess_->result_);
790 865 : final_mresult = std::move(postprocess_->mresult_);
791 865 : if (query_status != QUERY_SUCCESS)
792 : {
793 0 : QE_LOG(DEBUG,
794 : "post processing failed with error:"<< query_status);
795 0 : return query_status;
796 : }
797 865 : QE_TRACE(DEBUG, "End PostProcessing. row #s:" <<
798 : final_result->size());
799 865 : return QUERY_SUCCESS;
800 : }
801 :
802 0 : AnalyticsQuery::AnalyticsQuery(const std::string& qid, std::map<std::string,
803 : std::string>& json_api_data,
804 : int or_number,
805 : const std::vector<query_result_unit_t> * where_info,
806 : const TtlMap& ttlmap,
807 : EventManager *evm, std::vector<std::string> cassandra_ips,
808 : std::vector<int> cassandra_ports, int batch,
809 : int total_batches, const std::string& cassandra_user,
810 : const std::string& cassandra_password,
811 : QueryEngine* qe,
812 0 : void *handle):
813 : QueryUnit(NULL, this),
814 0 : filter_qe_logs(true),
815 0 : json_api_data_(json_api_data),
816 0 : where_info_(where_info),
817 0 : ttlmap_(ttlmap),
818 0 : where_start_(0),
819 0 : select_start_(0),
820 0 : postproc_start_(0),
821 0 : merge_needed(false),
822 0 : parallel_batch_num(batch),
823 0 : total_parallel_batches(total_batches),
824 0 : processing_needed(true),
825 0 : qe_(qe),
826 0 : handle_(handle),
827 0 : stats_(nullptr)
828 : {
829 0 : assert(dbif_ != NULL);
830 : // Need to do this for logging/tracing with query ids
831 0 : query_id = qid;
832 :
833 0 : QE_TRACE(DEBUG, __func__);
834 :
835 : // Initialize database connection
836 0 : QE_TRACE(DEBUG, "Initializing database");
837 :
838 0 : boost::system::error_code ec;
839 0 : if (!dbif_->Db_Init()) {
840 0 : QE_LOG(ERROR, "Database initialization failed");
841 0 : this->status_details = EIO;
842 : }
843 :
844 0 : if (!dbif_->Db_SetTablespace(qe_->keyspace())) {
845 0 : QE_LOG(ERROR, ": Create/Set KEYSPACE: " <<
846 : g_viz_constants.COLLECTOR_KEYSPACE << " FAILED");
847 0 : this->status_details = EIO;
848 : }
849 0 : for (std::vector<GenDb::NewCf>::const_iterator it = vizd_tables.begin();
850 0 : it != vizd_tables.end(); it++) {
851 0 : if (!dbif_->Db_UseColumnfamily(*it)) {
852 0 : QE_LOG(ERROR, "Database initialization:Db_UseColumnfamily failed");
853 0 : this->status_details = EIO;
854 : }
855 : }
856 0 : for (std::vector<GenDb::NewCf>::const_iterator it = vizd_stat_tables.begin();
857 0 : it != vizd_stat_tables.end(); it++) {
858 0 : if (!dbif_->Db_UseColumnfamily(*it)) {
859 0 : QE_LOG(ERROR, "Database initialization:Db_UseColumnfamily failed");
860 0 : this->status_details = EIO;
861 : }
862 : }
863 0 : for (std::vector<GenDb::NewCf>::const_iterator it = vizd_session_tables.begin();
864 0 : it != vizd_session_tables.end(); it++) {
865 0 : if (!dbif_->Db_UseColumnfamily(*it)) {
866 0 : QE_LOG(ERROR, "Database initialization:Db_UseColumnfamily failed");
867 0 : this->status_details = EIO;
868 : }
869 : }
870 0 : if (this->status_details != 0) {
871 : // Update connection info
872 0 : ConnectionState::GetInstance()->Update(ConnectionType::DATABASE,
873 0 : std::string(), ConnectionStatus::DOWN, dbif_->Db_GetEndpoints(),
874 0 : std::string());
875 : } else {
876 : // Update connection info
877 0 : ConnectionState::GetInstance()->Update(ConnectionType::DATABASE,
878 0 : std::string(), ConnectionStatus::UP, dbif_->Db_GetEndpoints(),
879 0 : std::string());
880 : }
881 0 : dbif_->Db_SetInitDone(true);
882 0 : Init(qid, json_api_data, or_number);
883 0 : }
884 :
885 2020 : AnalyticsQuery::AnalyticsQuery(const std::string& qid,
886 : GenDbIfPtr dbif_ptr,
887 : std::map<std::string, std::string> json_api_data,
888 : int or_number,
889 : const std::vector<query_result_unit_t> * where_info,
890 : const TtlMap &ttlmap, int batch, int total_batches,
891 : QueryEngine* qe,
892 2020 : void *handle) :
893 : QueryUnit(NULL, this),
894 2019 : dbif_(dbif_ptr),
895 2020 : query_id(qid),
896 2019 : filter_qe_logs(true),
897 2020 : json_api_data_(json_api_data),
898 2018 : where_info_(where_info),
899 2018 : ttlmap_(ttlmap),
900 2018 : where_start_(0),
901 2018 : select_start_(0),
902 2018 : postproc_start_(0),
903 2018 : merge_needed(false),
904 2018 : parallel_batch_num(batch),
905 2018 : total_parallel_batches(total_batches),
906 2018 : processing_needed(true),
907 2018 : qe_(qe),
908 2018 : handle_(handle),
909 6057 : stats_(nullptr) {
910 2018 : Init(qid, json_api_data, or_number);
911 2021 : }
912 :
913 19 : QueryEngine::QueryEngine(EventManager *evm,
914 : vector<string> redis_ip_ports,
915 : const std::string & redis_password,
916 : const bool redis_ssl_enable,
917 : const std::string & redis_keyfile,
918 : const std::string & redis_certfile,
919 : const std::string & redis_ca_cert,
920 : int max_tasks, int max_slice,
921 : const std::string & cassandra_user,
922 : const std::string & cassandra_password,
923 : bool cassandra_use_ssl,
924 : const std::string & cassandra_ca_certs,
925 19 : const std::string &host_ip) :
926 38 : qosp_(new QEOpServerProxy(evm,
927 : this, redis_ip_ports, redis_password, redis_ssl_enable, redis_keyfile,
928 19 : redis_certfile, redis_ca_cert, host_ip, max_tasks)),
929 19 : evm_(evm),
930 19 : cassandra_ports_(0),
931 19 : cassandra_user_(cassandra_user),
932 19 : cassandra_password_(cassandra_password),
933 19 : cassandra_use_ssl_(cassandra_use_ssl),
934 57 : cassandra_ca_certs_(cassandra_ca_certs)
935 : {
936 19 : max_slice_ = max_slice;
937 : // default keyspace
938 19 : keyspace_ = g_viz_constants.COLLECTOR_KEYSPACE_CQL;
939 19 : init_vizd_tables();
940 :
941 : // Initialize database connection
942 19 : QE_LOG_NOQID(DEBUG, "Initializing QE without database!");
943 :
944 19 : ttlmap_ = g_viz_constants.TtlValuesDefault;
945 19 : max_tasks_ = max_tasks;
946 19 : }
947 :
948 25 : QueryEngine::QueryEngine(EventManager *evm,
949 : std::vector<std::string> cassandra_ips,
950 : std::vector<int> cassandra_ports,
951 : vector<string> redis_ip_ports,
952 : const std::string & redis_password,
953 : const bool redis_ssl_enable,
954 : const std::string & redis_keyfile,
955 : const std::string & redis_certfile,
956 : const std::string & redis_ca_cert,
957 : int max_tasks, int max_slice,
958 : const std::string & cassandra_user,
959 : const std::string & cassandra_password,
960 : bool cassandra_use_ssl,
961 : const std::string & cassandra_ca_certs,
962 : const std::string & cluster_id,
963 25 : const std::string &host_ip) :
964 50 : qosp_(new QEOpServerProxy(evm,
965 : this, redis_ip_ports, redis_password, redis_ssl_enable, redis_keyfile,
966 25 : redis_certfile, redis_ca_cert, host_ip, max_tasks)),
967 25 : evm_(evm),
968 25 : cassandra_ports_(cassandra_ports),
969 25 : cassandra_ips_(cassandra_ips),
970 25 : cassandra_user_(cassandra_user),
971 25 : cassandra_password_(cassandra_password),
972 25 : cassandra_use_ssl_(cassandra_use_ssl),
973 50 : cassandra_ca_certs_(cassandra_ca_certs) {
974 25 : dbif_.reset(new cass::cql::CqlIf(evm, cassandra_ips,
975 25 : cassandra_ports[0], cassandra_user, cassandra_password,
976 25 : cassandra_use_ssl_, cassandra_ca_certs_));
977 25 : if (cluster_id.empty()) {
978 25 : keyspace_ = g_viz_constants.COLLECTOR_KEYSPACE_CQL;
979 : } else {
980 0 : keyspace_ = g_viz_constants.COLLECTOR_KEYSPACE_CQL + '_' + cluster_id;
981 : }
982 25 : max_slice_ = max_slice;
983 25 : max_tasks_ = max_tasks;
984 25 : oldDataExists = true;
985 25 : init_vizd_tables();
986 :
987 : // Initialize database connection
988 25 : QE_TRACE_NOQID(DEBUG, "Initializing database");
989 :
990 25 : boost::system::error_code ec;
991 25 : int retries = 0;
992 25 : bool retry = true;
993 50 : while (retry == true) {
994 25 : retry = false;
995 :
996 25 : if (!dbif_->Db_Init()) {
997 0 : QE_LOG_NOQID(ERROR, "Database initialization failed");
998 0 : retry = true;
999 : }
1000 :
1001 25 : if (!retry) {
1002 25 : if (!dbif_->Db_SetTablespace(keyspace_)) {
1003 0 : QE_LOG_NOQID(ERROR, ": Create/Set KEYSPACE: " <<
1004 : keyspace_ << " FAILED");
1005 0 : retry = true;
1006 : }
1007 : }
1008 :
1009 25 : if (!retry) {
1010 25 : for (std::vector<GenDb::NewCf>::const_iterator it = vizd_tables.begin();
1011 100 : it != vizd_tables.end(); it++) {
1012 75 : if (!dbif_->Db_UseColumnfamily(*it)) {
1013 0 : retry = true;
1014 0 : break;
1015 : }
1016 : }
1017 : }
1018 :
1019 25 : if (!retry) {
1020 25 : for (std::vector<GenDb::NewCf>::const_iterator it =
1021 25 : vizd_stat_tables.begin();
1022 50 : it != vizd_stat_tables.end(); it++) {
1023 25 : if (!dbif_->Db_UseColumnfamily(*it)) {
1024 0 : retry = true;
1025 0 : break;
1026 : }
1027 : }
1028 :
1029 : }
1030 :
1031 25 : if (!retry) {
1032 25 : for (std::vector<std::string>::const_iterator it =
1033 25 : g_viz_constants._STATS_TABLES.begin();
1034 25 : it != g_viz_constants._STATS_TABLES.end() - 1; it++) {
1035 25 : if (!dbif_->Db_UseColumnfamily(*it)) {
1036 25 : oldDataExists = false;
1037 25 : QE_LOG_NOQID(DEBUG, "Older table does not exist. will query only the new table");
1038 25 : break;
1039 : }
1040 : }
1041 : }
1042 25 : if (oldDataExists) {
1043 0 : QE_LOG_NOQID(DEBUG, "Older table exists. will query both the tables");
1044 : }
1045 :
1046 25 : if (!retry) {
1047 25 : for (std::vector<GenDb::NewCf>::const_iterator it =
1048 25 : vizd_session_tables.begin();
1049 50 : it != vizd_session_tables.end(); it++) {
1050 25 : if (!dbif_->Db_UseColumnfamily(*it)) {
1051 0 : retry = true;
1052 0 : break;
1053 : }
1054 : }
1055 :
1056 : }
1057 :
1058 25 : if (retry) {
1059 0 : std::stringstream ss;
1060 0 : ss << "initialization of database failed. retrying " << retries++ << " time";
1061 : // Update connection info
1062 0 : ConnectionState::GetInstance()->Update(ConnectionType::DATABASE,
1063 0 : std::string(), ConnectionStatus::DOWN,
1064 0 : dbif_->Db_GetEndpoints(), std::string());
1065 0 : Q_E_LOG_LOG("QeInit", SandeshLevel::SYS_WARN, ss.str());
1066 0 : dbif_->Db_Uninit();
1067 0 : sleep(5);
1068 0 : }
1069 : }
1070 : {
1071 25 : bool init_done = false;
1072 25 : retries = 0;
1073 50 : while (!init_done && retries < 12) {
1074 25 : init_done = true;
1075 :
1076 25 : GenDb::ColList col_list;
1077 25 : std::string cfname = g_viz_constants.SYSTEM_OBJECT_TABLE;
1078 25 : GenDb::DbDataValueVec key;
1079 25 : key.push_back(g_viz_constants.SYSTEM_OBJECT_ANALYTICS);
1080 :
1081 : bool ttl_cached[TtlType::GLOBAL_TTL+1];
1082 125 : for (int ttli=0; ttli<=TtlType::GLOBAL_TTL; ttli++)
1083 100 : ttl_cached[ttli] = false;
1084 :
1085 25 : if (dbif_->Db_GetRow(&col_list, cfname, key,
1086 : GenDb::DbConsistency::LOCAL_ONE)) {
1087 25 : for (GenDb::NewColVec::iterator it = col_list.columns_.begin();
1088 250 : it != col_list.columns_.end(); it++) {
1089 225 : std::string col_name;
1090 : try {
1091 225 : col_name = boost::get<std::string>(it->name->at(0));
1092 0 : } catch (boost::bad_get& ex) {
1093 0 : QE_LOG_NOQID(ERROR, __func__ << ": Exception on col_name get");
1094 0 : break;
1095 0 : }
1096 225 : if (col_name == g_viz_constants.SYSTEM_OBJECT_GLOBAL_DATA_TTL) {
1097 : try {
1098 25 : ttlmap_.insert(std::make_pair(TtlType::GLOBAL_TTL, boost::get<uint64_t>(it->value->at(0))));
1099 25 : ttl_cached[TtlType::GLOBAL_TTL] = true;
1100 0 : } catch (boost::bad_get& ex) {
1101 0 : QE_LOG_NOQID(ERROR, __func__ << "Exception for boost::get, what=" << ex.what());
1102 0 : }
1103 200 : } else if (col_name == g_viz_constants.SYSTEM_OBJECT_CONFIG_AUDIT_TTL) {
1104 : try {
1105 25 : ttlmap_.insert(std::make_pair(TtlType::CONFIGAUDIT_TTL, boost::get<uint64_t>(it->value->at(0))));
1106 25 : ttl_cached[TtlType::CONFIGAUDIT_TTL] = true;
1107 0 : } catch (boost::bad_get& ex) {
1108 0 : QE_LOG_NOQID(ERROR, __func__ << "Exception for boost::get, what=" << ex.what());
1109 0 : }
1110 175 : } else if (col_name == g_viz_constants.SYSTEM_OBJECT_STATS_DATA_TTL) {
1111 : try {
1112 25 : ttlmap_.insert(std::make_pair(TtlType::STATSDATA_TTL, boost::get<uint64_t>(it->value->at(0))));
1113 25 : ttl_cached[TtlType::STATSDATA_TTL] = true;
1114 0 : } catch (boost::bad_get& ex) {
1115 0 : QE_LOG_NOQID(ERROR, __func__ << "Exception for boost::get, what=" << ex.what());
1116 0 : }
1117 150 : } else if (col_name == g_viz_constants.SYSTEM_OBJECT_FLOW_DATA_TTL) {
1118 : try {
1119 25 : ttlmap_.insert(std::make_pair(TtlType::FLOWDATA_TTL, boost::get<uint64_t>(it->value->at(0))));
1120 25 : ttl_cached[TtlType::FLOWDATA_TTL] = true;
1121 0 : } catch (boost::bad_get& ex) {
1122 0 : QE_LOG_NOQID(ERROR, __func__ << "Exception for boost::get, what=" << ex.what());
1123 0 : }
1124 : }
1125 225 : }
1126 : }
1127 125 : for (int ttli=0; ttli<=TtlType::GLOBAL_TTL; ttli++)
1128 100 : if (ttl_cached[ttli] == false)
1129 0 : init_done = false;
1130 :
1131 25 : retries++;
1132 25 : if (!init_done)
1133 0 : sleep(5);
1134 25 : }
1135 25 : if (!init_done) {
1136 0 : ttlmap_ = g_viz_constants.TtlValuesDefault;
1137 0 : QE_LOG_NOQID(ERROR, __func__ << "ttls are set manually");
1138 : }
1139 : }
1140 25 : dbif_->Db_SetInitDone(true);
1141 : // Update connection info
1142 50 : ConnectionState::GetInstance()->Update(ConnectionType::DATABASE,
1143 50 : std::string(), ConnectionStatus::UP, dbif_->Db_GetEndpoints(),
1144 50 : std::string());
1145 25 : }
1146 :
1147 98 : QueryEngine::~QueryEngine() {
1148 49 : if (dbif_) {
1149 25 : dbif_->Db_Uninit();
1150 25 : dbif_->Db_SetInitDone(false);
1151 : }
1152 98 : }
1153 :
1154 : using std::vector;
1155 :
1156 : int
1157 127 : QueryEngine::QueryPrepare(QueryParams qp,
1158 : std::vector<uint64_t> &chunk_size,
1159 : bool & need_merge, bool & map_output,
1160 : std::string& where, uint32_t& wterms,
1161 : std::string& select, std::string& post,
1162 : uint64_t& time_period,
1163 : std::string &table) {
1164 127 : string& qid = qp.qid;
1165 127 : QE_LOG_NOQID(INFO,
1166 : " Got Query to prepare for QID " << qid);
1167 : int ret_code;
1168 127 : if (cassandra_ports_.size() == 1 && cassandra_ports_[0] == 0) {
1169 0 : chunk_size.push_back(999);
1170 0 : need_merge = false;
1171 0 : map_output = false;
1172 0 : ret_code = 0;
1173 0 : table = string("ObjectCollectorInfo");
1174 : } else {
1175 : AnalyticsQuery *q;
1176 254 : q = new AnalyticsQuery(qid, dbif_, qp.terms, -1, NULL, ttlmap_, 0,
1177 127 : qp.maxChunks, this);
1178 127 : chunk_size.clear();
1179 127 : q->get_query_details(need_merge, map_output, chunk_size,
1180 : where, wterms ,select, post, time_period, ret_code);
1181 127 : table = q->table();
1182 127 : delete q;
1183 : }
1184 127 : return ret_code;
1185 : }
1186 :
1187 : bool
1188 32 : QueryEngine::QueryAccumulate(QueryParams qp,
1189 : const QEOpServerProxy::BufferT& input,
1190 : QEOpServerProxy::BufferT& output) {
1191 :
1192 32 : QE_TRACE_NOQID(DEBUG, "Creating analytics query object for merge_processing");
1193 : AnalyticsQuery *q;
1194 64 : q = new AnalyticsQuery(qp.qid, dbif_, qp.terms, -1, NULL, ttlmap_, 1,
1195 32 : qp.maxChunks, this);
1196 32 : QE_TRACE_NOQID(DEBUG, "Calling merge_processing");
1197 32 : bool ret = q->merge_processing(input, output);
1198 32 : delete q;
1199 32 : return ret;
1200 : }
1201 :
1202 : bool
1203 4 : QueryEngine::QueryFinalMerge(QueryParams qp,
1204 : const std::vector<boost::shared_ptr<QEOpServerProxy::BufferT> >& inputs,
1205 : QEOpServerProxy::BufferT& output) {
1206 :
1207 4 : QE_TRACE_NOQID(DEBUG, "Creating analytics query object for final_merge_processing");
1208 : AnalyticsQuery *q;
1209 8 : q = new AnalyticsQuery(qp.qid, dbif_, qp.terms, -1, NULL, ttlmap_, 1,
1210 4 : qp.maxChunks, this);
1211 4 : QE_TRACE_NOQID(DEBUG, "Calling final_merge_processing");
1212 4 : bool ret = q->final_merge_processing(inputs, output);
1213 4 : delete q;
1214 4 : return ret;
1215 : }
1216 :
1217 : bool
1218 64 : QueryEngine::QueryFinalMerge(QueryParams qp,
1219 : const std::vector<boost::shared_ptr<QEOpServerProxy::OutRowMultimapT> >& inputs,
1220 : QEOpServerProxy::OutRowMultimapT& output) {
1221 64 : QE_TRACE_NOQID(DEBUG, "Creating analytics query object for final_merge_processing");
1222 : AnalyticsQuery *q;
1223 128 : q = new AnalyticsQuery(qp.qid, dbif_, qp.terms, -1, NULL, ttlmap_, 1,
1224 64 : qp.maxChunks, this);
1225 :
1226 64 : if (!q->is_stat_table_query(q->table())
1227 91 : && !q->is_session_query(q->table())
1228 91 : && !q->is_flow_query(q->table())) {
1229 0 : QE_TRACE_NOQID(DEBUG, "MultiMap merge_final is for Stats only");
1230 0 : delete q;
1231 0 : return false;
1232 : }
1233 64 : QE_TRACE_NOQID(DEBUG, "Calling final_merge_processing for Stats");
1234 :
1235 64 : q->selectquery_->stats_->MergeFinal(inputs, output);
1236 : // apply limit
1237 66 : if (q->postprocess_->limit &&
1238 2 : output.size() > (size_t)q->postprocess_->limit) {
1239 0 : QEOpServerProxy::OutRowMultimapT::iterator it = output.begin();
1240 0 : std::advance(it, (size_t)q->postprocess_->limit);
1241 0 : output.erase(it, output.end());
1242 : }
1243 64 : delete q;
1244 64 : return true;
1245 : }
1246 :
1247 : // Query Execution of WHERE term
1248 : bool
1249 910 : QueryEngine::QueryExecWhere(void * handle, QueryParams qp, uint32_t chunk,
1250 : uint32_t or_number)
1251 : {
1252 910 : string& qid = qp.qid;
1253 910 : QE_TRACE_NOQID(DEBUG,
1254 : " Got Where Query to execute for QID " << qid << " chunk:"<< chunk);
1255 912 : if (cassandra_ports_.size() == 1 && cassandra_ports_[0] == 0) {
1256 : std::unique_ptr<std::vector<query_result_unit_t> > where_output(
1257 0 : new std::vector<query_result_unit_t>());
1258 0 : QE_TRACE_NOQID(DEBUG, " Finished NULL query processing for QID " << qid << " chunk:" << chunk);
1259 0 : QEOpServerProxy::QPerfInfo qperf(0,0,0);
1260 0 : qperf.error = 0;
1261 :
1262 0 : qosp_->QueryResult(handle, qperf, std::auto_ptr<std::vector<query_result_unit_t>>(where_output.release()));
1263 0 : return true;
1264 0 : }
1265 1824 : boost::shared_ptr<AnalyticsQuery> q(new AnalyticsQuery(qid, dbif_, qp.terms,
1266 1822 : or_number, NULL, ttlmap_, chunk, qp.maxChunks, this, handle));
1267 : // populate into a vector mainted by QOSP
1268 912 : qosp_->AddAnalyticsQuery(qid, q);
1269 912 : QE_TRACE_NOQID(DEBUG, " Finished parsing and starting where for QID " << qid << " chunk:" << chunk);
1270 :
1271 912 : q->where_start_ = UTCTimestampUsec();
1272 : // Bind the callback function to where query
1273 912 : q->wherequery_->where_query_cb_ = boost::bind(&QEOpServerProxy::QueryResult, qosp_.get(), _1, _2, _3);
1274 912 : q->query_status = q->wherequery_->process_query();
1275 912 : bool query_status_ = false;
1276 912 : switch (q->query_status) {
1277 0 : case QUERY_PROCESSING_NOT_STARTED:
1278 : /* should not come here */
1279 : case QUERY_FAILURE:
1280 0 : break;
1281 7 : case QUERY_SUCCESS:
1282 14 : q->qperf_.chunk_where_time =
1283 7 : static_cast<uint32_t>((UTCTimestampUsec() - q->where_start_)
1284 7 : /1000);
1285 7 : q->qperf_.error = q->status_details;
1286 7 : qosp_->QueryResult(q->handle_, q->qperf_, std::auto_ptr<std::vector<query_result_unit_t>>(q->wherequery_->where_result_.release()));
1287 912 : case QUERY_IN_PROGRESS:
1288 912 : query_status_ = true;
1289 912 : break;
1290 : }
1291 912 : return query_status_;
1292 912 : }
1293 :
1294 : // Query Execution of SELECT and post-processing
1295 : bool
1296 864 : QueryEngine::QueryExec(void * handle, QueryParams qp, uint32_t chunk,
1297 : const std::vector<query_result_unit_t> * where_info)
1298 : {
1299 864 : string& qid = qp.qid;
1300 864 : QE_TRACE_NOQID(DEBUG,
1301 : " Got Query to execute for QID " << qid << " chunk:"<< chunk);
1302 : //GenDb::GenDbIf *db_if = dbif_.get();
1303 864 : if (cassandra_ports_.size() == 1 && cassandra_ports_[0] == 0) {
1304 0 : std::unique_ptr<QEOpServerProxy::BufferT> final_output(new QEOpServerProxy::BufferT);
1305 0 : QEOpServerProxy::OutRowT outrow = boost::assign::map_list_of(
1306 0 : "MessageTS", "1368037623434740")(
1307 0 : "Messagetype", "IFMapString")(
1308 0 : "ModuleId", "ControlNode")(
1309 0 : "Source","b1s1")(
1310 0 : "ObjectLog","\n<IFMapString type=\"sandesh\"><message type=\"string\" identifier=\"1\">Cancelling Response timer.</message><file type=\"string\" identifier=\"-32768\">src/ifmap/client/ifmap_state_machine.cc</file><line type=\"i32\" identifier=\"-32767\">578</line></IFMapString>");
1311 0 : QEOpServerProxy::MetadataT metadata;
1312 0 : std::unique_ptr<QEOpServerProxy::OutRowMultimapT> final_moutput(new QEOpServerProxy::OutRowMultimapT);
1313 0 : for (int i = 0 ; i < 100; i++)
1314 0 : final_output->push_back(std::make_pair(outrow, metadata));
1315 0 : QE_TRACE_NOQID(DEBUG, " Finished query processing for QID " << qid << " chunk:" << chunk);
1316 0 : QEOpServerProxy::QPerfInfo qperf(0,0,0);
1317 0 : qperf.error = 0;
1318 0 : qosp_->QueryResult(handle, qperf, std::auto_ptr<QEOpServerProxy::BufferT>(final_output.release()), std::auto_ptr<QEOpServerProxy::OutRowMultimapT>(final_moutput.release()));
1319 0 : return true;
1320 0 : }
1321 : AnalyticsQuery *q;
1322 1728 : q = new AnalyticsQuery(qid, dbif_, qp.terms, -1, where_info, ttlmap_, chunk,
1323 864 : qp.maxChunks, this);
1324 :
1325 864 : QE_TRACE_NOQID(DEBUG, " Finished parsing and starting processing for QID " << qid << " chunk:" << chunk);
1326 864 : q->process_query();
1327 :
1328 864 : QE_TRACE_NOQID(DEBUG, " Finished query processing for QID " << qid << " chunk:" << chunk);
1329 864 : q->qperf_.error = q->status_details;
1330 864 : qosp_->QueryResult(handle, q->qperf_, std::auto_ptr<QEOpServerProxy::BufferT>(q->final_result.release()), std::auto_ptr<QEOpServerProxy::OutRowMultimapT>(q->final_mresult.release()));
1331 864 : delete q;
1332 864 : return true;
1333 : }
1334 :
1335 4 : bool QueryEngine::GetCumulativeStats(std::vector<GenDb::DbTableInfo> *vdbti,
1336 : GenDb::DbErrors *dbe, std::vector<GenDb::DbTableInfo> *vstats_dbti)
1337 : const {
1338 : {
1339 4 : std::scoped_lock lock(smutex_);
1340 4 : stable_stats_.GetCumulative(vstats_dbti);
1341 4 : }
1342 4 : return dbif_->Db_GetCumulativeStats(vdbti, dbe);
1343 : }
1344 :
1345 :
1346 0 : std::ostream &operator<<(std::ostream &out, query_result_unit_t& res)
1347 : {
1348 0 : out << "T:" << res.timestamp << " : Need to extract other information";
1349 : #if 0
1350 : out << "T:" << res.timestamp << " : ";
1351 :
1352 : if (res.info.length() < 48) {
1353 : boost::uuids::uuid tmp_u;
1354 : res.get_uuid(tmp_u);
1355 : out << " UUID:" << tmp_u;
1356 : }
1357 : #endif
1358 :
1359 0 : return out;
1360 : }
1361 :
1362 : bool
1363 162258 : AnalyticsQuery::is_stat_table_query(const std::string & tname) {
1364 162258 : if (tname.compare(0, g_viz_constants.STAT_VT_PREFIX.length(),
1365 : g_viz_constants.STAT_VT_PREFIX)) {
1366 147936 : return false;
1367 : }
1368 14877 : return true;
1369 : }
1370 :
1371 : bool
1372 157751 : AnalyticsQuery::is_session_query(const std::string & tname) {
1373 311701 : return (tname == g_viz_constants.SESSION_SERIES_TABLE ||
1374 311756 : tname == g_viz_constants.SESSION_RECORD_TABLE);
1375 : }
1376 :
1377 : bool
1378 2005 : AnalyticsQuery::is_stat_fieldnames_table_query(const std::string & tname) {
1379 2005 : if (tname.compare(0, g_viz_constants.STAT_VT_FIELDNAMES_PREFIX.length(),
1380 : g_viz_constants.STAT_VT_FIELDNAMES_PREFIX)) {
1381 1968 : return false;
1382 : }
1383 37 : return true;
1384 : }
1385 :
1386 139779 : bool AnalyticsQuery::is_flow_query(const std::string & tname)
1387 : {
1388 277456 : return ((tname == g_viz_constants.FLOW_SERIES_TABLE) ||
1389 277465 : (tname == g_viz_constants.FLOW_TABLE));
1390 : }
1391 :
1392 : // validation functions
1393 4146 : bool AnalyticsQuery::is_message_table_query(const std::string &tname)
1394 : {
1395 4146 : return (tname == g_viz_constants.MESSAGE_TABLE);
1396 : }
1397 :
1398 8561 : bool AnalyticsQuery::is_message_table_query()
1399 : {
1400 8561 : return (table_ == g_viz_constants.MESSAGE_TABLE);
1401 : }
1402 :
1403 19274 : bool AnalyticsQuery::is_object_table_query(const std::string &tname)
1404 : {
1405 : return (
1406 35139 : (tname != g_viz_constants.MESSAGE_TABLE) &&
1407 31726 : (tname != g_viz_constants.FLOW_TABLE) &&
1408 31466 : (tname != g_viz_constants.FLOW_SERIES_TABLE) &&
1409 15602 : (tname != g_viz_constants.OBJECT_VALUE_TABLE) &&
1410 46780 : !is_stat_table_query(tname) &&
1411 30917 : !is_session_query(tname));
1412 : }
1413 :
1414 2141 : bool AnalyticsQuery::is_valid_where_field(const std::string& where_field)
1415 : {
1416 10247 : for(size_t i = 0; i < g_viz_constants._TABLES.size(); i++)
1417 : {
1418 9118 : if (g_viz_constants._TABLES[i].name == table_)
1419 : {
1420 6986 : for (size_t j = 0;
1421 6986 : j < g_viz_constants._TABLES[i].schema.columns.size(); j++)
1422 : {
1423 6985 : if ((g_viz_constants._TABLES[i].schema.columns[j].name ==
1424 7998 : where_field) &&
1425 1012 : g_viz_constants._TABLES[i].schema.columns[j].index)
1426 1012 : return true;
1427 : }
1428 0 : return false;
1429 : }
1430 : }
1431 1129 : if (is_stat_table_query(table_)) {
1432 789 : AnalyticsQuery *m_query = (AnalyticsQuery *)main_query;
1433 789 : if (m_query->stats().is_stat_table_static()) {
1434 484 : StatsQuery::column_t cdesc = m_query->stats().get_column_desc(where_field);
1435 484 : if (cdesc.index) return true;
1436 484 : } else {
1437 : // For dynamic Stat Table queries, allow anything in the where clause
1438 305 : return true;
1439 : }
1440 : }
1441 340 : return true;
1442 : }
1443 :
1444 126 : bool AnalyticsQuery::is_valid_sort_field(const std::string& sort_field) {
1445 126 : if (
1446 252 : (sort_field == SELECT_PACKETS) ||
1447 252 : (sort_field == SELECT_BYTES) ||
1448 378 : (sort_field == SELECT_SUM_PACKETS) ||
1449 126 : (sort_field == SELECT_SUM_BYTES)
1450 : )
1451 0 : return true;
1452 :
1453 126 : return selectquery_->is_present_in_select_column_fields(sort_field);
1454 : }
1455 :
1456 126 : std::string AnalyticsQuery::get_column_field_datatype(
1457 : const std::string& column_field) {
1458 234 : for(size_t i = 0; i < g_viz_constants._TABLES.size(); i++) {
1459 234 : if (g_viz_constants._TABLES[i].name == table_) {
1460 1284 : for (size_t j = 0;
1461 1284 : j < g_viz_constants._TABLES[i].schema.columns.size(); j++) {
1462 1284 : if (g_viz_constants._TABLES[i].schema.columns[j].name ==
1463 : column_field) {
1464 126 : return g_viz_constants._TABLES[i].schema.columns[j].datatype;
1465 : }
1466 : }
1467 0 : return std::string("");
1468 : }
1469 : }
1470 0 : if (stats_.get()) {
1471 0 : StatsQuery::column_t vt = stats().get_column_desc(column_field);
1472 0 : if (vt.datatype == QEOpServerProxy::STRING)
1473 0 : return string("string");
1474 0 : else if (vt.datatype == QEOpServerProxy::UINT64)
1475 0 : return string("int");
1476 0 : else if (vt.datatype == QEOpServerProxy::DOUBLE)
1477 0 : return string("double");
1478 : else
1479 0 : return string("");
1480 0 : }
1481 0 : return std::string("");
1482 : }
1483 :
1484 : std::map< std::string, int > trace_enable_map;
1485 0 : void TraceEnable::HandleRequest() const
1486 : {
1487 0 : TraceEnableRes *resp = new TraceEnableRes;
1488 0 : std::string status;
1489 0 : std::string trace_type = get_TraceType();
1490 0 : if (trace_type == WHERE_RESULT_TRACE || trace_type == SELECT_RESULT_TRACE ||
1491 0 : trace_type == POSTPROCESS_RESULT_TRACE) {
1492 0 : if (get_enable())
1493 : {
1494 0 : trace_enable_map.insert(std::make_pair(trace_type, 1));
1495 0 : status = "Trace buffer Enabled";
1496 : } else {
1497 0 : trace_enable_map.erase(trace_type);
1498 0 : status = "Trace buffer Disabled";
1499 : }
1500 : } else {
1501 0 : status = "Invalid Trace buffer";
1502 : }
1503 0 : resp->set_enable_disable_status(status);
1504 0 : resp->set_TraceType(trace_type);
1505 0 : resp->set_context(context());
1506 0 : resp->set_more(false);
1507 0 : resp->Response();
1508 0 : }
1509 :
1510 0 : void TraceStatusReq::HandleRequest() const {
1511 0 : std::vector<std::string> trace_buf_list;
1512 0 : trace_buf_list.push_back(WHERE_RESULT_TRACE);
1513 0 : trace_buf_list.push_back(SELECT_RESULT_TRACE);
1514 0 : trace_buf_list.push_back(POSTPROCESS_RESULT_TRACE);
1515 0 : std::vector<TraceStatusInfo> trace_status_list;
1516 0 : for (std::vector<std::string>::const_iterator it = trace_buf_list.begin();
1517 0 : it != trace_buf_list.end(); ++it) {
1518 0 : TraceStatusInfo trace_status;
1519 0 : trace_status.set_TraceType(*it);
1520 0 : if (IS_TRACE_ENABLED(*it)) {
1521 0 : trace_status.set_enable_disable("Enabled");
1522 : } else {
1523 0 : trace_status.set_enable_disable("Disabled");
1524 : }
1525 0 : trace_status_list.push_back(trace_status);
1526 0 : }
1527 0 : TraceStatusRes *resp = new TraceStatusRes;
1528 0 : resp->set_trace_status_list(trace_status_list);
1529 0 : resp->set_context(context());
1530 0 : resp->set_more(false);
1531 0 : resp->Response();
1532 0 : }
1533 :
1534 23 : bool QueryEngine::GetDiffStats(std::vector<GenDb::DbTableInfo> *vdbti,
1535 : GenDb::DbErrors *dbe, std::vector<GenDb::DbTableInfo> *vstats_dbti) {
1536 : {
1537 23 : std::scoped_lock lock(smutex_);
1538 23 : stable_stats_.GetDiffs(vstats_dbti);
1539 23 : }
1540 23 : return dbif_->Db_GetStats(vdbti, dbe);
1541 : }
1542 :
1543 39 : bool QueryEngine::GetCqlStats(cass::cql::DbStats *stats) const {
1544 39 : cass::cql::CqlIf *cql_if(dynamic_cast<cass::cql::CqlIf *>(dbif_.get()));
1545 39 : if (cql_if == NULL) {
1546 16 : return false;
1547 : }
1548 23 : return cql_if->Db_GetCqlStats(stats);
1549 : }
1550 :
1551 4 : bool QueryEngine::GetCqlMetrics(cass::cql::Metrics *metrics) const {
1552 4 : cass::cql::CqlIf *cql_if(dynamic_cast<cass::cql::CqlIf *>(dbif_.get()));
1553 4 : if (cql_if == NULL) {
1554 0 : return false;
1555 : }
1556 4 : cql_if->Db_GetCqlMetrics(metrics);
1557 4 : return true;
1558 : }
1559 :
1560 4 : void ShowQEDbStatsReq::HandleRequest() const {
1561 4 : std::vector<GenDb::DbTableInfo> vdbti, vstats_dbti;
1562 4 : GenDb::DbErrors dbe;
1563 : QESandeshContext *qec = static_cast<QESandeshContext *>(
1564 4 : Sandesh::client_context());
1565 4 : assert(qec);
1566 4 : ShowQEDbStatsResp *resp(new ShowQEDbStatsResp);
1567 4 : qec->QE()->GetCumulativeStats(&vdbti, &dbe, &vstats_dbti);
1568 4 : cass::cql::Metrics cmetrics;
1569 4 : qec->QE()->GetCqlMetrics(&cmetrics);
1570 4 : resp->set_table_info(vdbti);
1571 4 : resp->set_errors(dbe);
1572 4 : resp->set_statistics_table_info(vstats_dbti);
1573 4 : resp->set_cql_metrics(cmetrics);
1574 4 : resp->set_context(context());
1575 4 : resp->Response();
1576 4 : }
1577 :
|
