Line data Source code
1 : /*
2 : * Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
3 : * Copyright Contributors to the OpenSDN project
4 : */
5 :
6 : #include <boost/assign/list_of.hpp>
7 : #include "base/regex.h"
8 : #include "rapidjson/document.h"
9 : #include "rapidjson/stringbuffer.h"
10 : #include "rapidjson/writer.h"
11 : #include "query.h"
12 : #include "stats_query.h"
13 :
14 : using boost::assign::map_list_of;
15 : using contrail::regex;
16 : using contrail::regex_match;
17 : using contrail::regex_search;
18 :
19 1726 : bool PostProcessingQuery::sort_field_comparator(
20 : const QEOpServerProxy::ResultRowT& lhs,
21 : const QEOpServerProxy::ResultRowT& rhs) {
22 1726 : std::map<std::string, std::string>::const_iterator lhs_it, rhs_it;
23 1726 : for (std::vector<sort_field_t>::iterator sort_it = sort_fields.begin();
24 2731 : sort_it != sort_fields.end(); sort_it++) {
25 1726 : lhs_it = lhs.first.find((*sort_it).name);
26 1726 : QE_ASSERT(lhs_it != lhs.first.end());
27 1726 : rhs_it = rhs.first.find((*sort_it).name);
28 1726 : QE_ASSERT(rhs_it != rhs.first.end());
29 6904 : if ((*sort_it).type == std::string("int") ||
30 8630 : (*sort_it).type == std::string("long") ||
31 3452 : (*sort_it).type == std::string("ipv4")) {
32 0 : uint64_t lhs_val = 0, rhs_val = 0;
33 0 : stringToInteger(lhs_it->second, lhs_val);
34 0 : stringToInteger(rhs_it->second, rhs_val);
35 0 : if (lhs_val < rhs_val) return true;
36 0 : if (lhs_val > rhs_val) return false;
37 : } else {
38 1726 : if (lhs_it->second < rhs_it->second) return true;
39 1159 : if (lhs_it->second > rhs_it->second) return false;
40 : }
41 : }
42 :
43 1005 : return false;
44 : }
45 :
46 32 : bool PostProcessingQuery::merge_processing(
47 : const QEOpServerProxy::BufferT& input,
48 : QEOpServerProxy::BufferT& output)
49 : {
50 32 : if (status_details != 0)
51 : {
52 0 : QE_TRACE(DEBUG,
53 : "No need to process query, as there were errors previously");
54 0 : return false;
55 : }
56 :
57 : // Check if the result has to be sorted
58 32 : if (sorted) {
59 24 : QEOpServerProxy::BufferT *merged_result = &output;
60 24 : const QEOpServerProxy::BufferT *raw_result1 = &(input);
61 :
62 24 : if (result_.get() == NULL) {
63 24 : size_t merged_result_size = merged_result->size();
64 24 : merged_result->reserve(merged_result_size + raw_result1->size());
65 24 : copy(raw_result1->begin(), raw_result1->end(),
66 : std::back_inserter(*merged_result));
67 24 : if (merged_result_size) {
68 0 : if (sorting_type == ASCENDING) {
69 0 : std::inplace_merge(merged_result->begin(),
70 0 : merged_result->begin() + merged_result_size,
71 : merged_result->end(),
72 : boost::bind(&PostProcessingQuery::sort_field_comparator,
73 : this, _1, _2));
74 : } else {
75 0 : std::inplace_merge(merged_result->rbegin(),
76 0 : merged_result->rbegin() + raw_result1->size(),
77 0 : merged_result->rend(),
78 : boost::bind(&PostProcessingQuery::sort_field_comparator,
79 : this, _1, _2));
80 : }
81 : }
82 : } else {
83 0 : QEOpServerProxy::BufferT *raw_result2 = result_.get();
84 0 : size_t size1 = raw_result1->size();
85 0 : size_t size2 = raw_result2->size();
86 0 : QE_TRACE(DEBUG, "Merging results from vectors of size:" <<
87 : size1 << " and " << size2);
88 0 : merged_result->reserve(raw_result1->size() + raw_result2->size());
89 0 : if (sorting_type == ASCENDING) {
90 0 : std::merge(raw_result1->begin(), raw_result1->end(),
91 : raw_result2->begin(), raw_result2->end(),
92 : std::back_inserter(*merged_result),
93 : boost::bind(&PostProcessingQuery::sort_field_comparator,
94 : this, _1, _2));
95 : } else {
96 0 : std::merge(raw_result1->rbegin(), raw_result1->rend(),
97 0 : raw_result2->rbegin(), raw_result2->rend(),
98 : std::back_inserter(*merged_result),
99 : boost::bind(&PostProcessingQuery::sort_field_comparator,
100 : this, _1, _2));
101 : }
102 : }
103 : } else {
104 8 : QE_TRACE(DEBUG, "Merge_Processing: Adding inputs to output");
105 8 : QEOpServerProxy::BufferT *merged_result = &output;
106 8 : const QEOpServerProxy::BufferT *raw_result1 = &(input);
107 :
108 8 : if (result_.get() == NULL)
109 : {
110 8 : merged_result->reserve(raw_result1->size());
111 8 : copy(raw_result1->begin(), raw_result1->end(),
112 : std::back_inserter(*merged_result));
113 : } else {
114 :
115 0 : QEOpServerProxy::BufferT *raw_result2 = result_.get();
116 0 : size_t size1 = raw_result1->size();
117 0 : size_t size2 = raw_result2->size();
118 0 : QE_TRACE(DEBUG, "Merging results from vectors of size:" <<
119 : size1 << " and " << size2);
120 0 : merged_result->reserve(raw_result1->size() + raw_result2->size());
121 0 : copy(raw_result1->begin(), raw_result1->end(),
122 : std::back_inserter(*merged_result));
123 0 : copy(raw_result2->begin(), raw_result2->end(),
124 : std::back_inserter(*merged_result));
125 : }
126 8 : QE_TRACE(DEBUG, "Merge_Processing: Done adding inputs to output");
127 : }
128 :
129 : // Have the result ready and processing is done
130 32 : status_details = 0;
131 32 : return true;
132 : }
133 :
134 4 : bool PostProcessingQuery::final_merge_processing(
135 : const std::vector<boost::shared_ptr<QEOpServerProxy::BufferT> >& inputs,
136 : QEOpServerProxy::BufferT& output)
137 : {
138 4 : bool merge_done = false;
139 :
140 4 : if (status_details != 0)
141 : {
142 0 : QE_TRACE(DEBUG,
143 : "No need to process query, as there were errors previously");
144 0 : return false;
145 : }
146 :
147 4 : if (!merge_done) {
148 4 : QEOpServerProxy::BufferT *merged_result = &output;
149 4 : size_t final_vector_size = 0;
150 : // merge the results from parallel queries
151 36 : for (size_t i = 0; i < inputs.size(); i++) {
152 32 : final_vector_size += inputs[i]->size();
153 : }
154 4 : merged_result->reserve(final_vector_size);
155 4 : QE_TRACE(DEBUG, "Merging results between " << inputs.size()
156 : << " vectors with final vector size:" << final_vector_size);
157 36 : for (size_t i = 0; i < inputs.size(); i++) {
158 32 : QEOpServerProxy::BufferT *raw_result = inputs[i].get();
159 32 : copy(raw_result->begin(), raw_result->end(),
160 : std::back_inserter(*merged_result));
161 : }
162 : }
163 :
164 4 : if (sorted) {
165 3 : QEOpServerProxy::BufferT *merged_result = &output;
166 3 : if (sorting_type == ASCENDING) {
167 2 : std::sort(merged_result->begin(), merged_result->end(),
168 : boost::bind(&PostProcessingQuery::sort_field_comparator,
169 : this, _1, _2));
170 : } else {
171 1 : std::sort(merged_result->rbegin(), merged_result->rend(),
172 : boost::bind(&PostProcessingQuery::sort_field_comparator,
173 : this, _1, _2));
174 : }
175 : }
176 :
177 4 : if (limit) {
178 2 : QEOpServerProxy::BufferT *merged_result = &output;
179 2 : QE_TRACE(DEBUG, "Apply Limit [" << limit << "]");
180 2 : if (merged_result->size() > (size_t)limit) {
181 0 : merged_result->resize(limit);
182 : }
183 : }
184 :
185 : // Have the result ready and processing is done
186 4 : status_details = 0;
187 4 : return true;
188 : }
189 :
190 857 : query_status_t PostProcessingQuery::process_query() {
191 857 : if (status_details != 0)
192 : {
193 0 : QE_TRACE(DEBUG,
194 : "No need to process query, as there were errors previously");
195 0 : return QUERY_FAILURE;
196 : }
197 :
198 857 : AnalyticsQuery *mquery = (AnalyticsQuery *)main_query;
199 857 : result_ = std::move(mquery->selectquery_->result_);
200 857 : mresult_ = std::move(mquery->selectquery_->mresult_);
201 857 : QEOpServerProxy::BufferT *raw_result = result_.get();
202 :
203 : /* filter are ANDs over OR
204 : * [ [ e1 AND e2 ] OR [ e3 ] ]
205 : */
206 : /* below is filter processing for stats table queries
207 : */
208 857 : if (filter_list.size()) {
209 244 : size_t num_filtered=0;
210 244 : MapBufT::iterator kt = mresult_->end();
211 244 : for (MapBufT::iterator it = mresult_->begin();
212 376 : it!= mresult_->end(); it++) {
213 :
214 132 : if (kt!=mresult_->end()) {
215 86 : mresult_->erase(kt);
216 86 : kt = mresult_->end();
217 : }
218 132 : std::map<std::string, QEOpServerProxy::SubVal>& attrs = it->second.first;
219 132 : bool delete_row = true;
220 132 : std::string unknown_attr;
221 230 : for (size_t j = 0; j < filter_list.size(); j++) {
222 132 : std::vector<filter_match_t>& filter_and = filter_list[j];
223 132 : bool and_check = true;
224 :
225 160 : for (size_t k = 0; k < filter_and.size(); k++) {
226 : std::map<std::string, QEOpServerProxy::SubVal>::const_iterator iter =
227 132 : attrs.find(filter_and[k].name);
228 132 : if (iter == attrs.end()) {
229 6 : unknown_attr = filter_and[k].name;
230 104 : break;
231 : } else {
232 126 : unknown_attr.clear();
233 : }
234 126 : std::ostringstream vstream;
235 126 : vstream << iter->second;
236 :
237 126 : switch(filter_and[k].op) {
238 0 : case EQUAL:
239 0 : if (filter_and[k].value != vstream.str())
240 : {
241 0 : and_check = false;
242 : }
243 0 : break;
244 0 : case NOT_EQUAL:
245 0 : if (filter_and[k].value == vstream.str())
246 : {
247 0 : and_check = false;
248 : }
249 0 : break;
250 8 : case LEQ:
251 8 : if (iter->second.which() ==
252 : QEOpServerProxy::UINT64) {
253 : uint64_t filter_val;
254 4 : stringToInteger(filter_and[k].value,
255 : filter_val);
256 : uint64_t col_val =
257 4 : boost::get<uint64_t>(iter->second);
258 4 : if (col_val > filter_val) {
259 2 : and_check = false;
260 : }
261 4 : } else if (iter->second.which() ==
262 : QEOpServerProxy::DOUBLE) {
263 : double filter_val;
264 4 : stringToInteger(filter_and[k].value,
265 : filter_val);
266 : double col_val =
267 4 : boost::get<double>(iter->second);
268 4 : if (col_val > filter_val) {
269 3 : and_check = false;
270 : }
271 : }
272 8 : break;
273 8 : case GEQ:
274 8 : if (iter->second.which() ==
275 : QEOpServerProxy::UINT64) {
276 : uint64_t filter_val;
277 4 : stringToInteger(filter_and[k].value,
278 : filter_val);
279 : uint64_t col_val =
280 4 : boost::get<uint64_t>(iter->second);
281 4 : if (col_val < filter_val) {
282 2 : and_check = false;
283 : }
284 4 : } else if (iter->second.which() ==
285 : QEOpServerProxy::DOUBLE) {
286 : double filter_val;
287 4 : stringToInteger(filter_and[k].value,
288 : filter_val);
289 : double col_val =
290 4 : boost::get<double>(iter->second);
291 4 : if (col_val < filter_val) {
292 2 : and_check = false;
293 : }
294 : }
295 8 : break;
296 110 : case REGEX_MATCH:
297 110 : if (!regex_match(vstream.str(),
298 110 : filter_and[k].match_e)) {
299 89 : and_check = false;
300 : }
301 110 : break;
302 0 : default:
303 : // upsupported filter operation
304 0 : QE_TRACE(DEBUG, "Unsupported filter operation " << filter_and[k].op);
305 0 : break;
306 : }
307 126 : if (and_check == false)
308 98 : break;
309 126 : }
310 132 : if (and_check == true) {
311 34 : delete_row = false;
312 34 : break;
313 : }
314 : }
315 132 : if (unknown_attr.size()) {
316 6 : QE_TRACE(DEBUG, "Unknown filter attr in row " << unknown_attr);
317 : }
318 132 : if (delete_row) {
319 98 : num_filtered++;
320 98 : kt = it;
321 : }
322 132 : }
323 244 : if (kt!=mresult_->end()) {
324 12 : mresult_->erase(kt);
325 12 : kt = mresult_->end();
326 : }
327 244 : QE_TRACE(DEBUG, "# of entries filtered is " << num_filtered);
328 :
329 : }
330 :
331 : /* below is filter processing for non stats table queries
332 : */
333 857 : if (filter_list.size() != 0) {
334 244 : QEOpServerProxy::BufferT filtered_table;
335 : // do filter operation
336 244 : QE_TRACE(DEBUG, "Doing filter operation");
337 1330 : for (size_t i = 0; i < raw_result->size(); i++) {
338 1086 : QEOpServerProxy::ResultRowT row = (*raw_result)[i];
339 1086 : bool delete_row = true;
340 :
341 1462 : for (size_t j = 0; j < filter_list.size(); j++) {
342 1125 : std::vector<filter_match_t>& filter_and = filter_list[j];
343 1125 : bool and_check = true;
344 :
345 2040 : for (size_t k = 0; k < filter_and.size(); k++) {
346 1291 : std::map<std::string, std::string>::iterator iter;
347 1291 : iter = row.first.find(filter_and[k].name);
348 1291 : if (iter == row.first.end())
349 : {
350 563 : if (!(filter_and[k].ignore_col_absence)) {
351 0 : and_check = false;
352 376 : break;
353 : }
354 563 : continue;
355 : }
356 :
357 728 : switch(filter_and[k].op)
358 : {
359 176 : case EQUAL:
360 176 : if (filter_and[k].value != iter->second)
361 : {
362 102 : and_check = false;
363 : }
364 176 : break;
365 :
366 213 : case NOT_EQUAL:
367 213 : if (filter_and[k].value == iter->second)
368 : {
369 49 : and_check = false;
370 : }
371 213 : break;
372 :
373 0 : case LEQ:
374 : {
375 : int filter_value =
376 0 : atoi(filter_and[k].value.c_str());
377 0 : int column_value= atoi(iter->second.c_str());
378 0 : if (column_value > filter_value)
379 : {
380 0 : and_check = false;
381 : }
382 0 : break;
383 : }
384 :
385 0 : case GEQ:
386 : {
387 : int filter_value =
388 0 : atoi(filter_and[k].value.c_str());
389 0 : int column_value= atoi(iter->second.c_str());
390 0 : if (column_value < filter_value)
391 : {
392 0 : and_check = false;
393 : }
394 0 : break;
395 : }
396 :
397 339 : case REGEX_MATCH:
398 : {
399 339 : if (!regex_match(iter->second,
400 339 : filter_and[k].match_e))
401 : {
402 225 : and_check = false;
403 : }
404 339 : break;
405 : }
406 :
407 0 : default:
408 : // upsupported filter operation
409 0 : QE_LOG(ERROR, "Unsupported filter operation: " <<
410 : filter_and[k].op);
411 0 : return QUERY_FAILURE;
412 : }
413 728 : if (and_check == false)
414 376 : break;
415 : }
416 :
417 1125 : if (and_check == true) {
418 749 : QE_TRACE(DEBUG, "filter out entry #:" << i);
419 749 : delete_row = false;
420 749 : break;
421 : }
422 : }
423 1086 : if (!delete_row) {
424 749 : filtered_table.push_back(row);
425 : }
426 1086 : }
427 244 : *raw_result = filtered_table;
428 244 : }
429 :
430 : // Check if the result has to be sorted
431 857 : if (sorted) {
432 34 : if (sorting_type == ASCENDING) {
433 26 : std::sort(raw_result->begin(), raw_result->end(),
434 : boost::bind(&PostProcessingQuery::sort_field_comparator,
435 : this, _1, _2));
436 : } else {
437 8 : std::sort(raw_result->rbegin(), raw_result->rend(),
438 : boost::bind(&PostProcessingQuery::sort_field_comparator,
439 : this, _1, _2));
440 : }
441 : }
442 :
443 : // If the flow series query is parallelized, we should apply the limit
444 : // only after the result from all the tasks are merged
445 : // (@ final_merge_processing).
446 872 : if ((mquery->table() != g_viz_constants.FLOW_SERIES_TABLE ||
447 872 : (mquery->table() == g_viz_constants.FLOW_SERIES_TABLE &&
448 872 : !mquery->is_query_parallelized())) && limit) {
449 27 : QE_TRACE(DEBUG, "Apply Limit [" << limit << "]");
450 27 : if (raw_result->size() > (size_t)limit) {
451 2 : raw_result->resize(limit);
452 : }
453 27 : if (mresult_->size() > (size_t)limit) {
454 4 : MapBufT::iterator it = mresult_->begin();
455 4 : std::advance(it, limit);
456 4 : mresult_->erase(it, mresult_->end());
457 : }
458 : }
459 :
460 857 : if (IS_TRACE_ENABLED(POSTPROCESS_RESULT_TRACE))
461 : {
462 0 : std::vector<QEOpServerProxy::ResultRowT>::iterator res_it;
463 0 : QE_TRACE(DEBUG, "== Post Processing Result ==");
464 0 : for (res_it = raw_result->begin(); res_it != raw_result->end();
465 0 : ++res_it) {
466 0 : std::vector<final_result_col> row_entry;
467 0 : std::map<std::string, std::string>::iterator map_it;
468 0 : for (map_it = (*res_it).first.begin();
469 0 : map_it != (*res_it).first.end(); ++map_it) {
470 0 : final_result_col col;
471 0 : col.set_col(map_it->first); col.set_value(map_it->second);
472 0 : row_entry.push_back(col);
473 : //QE_TRACE(DEBUG, map_it->first << " : " << map_it->second);
474 0 : }
475 0 : FINAL_RESULT_ROW_TRACE(QeTraceBuf, mquery->query_id, row_entry);
476 0 : }
477 : }
478 :
479 : #if 0
480 : //if ((limit) && (!sorted))
481 : for (int i = 0 ; i < 200000; i++)
482 : raw_result->push_back(map_list_of(
483 : "destvn","abc-cor\"poration:front-end-network:001")(
484 : "sourceip","168430090")("destip","3232238090")(
485 : "sourcevn","abc-corporation:front-end-network:002")(
486 : "protocol","80")("dport","62000")("sport","1000")(
487 : "sum(packets)","4294967196")
488 : );
489 : #endif
490 :
491 : // Have the result ready and processing is done
492 857 : status_details = 0;
493 857 : parent_query->subquery_processed(this);
494 857 : return QUERY_SUCCESS;
495 : }
|
