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