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/tuple/tuple.hpp>
7 : #include <boost/foreach.hpp>
8 : #include "query.h"
9 : #include "stats_query.h"
10 : #include "base/work_pipeline.h"
11 :
12 : /*
13 : * This function performs GetRowAsync for each row key
14 : * Input: It takes rowkeys for which we have to get col
15 : values.Multiple instances execute this call
16 : simultaneously and possibly multiple times(steps)
17 : The instances store intermediate result in exts
18 : res stores consolidated result/instance
19 : * Ouput: Keeps calling itself for as many rowkeys involved
20 : Returns NULL if no more row key to be queried
21 : */
22 216135 : ExternalBase::Efn DbQueryUnit::QueryExec(uint32_t inst,
23 : const vector<q_result *> & exts,
24 : const Input & inp, Stage0Out & res) {
25 :
26 216135 : uint32_t step = exts.size();
27 216238 : AnalyticsQuery *m_query = (AnalyticsQuery *)main_query;
28 216238 : Input & cinp = const_cast<Input &> (inp);
29 : // Previous row fetch resulted in error, dont issue any more requests
30 216238 : if (query_fetch_error) {
31 360 : return NULL;
32 : }
33 217021 : if (step) {
34 215200 : res.query_result.insert(res.query_result.end(),
35 107610 : exts[step-1]->begin(),
36 107661 : exts[step-1]->end());
37 : }
38 216763 : res.current_row = cinp.row_count.fetch_add(1);
39 216763 : const uint32_t current_row = res.current_row;
40 : // continue as long as we have not reached end of rows
41 216763 : if (current_row < inp.total_rows) {
42 : // Frame the getrow context
43 108358 : GetRowInput *ip_ctx = new GetRowInput();
44 108257 : ip_ctx->rowkey = cinp.keys[current_row];
45 108036 : ip_ctx->cfname = cinp.cf_name;
46 108078 : ip_ctx->crange = cinp.cr;
47 108062 : ip_ctx->where_vec = cinp.where_vec;
48 108102 : ip_ctx->chunk_no = m_query->parallel_batch_num;
49 108102 : ip_ctx->qid = m_query->query_id;
50 108228 : ip_ctx->sub_qid = sub_query_id;
51 108228 : ip_ctx->row_no = current_row;
52 108228 : ip_ctx->inst = inst;
53 215849 : return boost::bind(&DbQueryUnit::PipelineCb, this, cinp.cf_name,
54 216386 : cinp.keys[current_row], cinp.cr, cinp.where_vec, ip_ctx, _1);
55 : } else {
56 : // done processing fetching all rows
57 108273 : return NULL;
58 : }
59 : }
60 :
61 108088 : bool DbQueryUnit::PipelineCb(std::string &cfname, GenDb::DbDataValueVec &rowkey,
62 : const GenDb::ColumnNameRange &cr,
63 : GenDb::WhereIndexInfoVec &where_vec,
64 : GetRowInput * ip_ctx, void *privdata) {
65 :
66 : // prepend T2: to value in each tuple in where_vec
67 108088 : std::string T2_string = GenDb::DbDataValueToString(rowkey.at(0));
68 321462 : BOOST_FOREACH(GenDb::WhereIndexInfo &where_info, where_vec) {
69 106695 : std::string value = GenDb::DbDataValueToString(where_info.get<2>());
70 106648 : if (boost::starts_with(value, "%")) {
71 1583 : continue;
72 : }
73 105031 : std::ostringstream where_oss;
74 104809 : where_oss << T2_string << ":" << value;
75 105217 : where_info.get<2>() = where_oss.str();
76 106707 : }
77 : /*
78 : * Call GetRowAsync, with args prepopulated
79 : */
80 108344 : AnalyticsQuery *m_query = (AnalyticsQuery *)main_query;
81 216280 : return m_query->dbif_->Db_GetRowAsync(cfname, rowkey, cr, where_vec,
82 : GenDb::DbConsistency::LOCAL_ONE,
83 216015 : boost::bind(&DbQueryUnit::cb, this, _1, _2, ip_ctx, privdata));
84 108052 : }
85 :
86 : /*
87 : * This function merges all the instance's results into the res
88 : */
89 1802 : bool DbQueryUnit::QueryMerge(const std::vector<boost::shared_ptr<Stage0Out> >
90 : & subs,
91 : const boost::shared_ptr<Input> & inp,
92 : Output & res) {
93 : // Merge all the instances result. Each instance is a vector of
94 : // query_results from steps
95 3603 : res.query_result = boost::shared_ptr<std::vector<query_result_unit_t> >
96 3603 : (new std::vector<query_result_unit_t>());
97 1801 : for (vector<boost::shared_ptr<Stage0Out> >::const_iterator it =
98 111595 : subs.begin(); it!=subs.end(); it++) {
99 323959 : res.query_result->insert(res.query_result->end(),
100 107986 : (*it)->query_result.begin(),
101 107998 : (*it)->query_result.end());
102 : }
103 1801 : return true;
104 : }
105 :
106 : /*
107 : * This function populates the rowkeys that have to be used
108 : * in querying.It populates it based on the cftype and the t2
109 : * and t1 values
110 : */
111 1802 : std::vector<GenDb::DbDataValueVec> DbQueryUnit::populate_row_keys() {
112 1802 : AnalyticsQuery *m_query = (AnalyticsQuery *)main_query;
113 1802 : uint32_t t2_start = m_query->from_time() >> g_viz_constants.RowTimeInBits;
114 1802 : uint32_t t2_end = m_query->end_time() >> g_viz_constants.RowTimeInBits;
115 :
116 1802 : if (m_query->is_object_table_query(m_query->table()))
117 : {
118 : GenDb::DbDataValue timestamp_start =
119 1056 : (uint32_t)std::numeric_limits<int32_t>::min();
120 1056 : cr.start_.push_back(timestamp_start);
121 1055 : }
122 : GenDb::DbDataValue timestamp_end =
123 1801 : (uint32_t)std::numeric_limits<int32_t>::max();
124 1801 : cr.finish_.push_back(timestamp_end);
125 :
126 1801 : std::vector<GenDb::DbDataValueVec> keys; // vector of keys for multi-row get
127 1801 : GenDb::ColListVec mget_res; // vector of result for each row
128 15608 : for (uint32_t t2 = t2_start; t2 <= t2_end; t2++)
129 : {
130 13806 : GenDb::ColList result;
131 13806 : GenDb::DbDataValueVec rowkey;
132 :
133 13807 : rowkey.push_back(t2);
134 13786 : if (m_query->is_stat_table_query(m_query->table())) {
135 1716 : uint8_t partition_no = 0;
136 1716 : rowkey.push_back(partition_no);
137 : }
138 41422 : if (m_query->is_flow_query(m_query->table()) ||
139 27588 : m_query->is_session_query(m_query->table())) {
140 3139 : for (uint8_t part_no = (uint8_t)g_viz_constants.PARTITION_MIN;
141 3139 : part_no < (uint8_t)g_viz_constants.PARTITION_MAX + 1;
142 : part_no++) {
143 2825 : GenDb::DbDataValueVec tmp_rowkey(rowkey);
144 2810 : tmp_rowkey.push_back(part_no);
145 2803 : if (m_query->is_flow_query(m_query->table())) {
146 1604 : for (uint8_t is_si = 0; is_si < 2; is_si++) {
147 1072 : GenDb::DbDataValueVec tmp_rowkey2(tmp_rowkey);
148 1062 : tmp_rowkey2.push_back(is_si);
149 1062 : for (GenDb::DbDataValueVec::iterator it =
150 3189 : row_key_suffix.begin(); it!=row_key_suffix.end();
151 1066 : it++) {
152 1062 : GenDb::DbDataValueVec tmp_rowkey3(tmp_rowkey2);
153 1064 : tmp_rowkey3.push_back(*it);
154 1072 : keys.push_back(tmp_rowkey3);
155 1072 : }
156 1065 : }
157 : } else {
158 2279 : for (GenDb::DbDataValueVec::iterator it =
159 9095 : row_key_suffix.begin(); it!=row_key_suffix.end();
160 4541 : it++) {
161 4540 : tmp_rowkey.push_back(*it);
162 : }
163 2272 : keys.push_back(tmp_rowkey);
164 : }
165 2817 : }
166 : } else {
167 13514 : if (!t_only_row)
168 : {
169 2046 : for (GenDb::DbDataValueVec::iterator it =
170 6964 : row_key_suffix.begin(); it!=row_key_suffix.end();
171 2871 : it++) {
172 2878 : rowkey.push_back(*it);
173 : }
174 : }
175 :
176 : // If querying message_index_tables, partition_no is an
177 : // additional row_key
178 : // It spans values 0..15
179 13491 : if (t_only_col) {
180 112616 : for (uint8_t part_no = (uint8_t)g_viz_constants.PARTITION_MIN;
181 112616 : part_no < (uint8_t)g_viz_constants.PARTITION_MAX + 1;
182 : part_no++) {
183 101472 : GenDb::DbDataValueVec tmp_rowkey(rowkey);
184 100482 : tmp_rowkey.push_back(part_no);
185 100681 : keys.push_back(tmp_rowkey);
186 100739 : }
187 : } else {
188 2030 : keys.push_back(rowkey);
189 : }
190 : }
191 13499 : }
192 3604 : return keys;
193 1802 : }
194 :
195 : /*
196 : * This function calls the GetRowAsync based on the input
197 : * passed to it. It creates a pipeline which executes
198 : * the multiple GetRow calls asynchronously. It also
199 : * provides a callback which is used to collect the
200 : * results from the GetRow operation
201 : */
202 1802 : query_status_t DbQueryUnit::process_query()
203 : {
204 1802 : AnalyticsQuery *m_query = (AnalyticsQuery *)main_query;
205 1802 : uint32_t t2_start = m_query->from_time() >> g_viz_constants.RowTimeInBits;
206 1802 : uint32_t t2_end = m_query->end_time() >> g_viz_constants.RowTimeInBits;
207 :
208 1802 : QE_TRACE(DEBUG, " Async Database query for " <<
209 : (t2_end - t2_start + 1) << " rows");
210 1802 : QE_TRACE(DEBUG, " Async Database query for T2_start:"
211 : << t2_start
212 : << " T2_end:" << t2_end
213 : << " cf:" << cfname
214 : << " where_vec size:" << where_vec.size()
215 : << " column_start size:" << cr.start_.size()
216 : << " column_end size:" << cr.finish_.size()
217 : << " where_vec size:" << where_vec.size());
218 1802 : std::vector<GenDb::DbDataValueVec> keys = populate_row_keys();
219 :
220 : /* Create a pipeline to fetch all rows corresponding to keys */
221 1802 : int max_tasks = keys.size();
222 1802 : std::vector<std::pair<int,int> > tinfo;
223 110566 : for (uint idx=0; idx<(uint)max_tasks; idx++) {
224 108764 : tinfo.push_back(make_pair(0, -1));
225 : }
226 :
227 : QEPipeT * wp = new QEPipeT(
228 3604 : new WorkStage<Input, Output, q_result, Stage0Out>(
229 : tinfo,
230 : boost::bind(&DbQueryUnit::QueryExec, this, _1, _2, _3, _4),
231 1802 : boost::bind(&DbQueryUnit::QueryMerge, this, _1, _2, _3)));
232 :
233 : // Populate the input to the pipeline
234 1802 : boost::shared_ptr<Input> inp(new Input());
235 1802 : inp.get()->row_count = 0;
236 1802 : inp.get()->total_rows = keys.size();
237 1802 : inp.get()->cf_name = cfname;
238 1802 : inp.get()->cr = cr;
239 1802 : inp.get()->where_vec = where_vec;
240 1802 : inp.get()->keys = keys;
241 : // Start the pipeline with callback and input
242 1802 : wp->Start(boost::bind(&DbQueryUnit::WPCompleteCb, this, wp, _1), inp);
243 1802 : return QUERY_IN_PROGRESS;
244 1802 : }
245 :
246 : /*
247 : * This is called after getting all the results from
248 : * all the row keys. It copies result to the query_result
249 : * and deletes the pipeline
250 : */
251 1802 : void DbQueryUnit::WPCompleteCb(QEPipeT *wp, bool ret_code) {
252 1802 : boost::shared_ptr<Output> res = wp->Result();
253 : //copy pipeline output to DbQueryUnit query_output
254 1802 : query_result = (res->query_result);
255 1802 : int size = res->query_result->size();
256 1802 : QE_TRACE(DEBUG, " Database query completed with Async "
257 : << size << " rows");
258 : // Have the result ready and processing is done
259 : // sort the result before returning
260 1802 : std::sort(query_result->begin(), query_result->end());
261 1802 : if (IS_TRACE_ENABLED(WHERE_RESULT_TRACE)) {
262 0 : std::stringstream ss;
263 0 : for (std::vector<query_result_unit_t>::const_iterator it =
264 0 : query_result->begin(); it != query_result->end(); it++) {
265 0 : const query_result_unit_t &result_unit(*it);
266 0 : ss << "T: " << result_unit.timestamp << ": ";
267 0 : for (GenDb::DbDataValueVec::const_iterator rt =
268 0 : result_unit.info.begin(); rt != result_unit.info.end();
269 0 : rt++) {
270 0 : ss << " " << *rt;
271 : }
272 0 : ss << std::endl;
273 : }
274 0 : QE_TRACE(DEBUG, "Result: " << cfname << ": " << ss.str());
275 0 : }
276 1802 : delete wp;
277 1802 : if (ret_code && !query_fetch_error) {
278 1799 : status_details = 0;
279 1799 : query_status = QUERY_SUCCESS;
280 : } else {
281 3 : query_status = QUERY_FAILURE;
282 : }
283 1802 : parent_query->subquery_processed(this);
284 1802 : }
285 :
286 108443 : void DbQueryUnit::cb(GenDb::DbOpResult::type dresult,
287 : std::auto_ptr<GenDb::ColList> column_list,
288 : GetRowInput * get_row_ctx, void * privdata) {
289 108443 : std::unique_ptr<q_result> q_result_ptr(new q_result);
290 108442 : std::unique_ptr<GetRowInput> gri(get_row_ctx);
291 : uint32_t t2;
292 108441 : uint8_t session_type = 0;
293 108441 : uint8_t is_si = 0;
294 108441 : AnalyticsQuery *m_query = (AnalyticsQuery *)main_query;
295 108441 : QueryEngine *qe = m_query->qe_;
296 : try {
297 108441 : GenDb::DbDataValueVec val = gri.get()->rowkey;
298 108424 : t2 = boost::get<uint32_t>(val.at(0));
299 108421 : if (m_query->is_session_query(m_query->table())
300 108425 : || m_query->is_flow_query(m_query->table())) {
301 3383 : session_type = boost::get<uint8_t>(val.at(3));
302 3384 : is_si = boost::get<uint8_t>(val.at(2));
303 : }
304 108424 : } catch (boost::bad_get& ex) {
305 0 : assert(0);
306 : }
307 108414 : if (dresult != GenDb::DbOpResult::OK) {
308 : // Dont issue any more requests
309 12 : query_fetch_error = true;
310 12 : ExternalProcIf<q_result> * rpi(
311 : reinterpret_cast<ExternalProcIf<q_result> *>(privdata));
312 12 : if (m_query->is_stat_table_query(m_query->table())) {
313 4 : std::scoped_lock lock(qe->smutex_);
314 4 : qe->stable_stats_.Update(m_query->stat_name_attr, false, true,
315 : false, 1);
316 4 : }
317 12 : rpi->Response(std::move(q_result_ptr));
318 12 : return;
319 : }
320 : // Update the reads against the stat
321 108402 : if (m_query->is_stat_table_query(m_query->table())) {
322 1599 : std::scoped_lock lock(qe->smutex_);
323 1599 : qe->stable_stats_.Update(m_query->stat_name_attr, false, false,
324 : false, 1);
325 1599 : }
326 108395 : GenDb::NewColVec::iterator i;
327 :
328 113236 : for (i = column_list->columns_.begin(); i != column_list->columns_.end();
329 4811 : i++) {
330 : {
331 4811 : query_result_unit_t result_unit;
332 : uint32_t t1;
333 4811 : if (m_query->is_stat_table_query(m_query->table())) {
334 : try {
335 221 : t1 = boost::get<uint32_t>(i->name->at(i->name->size()-2));
336 0 : } catch (boost::bad_get& ex) {
337 0 : assert(0);
338 : }
339 4590 : } else if (m_query->is_session_query(m_query->table())
340 4590 : || m_query->is_flow_query(m_query->table())) {
341 408 : int ts_at = 2;
342 : try {
343 408 : t1 = boost::get<uint32_t>(i->name->at(ts_at));
344 0 : } catch (boost::bad_get& ex) {
345 0 : assert(0);
346 : }
347 : } else {
348 : // For MessageIndex tables t1 is stored in the first column
349 : // except for timestamp table
350 4182 : int ts_at = 0;
351 4182 : if (t_only_col) {
352 4062 : ts_at = i->name->size() - 2;
353 : } else {
354 120 : ts_at = i->name->size() - 1;
355 : }
356 4182 : assert(ts_at >= 0);
357 : try {
358 4182 : t1 = boost::get<uint32_t>(i->name->at(ts_at));
359 0 : } catch (boost::bad_get& ex) {
360 0 : assert(0);
361 : }
362 : }
363 4811 : result_unit.timestamp = TIMESTAMP_FROM_T2T1(t2, t1);
364 :
365 5376 : if
366 9383 : ((result_unit.timestamp < m_query->from_time()) ||
367 4572 : (result_unit.timestamp > m_query->end_time()))
368 : {
369 : //QE_TRACE(DEBUG, "Discarding timestamp "
370 : // << result_unit.timestamp);
371 : // got a result outside of the time range
372 565 : continue;
373 : }
374 :
375 : // Add to result vector
376 4246 : if (m_query->is_stat_table_query(m_query->table())) {
377 220 : std::string attribstr;
378 : boost::uuids::uuid uuid;
379 :
380 : try {
381 220 : uuid = boost::get<boost::uuids::uuid>(i->name->at(i->name->size()-1));
382 0 : } catch (boost::bad_get& ex) {
383 0 : QE_ASSERT(0);
384 0 : } catch (const std::out_of_range& oor) {
385 0 : QE_ASSERT(0);
386 : }
387 :
388 : try {
389 220 : attribstr = boost::get<std::string>(i->value->at(i->value->size()-1));
390 0 : } catch (boost::bad_get& ex) {
391 0 : QE_ASSERT(0);
392 0 : } catch (const std::out_of_range& oor) {
393 0 : QE_ASSERT(0);
394 : }
395 :
396 220 : result_unit.set_stattable_info(
397 : attribstr,
398 : uuid);
399 4246 : } else if (m_query->is_session_query(m_query->table())
400 4026 : || m_query->is_flow_query(m_query->table())) {
401 243 : result_unit.info.clear();
402 243 : result_unit.info.push_back(is_si);
403 243 : result_unit.info.push_back(session_type);
404 243 : result_unit.info.push_back(i->name->at(0));
405 243 : result_unit.info.push_back(i->name->at(1));
406 243 : result_unit.info.push_back(i->name->at(2));
407 243 : result_unit.info.push_back(i->name->at(3));
408 243 : GenDb::DbDataValueVec::const_iterator itr;
409 : GenDb::DbDataValueVec::const_iterator end =
410 243 : (m_query->selectquery_->unroll_needed?
411 243 : (i->value->end()):(i->value->end() - 1));
412 7155 : for (itr = i->value->begin(); itr != end;
413 6912 : itr++) {
414 6912 : result_unit.info.push_back(*itr);
415 : }
416 : } else {
417 : // If message index table uuid is not the value, but
418 : // column name
419 3783 : if (t_only_col) {
420 3664 : GenDb::DbDataValueVec val = gri.get()->rowkey;
421 3664 : message_table_query_get_row(val, i, result_unit);
422 3664 : } else {
423 119 : result_unit.info = *i->value;
424 : }
425 : }
426 4245 : q_result_ptr->push_back(result_unit);
427 4811 : }
428 :
429 : }
430 108410 : if (privdata) {
431 108412 : ExternalProcIf<q_result> * rpi(
432 : reinterpret_cast<ExternalProcIf<q_result> *>(privdata));
433 108412 : rpi->Response(std::move(q_result_ptr));
434 : }
435 :
436 108455 : }
437 :
438 3664 : void DbQueryUnit::message_table_query_get_row(
439 : GenDb::DbDataValueVec const &val,
440 : GenDb::NewColVec::iterator const &res_it,
441 : query_result_unit_t &result_unit) {
442 : // cassandra returns fields in the ascending order by column-name.
443 : // pushing fields in order as per schema.
444 : // key = rowkey[0] T2
445 : // key2 = rowkey[1] partition
446 : // column1 = name[0] T1
447 : // column2 = name[1] UUID
448 : // column3 = value[10] T2:Source
449 : // column4 = value[11] T2:Messagetype
450 : // column5 = value[12] T2:ModuleId
451 : // column6 = value[13] T2:<object-type1>:<object-value1>)
452 : // column7 = value[14]
453 : // column8 = value[15]
454 : // column9 = value[16]
455 : // column10 = value[0]
456 : // column11 = value[1]
457 : // column12 = value[2]
458 : // column13 = value[3]
459 : // column14 = value[4]
460 : // column15 = value[5]
461 : // column16 = value[6]
462 : // column17 = value[7]
463 : // column18 = value[8]
464 : // column19 = value[9]
465 : // DATA = value[17]
466 3664 : result_unit.info.push_back(val.at(0));
467 3664 : result_unit.info.push_back(val.at(1));
468 3664 : result_unit.info.push_back(res_it->name->at(0));
469 3664 : result_unit.info.push_back(res_it->name->at(1));
470 3664 : result_unit.info.push_back(res_it->value->at(10));
471 3664 : result_unit.info.push_back(res_it->value->at(11));
472 3664 : result_unit.info.push_back(res_it->value->at(12));
473 3664 : result_unit.info.push_back(res_it->value->at(13));
474 3664 : result_unit.info.push_back(res_it->value->at(14));
475 3664 : result_unit.info.push_back(res_it->value->at(15));
476 3664 : result_unit.info.push_back(res_it->value->at(16));
477 3664 : result_unit.info.push_back(res_it->value->at(0));
478 3664 : result_unit.info.push_back(res_it->value->at(1));
479 3664 : result_unit.info.push_back(res_it->value->at(2));
480 3664 : result_unit.info.push_back(res_it->value->at(3));
481 3664 : result_unit.info.push_back(res_it->value->at(4));
482 3664 : result_unit.info.push_back(res_it->value->at(5));
483 3664 : result_unit.info.push_back(res_it->value->at(6));
484 3664 : result_unit.info.push_back(res_it->value->at(7));
485 3663 : result_unit.info.push_back(res_it->value->at(8));
486 3664 : result_unit.info.push_back(res_it->value->at(9));
487 3664 : result_unit.info.push_back(res_it->value->at(17));
488 3664 : }
|
