Line data Source code
1 : /*
2 : * Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
3 : */
4 :
5 : /*
6 : * This file details various data structures used for query
7 : * engine processing
8 : *
9 : * Overall the data structure heirarchy is as following:
10 : *
11 : * AnalyticsQuery--
12 : * |
13 : * |--SelectQuery
14 : * |
15 : * |--PostProcessingQuery
16 : * |
17 : * |--WhereQuery (multiple ANDs)
18 : * |
19 : * |-DbQueryUnit
20 : * |
21 : * |-DbQueryUnit
22 : * ...
23 : *
24 : */
25 : #ifndef QUERY_H_
26 : #define QUERY_H_
27 :
28 : #include <atomic>
29 : #include <sstream>
30 : #include <iostream>
31 : #include <string>
32 : #include <map>
33 : #include <mutex>
34 : #include <vector>
35 : #include <set>
36 : #include <algorithm>
37 : #include <exception>
38 : #include <cstdlib>
39 : #include <utility>
40 : #include <list>
41 :
42 : #include <boost/uuid/uuid.hpp>
43 : #include <boost/uuid/uuid_generators.hpp>
44 : #include <boost/uuid/uuid_io.hpp>
45 : #include <boost/algorithm/string/predicate.hpp>
46 : #include <boost/lexical_cast.hpp>
47 : #include <boost/algorithm/string.hpp>
48 : #include "io/event_manager.h"
49 : #include <boost/bind/bind.hpp>
50 : #include <boost/assign/list_of.hpp>
51 : #include <boost/scoped_ptr.hpp>
52 : #include <boost/ptr_container/ptr_map.hpp>
53 : #include "base/util.h"
54 : #include "base/task.h"
55 : #include "base/address.h"
56 : #include "base/parse_object.h"
57 : #include <boost/asio.hpp>
58 : #include <boost/tuple/tuple.hpp>
59 : #include <boost/foreach.hpp>
60 : #include "base/logging.h"
61 : #include "hiredis/hiredis.h"
62 : #include "hiredis/boostasio.hpp"
63 : #include <contrail-collector/redis_connection.h>
64 : #include "base/regex.h"
65 : #include "base/work_pipeline.h"
66 : #include "database/gendb_if.h"
67 : #include "database/gendb_statistics.h"
68 : #include <contrail-collector/viz_message.h>
69 : #include "json_parse.h"
70 : #include "QEOpServerProxy.h"
71 : #include <sandesh/sandesh_ctrl_types.h>
72 : #include <sandesh/sandesh_trace.h>
73 : #include <sandesh/common/vns_constants.h>
74 : #include <analytics/viz_constants.h>
75 : #include <query_engine/qe_constants.h>
76 : #include <query_engine/qe_types.h>
77 : #include "rapidjson/document.h"
78 : #include <boost/shared_ptr.hpp>
79 :
80 : using namespace boost::placeholders;
81 :
82 : extern std::map< std::string, int > trace_enable_map;
83 : #define IS_TRACE_ENABLED(trace_flag_string) \
84 : ((trace_enable_map.find(trace_flag_string) != trace_enable_map.end()))
85 :
86 : // following is the list of trace flags
87 : #define WHERE_RESULT_TRACE "where_result_trace"
88 : #define SELECT_RESULT_TRACE "select_result_trace"
89 : #define POSTPROCESS_RESULT_TRACE "postprocess_result_trace"
90 :
91 : #define QE_ASSERT(cond) assert((cond))
92 :
93 : #define TIMESTAMP_FROM_T2T1(t2, t1) (((uint64_t)(t2)) << g_viz_constants.RowTimeInBits) |\
94 : ((t1) & g_viz_constants.RowTimeInMask);
95 :
96 : // for Sandesh trace
97 : #define QE_TRACE_BUF "QeTraceBuf"
98 :
99 : extern SandeshTraceBufferPtr QeTraceBuf;
100 :
101 : // fix this later
102 : #define QE_LOG(level, log_msg) { \
103 : std::string qid("");\
104 : std::string table("");\
105 : int batch_num = 0;\
106 : if ((AnalyticsQuery *)(this->main_query))\
107 : {\
108 : qid = (((AnalyticsQuery *)(this->main_query))->query_id);\
109 : table = (((AnalyticsQuery *)(this->main_query))->table());\
110 : batch_num = (((AnalyticsQuery *)(this->main_query))->parallel_batch_num);\
111 : }\
112 : std::stringstream ss; ss << ":" << batch_num << ":" << log_msg; \
113 : Q_E_QUERY_LOG_SEND(qid, table, ss.str());\
114 : }
115 :
116 : // Following log message will be printed for only one thread among all threads
117 : #define QE_LOG_GLOBAL(level, log_msg) { \
118 : std::string qid("");\
119 : std::string table("");\
120 : int batch_num = 0;\
121 : if ((AnalyticsQuery *)(this->main_query))\
122 : {\
123 : qid = (((AnalyticsQuery *)(this->main_query))->query_id);\
124 : table = (((AnalyticsQuery *)(this->main_query))->table());\
125 : batch_num = (((AnalyticsQuery *)(this->main_query))->parallel_batch_num);\
126 : }\
127 : if (batch_num == 0) {\
128 : std::stringstream ss; ss << ":" << batch_num << ":" << log_msg; \
129 : Q_E_QUERY_LOG_SEND(qid, table, ss.str());\
130 : }\
131 : }
132 :
133 : #define QE_LOG_NOQID(level, log_msg) { \
134 : std::stringstream ss; ss << log_msg; \
135 : Q_E_LOG_SEND(ss.str());\
136 : }
137 :
138 : #define QE_TRACE_NOQID(level, trace_msg) {\
139 : std::stringstream ss; ss << level << ":" << trace_msg; \
140 : QUERY_TRACE_TRACE(QeTraceBuf, std::string(""), ss.str());\
141 : }
142 :
143 : #define QE_TRACE(level, trace_msg) {\
144 : std::stringstream ss; ss << level << ":" << trace_msg; \
145 : QUERY_TRACE_TRACE(QeTraceBuf, \
146 : (((AnalyticsQuery *)(this->main_query))? \
147 : (((AnalyticsQuery *)(this->main_query))->query_id):\
148 : std::string("")), ss.str());\
149 : }
150 :
151 : #if !defined(DEBUG)
152 : #define DEBUG "DEBUG"
153 : #endif
154 :
155 : // this is to return errors from the member functions of the QueryUnit class
156 : // and its derived classes. These macros just set the status_details field
157 : // and return from the function
158 : #define QE_PARSE_ERROR(cond) if (!(cond)) \
159 : { QE_LOG(DEBUG, "EBADMSG"); this->status_details = EBADMSG; return;}
160 : #define QE_INVALIDARG_ERROR(cond) if (!(cond)) \
161 : { QE_LOG(DEBUG, "EINVAL"); this->status_details = EINVAL; return;}
162 : #define QE_IO_ERROR(cond) if (!(cond)) \
163 : { QE_LOG(DEBUG, "EIO"); this->status_details = EIO; return;}
164 : #define QE_NOENT_ERROR(cond) if (!(cond)) \
165 : { QE_LOG(DEBUG, "ENOENT"); this->status_details = ENOENT; return;}
166 : #define QE_PARSE_ERROR_RETURN(cond, ret_val) if (!(cond)) \
167 : { this->status_details = EBADMSG; return ret_val;}
168 : #define QE_INVALIDARG_ERROR_RETURN(cond, ret_val) if (!(cond)) \
169 : { this->status_details = EINVAL; return ret_val;}
170 : #define QE_IO_ERROR_RETURN(cond, ret_val) if (!(cond)) \
171 : { this->status_details = EIO; return ret_val;}
172 : #define QE_NOENT_ERROR_RETURN(cond, ret_val) if (!(cond)) \
173 : { this->status_details = ENOENT; return ret_val;}
174 : #define QE_QUERY_FETCH_ERROR() {\
175 : this->status_details = EIO; \
176 : AnalyticsQuery *m_query = (AnalyticsQuery *)main_query; \
177 : if (m_query) { \
178 : m_query->qperf_.error = status_details; \
179 : m_query->status_details = status_details; \
180 : m_query->query_status = QUERY_FAILURE; \
181 : } \
182 : QE_LOG(ERROR, "QUERY failed to get rows " << QUERY_FAILURE); \
183 : }
184 :
185 : extern bool oldDataExists;
186 :
187 : typedef boost::shared_ptr<GenDb::GenDbIf> GenDbIfPtr;
188 :
189 : // flow sample stats which is stored in Cassandra flow index tables
190 : struct flow_stats {
191 2007 : flow_stats(uint64_t ibytes=0, uint64_t ipkts=0, bool ishort_flow=false) :
192 2007 : bytes(ibytes), pkts(ipkts), short_flow(ishort_flow) {
193 2007 : }
194 : bool operator==(const flow_stats &rhs) const {
195 : return bytes == rhs.bytes &&
196 : pkts == rhs.pkts &&
197 : short_flow == rhs.short_flow &&
198 : flow_list == rhs.flow_list;
199 : }
200 : uint64_t bytes;
201 : uint64_t pkts;
202 : bool short_flow;
203 : std::set<boost::uuids::uuid> flow_list;
204 : };
205 :
206 : // 8-tuple corresponding to a flow
207 : struct flow_tuple {
208 : flow_tuple() : protocol(0), source_port(0), dest_port(0), direction(0) {
209 : }
210 :
211 : flow_tuple(const std::string& vr, const std::string& svn,
212 : const std::string& dvn, const IpAddress& sip,
213 : const IpAddress& dip, uint32_t proto,
214 : uint32_t sport, uint32_t dport, uint32_t dir) :
215 : vrouter(vr), source_vn(svn), dest_vn(dvn), source_ip(sip),
216 : dest_ip(dip), protocol(proto), source_port(sport),
217 : dest_port(dport), direction(dir) {
218 : }
219 :
220 : bool operator<(const flow_tuple& rhs) const {
221 : if (vrouter < rhs.vrouter) return true;
222 : if (vrouter > rhs.vrouter) return false;
223 :
224 : if (source_vn < rhs.source_vn) return true;
225 : if (source_vn > rhs.source_vn) return false;
226 :
227 : if (dest_vn < rhs.dest_vn) return true;
228 : if (dest_vn > rhs.dest_vn) return false;
229 :
230 : if (source_ip < rhs.source_ip) return true;
231 : if (source_ip > rhs.source_ip) return false;
232 :
233 : if (dest_ip < rhs.dest_ip) return true;
234 : if (dest_ip > rhs.dest_ip) return false;
235 :
236 : if (protocol < rhs.protocol) return true;
237 : if (protocol > rhs.protocol) return false;
238 :
239 : if (source_port < rhs.source_port) return true;
240 : if (source_port > rhs.source_port) return false;
241 :
242 : if (dest_port < rhs.dest_port) return true;
243 : if (dest_port > rhs.dest_port) return false;
244 :
245 : if (direction < rhs.direction) return true;
246 : if (direction > rhs.direction) return false;
247 :
248 : return false;
249 : }
250 :
251 : bool operator==(const flow_tuple& rhs) const {
252 : return (vrouter == rhs.vrouter) &&
253 : (source_vn == rhs.source_vn) &&
254 : (dest_vn == rhs.dest_vn) &&
255 : (source_ip == rhs.source_ip) &&
256 : (dest_ip == rhs.dest_ip) &&
257 : (protocol == rhs.protocol) &&
258 : (source_port == rhs.source_port) &&
259 : (dest_port == rhs.dest_port) &&
260 : (direction == rhs.direction);
261 : }
262 :
263 : friend std::ostream& operator<<(std::ostream& out, const flow_tuple& ft);
264 :
265 : std::string vrouter;
266 : std::string source_vn;
267 : std::string dest_vn;
268 : IpAddress source_ip;
269 : IpAddress dest_ip;
270 : uint32_t protocol;
271 : uint32_t source_port;
272 : uint32_t dest_port;
273 : uint32_t direction;
274 : };
275 :
276 : struct uuid_flow_stats {
277 : uuid_flow_stats(uint64_t t, flow_stats stats) :
278 : last_timestamp(t), last_stats(stats) {
279 : }
280 : // indiates the timestamp when the last_stats was updated
281 : uint64_t last_timestamp;
282 : // indicates the aggregate value of the stats seen @ last_ts
283 : flow_stats last_stats;
284 : };
285 :
286 : // Result of basic unit of any analytics event query
287 : struct query_result_unit_t {
288 : // timestamp of an analytics event (log, flow record sample etc..)
289 : uint64_t timestamp;
290 :
291 : // this string will be
292 : // UUID is in case of database queries for flow-records/flow-series
293 : // stats+UUID after database queries for flow-records WHERE queries
294 : // stats+UUID+8-tuple afer flow-series WHERE query
295 : // AttribJSON+UUID for StatsTable queries
296 : // key,key2,column1-column19,DATA for messagetablev2
297 : GenDb::DbDataValueVec info;
298 :
299 : // Following APIs will be invoked based on the table being queried
300 :
301 : void set_stattable_info(
302 : const std::string& attribstr,
303 : const boost::uuids::uuid& uuid);
304 :
305 : void get_stattable_info(
306 : std::string& attribstr,
307 : boost::uuids::uuid& uuid) const;
308 :
309 : // Get UUID from the info field
310 : void get_uuid(boost::uuids::uuid& u) const;
311 : // Get UUID and stats
312 : void get_uuid_stats(boost::uuids::uuid& u, flow_stats& stats) const;
313 : // Get UUID and stats and 8-tuple
314 : void get_objectid(std::string&) const;
315 : // for sorting and set operations
316 : bool operator<(const query_result_unit_t& rhs) const;
317 :
318 : // for printing
319 : friend std::ostream &operator<<(std::ostream &out,
320 : query_result_unit_t&);
321 : } ;
322 :
323 : typedef std::vector<query_result_unit_t> WhereResultT;
324 :
325 : // Different status codes of query processing
326 : enum query_status_t {
327 : QUERY_PROCESSING_NOT_STARTED = 0,
328 : QUERY_IN_PROGRESS = 1,
329 : QUERY_SUCCESS = 2,
330 : QUERY_FAILURE = 3
331 : };
332 :
333 : // basic query unit (other classes will inherit from this)
334 : class QueryUnit {
335 : public:
336 : QueryUnit(QueryUnit *p_query, QueryUnit *m_query);
337 : virtual ~QueryUnit();
338 : virtual query_status_t process_query() = 0;
339 :
340 : typedef QEOpServerProxy::BufferT BufT;
341 : typedef QEOpServerProxy::OutRowMultimapT MapBufT;
342 :
343 : // following callback function is called whenever a subquery
344 : // is prcoessed
345 2643 : virtual void subquery_processed(QueryUnit *subquery) {};
346 :
347 : // list of child queries
348 : std::vector<QueryUnit *> sub_queries;
349 :
350 : // if this is a child/sub query, then following is pointer to
351 : // the parent query
352 : QueryUnit *parent_query;
353 : // Pointer to the highest level query
354 : QueryUnit *main_query;
355 :
356 : int pending_subqueries;
357 : query_status_t query_status;
358 : uint32_t status_details;
359 :
360 : // After query is processed following vector is populated
361 : boost::shared_ptr<WhereResultT> query_result;
362 : };
363 :
364 : class QueryResultMetaData {
365 : public:
366 : QueryResultMetaData() {
367 : }
368 : virtual ~QueryResultMetaData() = 0;
369 : };
370 :
371 : class fsMetaData : public QueryResultMetaData {
372 : public:
373 : fsMetaData(const std::set<boost::uuids::uuid>& flows)
374 : : uuids(flows.begin(), flows.end()) {
375 : }
376 : ~fsMetaData() {
377 : }
378 : std::set<boost::uuids::uuid> uuids;
379 : };
380 :
381 : struct GetRowInput {
382 : GenDb::DbDataValueVec rowkey;
383 : std::string cfname;
384 : GenDb::ColumnNameRange crange;
385 : GenDb::WhereIndexInfoVec where_vec;
386 : int chunk_no;
387 : std::string qid;
388 : int sub_qid;
389 : int row_no;
390 : int inst;
391 : };
392 :
393 : // max number of entries to extract from db
394 : #define MAX_DB_QUERY_ENTRIES 100000000
395 : // This class provides interface for doing single index database query
396 : class DbQueryUnit : public QueryUnit {
397 : public:
398 4113 : DbQueryUnit(QueryUnit *p_query, QueryUnit *m_query):
399 4113 : QueryUnit(p_query, m_query)
400 4113 : { cr.count_ = MAX_DB_QUERY_ENTRIES;
401 4113 : t_only_col = false; t_only_row = false;
402 4113 : sub_query_id = (p_query->sub_queries.size())-1;
403 4113 : query_fetch_error = false;
404 4113 : };
405 : virtual query_status_t process_query();
406 :
407 :
408 : // portion of column family name other than T1
409 : std::string cfname;
410 : // all type of match operations can be representated as
411 : // getrangeslice operations on Cassandra DB. For e.g.
412 : // NOT_EQUAL operation will be two getrangeslice operation
413 : GenDb::ColumnNameRange cr;
414 : GenDb::WhereIndexInfoVec where_vec;
415 : // row key suffix will be used to append DIR to flow
416 : // series/records query
417 : GenDb::DbDataValueVec row_key_suffix;
418 : bool t_only_col; // only T is in column name
419 : bool t_only_row; // only T2 is in row key
420 : int sub_query_id;
421 : typedef std::vector<query_result_unit_t> q_result;
422 : struct Input {
423 : std::atomic<uint32_t> row_count;
424 : std::atomic<uint32_t> total_rows;
425 : std::string cf_name;
426 : GenDb::ColumnNameRange cr;
427 : GenDb::WhereIndexInfoVec where_vec;
428 : std::vector<GenDb::DbDataValueVec> keys;
429 : };
430 :
431 : struct Output {
432 : boost::shared_ptr<WhereResultT> query_result;
433 : };
434 : struct Stage0Out {
435 : WhereResultT query_result;
436 : uint32_t current_row;
437 : };
438 : ExternalBase::Efn QueryExec(uint32_t inst,
439 : const std::vector<q_result *> & exts,
440 : const Input & inp, Stage0Out & res);
441 : typedef WorkPipeline<Input, Output> QEPipeT;
442 : bool QueryMerge(const std::vector<boost::shared_ptr<Stage0Out> > & subs,
443 : const boost::shared_ptr<Input> & inp, Output & res);
444 : void cb(GenDb::DbOpResult::type dresult, std::auto_ptr<GenDb::ColList>
445 : columns, GetRowInput *get_row_ctx, void *privdata);
446 : void message_table_query_get_row(GenDb::DbDataValueVec const &val,
447 : GenDb::NewColVec::iterator const &res_it,
448 : query_result_unit_t &result_unit);
449 : void WPCompleteCb(QEPipeT *wp, bool ret_code);
450 : std::vector<GenDb::DbDataValueVec> populate_row_keys();
451 : bool PipelineCb(std::string &, GenDb::DbDataValueVec &,
452 : const GenDb::ColumnNameRange &, GenDb::WhereIndexInfoVec &,
453 : GetRowInput *, void *);
454 : bool query_fetch_error;
455 : };
456 :
457 : struct SetOperationUnit {
458 : static void op_and(std::string qi, WhereResultT& res,
459 : std::vector<WhereResultT*> inp);
460 : static void op_or(std::string qi, WhereResultT& res,
461 : std::vector<WhereResultT*> inp);
462 : };
463 :
464 : typedef boost::function<void (void *, QEOpServerProxy::QPerfInfo,
465 : std::auto_ptr<WhereResultT>)> WhereQueryCbT;
466 : // Where processing class
467 : // Result is available for SELECT processing in query_result field
468 : // It will be an array of timestamp and
469 : // UUID is in case of messages/object-trace WHERE queries
470 : // stats+UUID after database queries for flow-records WHERE queries
471 : // stats+UUID+8-tuple for flow-series WHERE query
472 :
473 : struct filter_match_t {
474 : std::string name; // column name of match and filter
475 : std::string value; // column value to compare with
476 : match_op op; // matching op
477 : bool ignore_col_absence; // ignore (i.e. do not delete) if col is absent
478 : contrail::regex match_e; // matching regex expression
479 :
480 780 : filter_match_t():ignore_col_absence(false) {};
481 : };
482 :
483 : class WhereQuery : public QueryUnit {
484 : public:
485 :
486 : bool StatTermParse(QueryUnit *main_query, const contrail_rapidjson::Value& where_term,
487 : std::string& pname, match_op& pop,
488 : GenDb::DbDataValue& pval, GenDb::DbDataValue& pval2,
489 : std::string& sname, match_op& sop,
490 : GenDb::DbDataValue& sval, GenDb::DbDataValue& sval2);
491 :
492 : bool StatTermProcess(const contrail_rapidjson::Value& where_term,
493 : QueryUnit* pnode, QueryUnit *main_query);
494 :
495 : WhereQuery(const std::string& where_json_string, int session_type,
496 : int is_si,int direction, int32_t or_number, QueryUnit *main_query);
497 : // construtor for UT
498 : WhereQuery(QueryUnit *mq);
499 : virtual query_status_t process_query();
500 : virtual void subquery_processed(QueryUnit *subquery);
501 : void handle_object_type_value(AnalyticsQuery *m_query,
502 : DbQueryUnit *db_query,
503 : bool object_id_specified);
504 : bool populate_where_vec(AnalyticsQuery *m_query,
505 : GenDb::WhereIndexInfoVec *where_vec,
506 : const std::string& query_col,
507 : const GenDb::Op::type db_op,
508 : const std::string& value);
509 : std::string query_column_to_cass_column(AnalyticsQuery *m_query,
510 : const std::string& query_column);
511 :
512 : // filter list to store filters converted from where cluase
513 : std::vector<std::vector<filter_match_t> > filter_list_;
514 : std::vector<std::string> additional_select_;
515 :
516 : // 0 is for egress and 1 for ingress
517 : int32_t direction_ing;
518 : const std::string json_string_;
519 : uint32_t wterms_;
520 : std::unique_ptr<WhereResultT> where_result_;
521 : // Used to store the sub_query(Each DbQueryUnits) results
522 : std::vector<WhereResultT*> inp;
523 : std::vector<WhereResultT*> inp_new_data;
524 : typedef boost::function<void(void *, QEOpServerProxy::QPerfInfo,
525 : std::auto_ptr<WhereResultT> )> WhereQueryCbT;
526 : WhereQueryCbT where_query_cb_;
527 :
528 : void populate_session_where_vec_list(std::vector<GenDb::WhereIndexInfoVec> *where_vec_list,
529 : const GenDb::WhereIndexInfoVec &rest_where_vec,
530 : const GenDb::WhereIndexInfoVec &labels_vec,
531 : const GenDb::WhereIndexInfoVec &remote_labels_vec,
532 : const GenDb::WhereIndexInfoVec &arbitrary_tags_vec,
533 : const GenDb::WhereIndexInfoVec &remote_arbitrary_tags_vec);
534 :
535 : private:
536 : std::mutex vector_push_mutex_;
537 : };
538 :
539 : typedef std::vector<std::string> final_result_row_t;
540 :
541 : struct final_result_t {
542 : // These are actual names used in the database
543 : std::vector<std::string> columns;
544 : std::vector<final_result_row_t> final_result_table;
545 : uint status;
546 : };
547 :
548 : enum agg_op_t {
549 : RAW = 1,
550 : SUM = 2,
551 : AGG_OP_INVALID = 3,
552 : };
553 :
554 : enum stat_type_t {
555 : PKT_STATS = 1,
556 : BYTE_STATS = 2,
557 : };
558 :
559 : // Aggregated stats
560 : struct agg_stats_t {
561 : agg_op_t agg_op;
562 : stat_type_t stat_type;
563 : };
564 :
565 : class StatsSelect;
566 :
567 : // following data structure does the processing of SELECT portion of query
568 : class SelectQuery : public QueryUnit {
569 : public:
570 : SelectQuery(QueryUnit *main_query,
571 : const std::map<std::string, std::string>& json_api_data);
572 :
573 : virtual query_status_t process_query();
574 :
575 : // Query related fields
576 : std::string json_string_;
577 : std::vector<std::string> select_column_fields;
578 : std::vector<agg_stats_t> agg_stats;
579 : // Whethere timestamp will be one of the columns
580 : bool provide_timeseries;
581 : // relevant only if provide_timeseries is set
582 : uint granularity;
583 :
584 : // column family name (column family to query to get column
585 : // fields/NULL for flow series)
586 : std::string cfname;
587 : bool unroll_needed;
588 :
589 : std::unique_ptr<BufT> result_;
590 : std::unique_ptr<MapBufT> mresult_;
591 :
592 : std::unique_ptr<StatsSelect> stats_;
593 :
594 : enum fs_query_type {
595 : FS_SELECT_INVALID = 0x0,
596 : FS_SELECT_T = 0x1,
597 : FS_SELECT_TS = 0x2,
598 : FS_SELECT_FLOW_TUPLE = 0x4,
599 : FS_SELECT_STATS = 0x8,
600 : FS_SELECT_T_FLOW_TUPLE = 0x5,
601 : FS_SELECT_T_STATS = 0x9,
602 : FS_SELECT_FLOW_TUPLE_STATS = 0xC,
603 : FS_SELECT_TS_FLOW_TUPLE = 0x6,
604 : FS_SELECT_TS_STATS = 0xA,
605 : FS_SELECT_T_FLOW_TUPLE_STATS = 0xD,
606 : FS_SELECT_TS_FLOW_TUPLE_STATS = 0xE
607 : };
608 :
609 : uint8_t flowseries_query_type() {
610 : return fs_query_type_;
611 : }
612 :
613 138 : bool is_present_in_select_column_fields(const std::string& field) {
614 138 : std::vector<std::string>::iterator it;
615 138 : it = std::find(select_column_fields.begin(),
616 : select_column_fields.end(),
617 : field);
618 138 : if (it == select_column_fields.end()) {
619 0 : return false;
620 : }
621 138 : return true;
622 : }
623 395 : bool ObjectIdQuery() {
624 739 : return ((select_column_fields.size() == 1 &&
625 739 : select_column_fields[0] == g_viz_constants.OBJECT_ID));
626 : }
627 : friend class SelectTest;
628 : private:
629 : bool is_valid_select_field(const std::string& select_field) const;
630 : //
631 : // Object table query
632 : //
633 : void get_query_column_value(const GenDb::DbDataValueVec &info,
634 : unsigned int index,
635 : std::string *query_column,
636 : GenDb::DbDataValue *value,
637 : std::string *object_id);
638 : bool process_object_query_specific_select_params(
639 : const std::string& sel_field,
640 : std::map<std::string, GenDb::DbDataValue>& col_res_map,
641 : std::map<std::string, std::string>& cmap,
642 : const boost::uuids::uuid& uuid,
643 : std::map<boost::uuids::uuid, std::string>&);
644 :
645 : // For flow class id in select field
646 :
647 : // flow class id to flow tuple map
648 : std::map<size_t, flow_tuple> flow_class_id_map;
649 :
650 : //
651 : // Flow Series Query
652 : //
653 : static const uint64_t kMicrosecInSec = 1000 * 1000;
654 :
655 : uint8_t fs_query_type_;
656 : bool is_flow_tuple_specified();
657 : void evaluate_fs_query_type();
658 : typedef void (SelectQuery::*process_fs_query_callback)(const uint64_t&,
659 : const boost::uuids::uuid&, const flow_stats&, const flow_tuple&);
660 : typedef void (SelectQuery::*populate_fs_result_callback)();
661 : typedef std::map<uint8_t, process_fs_query_callback>
662 : process_fs_query_cb_map_t;
663 : static process_fs_query_cb_map_t process_fs_query_cb_map_;
664 : static process_fs_query_cb_map_t process_fs_query_cb_map_init();
665 : typedef std::map<uint8_t, populate_fs_result_callback>
666 : populate_fs_result_cb_map_t;
667 : static populate_fs_result_cb_map_t populate_fs_result_cb_map_;
668 : static populate_fs_result_cb_map_t populate_fs_result_cb_map_init();
669 : typedef std::map<const boost::uuids::uuid, uuid_flow_stats>
670 : fs_uuid_stats_map_t;
671 :
672 : // Called from process_query() for FLOW SERIES Query
673 : query_status_t process_fs_query(process_fs_query_callback,
674 : populate_fs_result_callback);
675 :
676 : // flowclass is populated from tuple based on the
677 : // tuple fields in the select_column_fields
678 : void get_flow_class(const flow_tuple& tuple, flow_tuple& flowclass);
679 :
680 : void fs_write_final_result_row(const uint64_t *t, const flow_tuple *tuple,
681 : const flow_stats *raw_stats, const flow_stats *sum_stats,
682 : const flow_stats *avg_stats,
683 : const std::set<boost::uuids::uuid> *flow_list = NULL);
684 :
685 : uint64_t fs_get_time_slice(const uint64_t& t);
686 : // Common Flow series queries
687 :
688 : // 1. SELECT with T=<granularity>, stats fields
689 : typedef std::map<const uint64_t, flow_stats> fs_ts_stats_map_t;
690 : fs_ts_stats_map_t fs_ts_stats_map_;
691 : void process_fs_query_with_ts_stats_fields(const uint64_t&,
692 : const boost::uuids::uuid&, const flow_stats&, const flow_tuple&);
693 : void populate_fs_query_result_with_ts_stats_fields();
694 :
695 : // 2. SELECT with flow tuple fields, stats fields
696 : typedef std::map<const flow_tuple, flow_stats> fs_tuple_stats_map_t;
697 : fs_tuple_stats_map_t fs_tuple_stats_map_;
698 : void process_fs_query_with_tuple_stats_fields(const uint64_t&,
699 : const boost::uuids::uuid&, const flow_stats&, const flow_tuple&);
700 : void populate_fs_query_result_with_tuple_stats_fields();
701 :
702 : // 3. SELECT with T=<granularity>, flow tuple fields, stats fields
703 : typedef std::map<const flow_tuple, fs_ts_stats_map_t>
704 : fs_ts_tuple_stats_map_t;
705 : fs_ts_tuple_stats_map_t fs_ts_tuple_stats_map_;
706 : void process_fs_query_with_ts_tuple_stats_fields(const uint64_t&,
707 : const boost::uuids::uuid&, const flow_stats&, const flow_tuple&);
708 : void populate_fs_query_result_with_ts_tuple_stats_fields();
709 :
710 : // Rare Flow series queries
711 :
712 : // 1. SELECT with stats fields
713 : flow_stats fs_flow_stats_;
714 : void process_fs_query_with_stats_fields(const uint64_t&,
715 : const boost::uuids::uuid&, const flow_stats&, const flow_tuple&);
716 : void populate_fs_query_result_with_stats_fields();
717 :
718 : // 2. SELECT with T=<granularity>, flow tuple fields
719 : typedef std::map<const uint64_t, std::set<flow_tuple> > fs_ts_tuple_map_t;
720 : fs_ts_tuple_map_t fs_ts_tuple_map_;
721 : void process_fs_query_with_ts_tuple_fields(const uint64_t&,
722 : const boost::uuids::uuid&, const flow_stats&, const flow_tuple&);
723 : void populate_fs_query_result_with_ts_tuple_fields();
724 :
725 : // 3. SELECT with flow tuple fields
726 : void process_fs_query_with_tuple_fields(const uint64_t&,
727 : const boost::uuids::uuid&, const flow_stats&, const flow_tuple&);
728 : void populate_fs_query_result_with_tuple_fields();
729 :
730 : void process_fs_query_with_time(const uint64_t&,
731 : const boost::uuids::uuid&, const flow_stats&, const flow_tuple&);
732 : void populate_fs_query_result_with_time();
733 :
734 : // 4. SELECT with T=<granularity>
735 : typedef std::set<uint64_t> fs_ts_t;
736 : fs_ts_t fs_ts_list_;
737 : void process_fs_query_with_ts(const uint64_t&,
738 : const boost::uuids::uuid&, const flow_stats&, const flow_tuple&);
739 : void populate_fs_query_result_with_ts();
740 :
741 : // 5. SELECT with T
742 : // 6. SELECT with T, flow tuple fields
743 : // 7. SELECT with T, stats fields
744 : // 8. SELECT with T, flow tuple, stats
745 : void process_fs_query_with_time_tuple_stats_fields(const uint64_t&,
746 : const boost::uuids::uuid&, const flow_stats&, const flow_tuple&);
747 : void populate_fs_query_result_with_time_tuple_stats_fields();
748 : };
749 :
750 :
751 : struct sort_field_t {
752 126 : sort_field_t(const std::string& sort_name, const std::string& datatype) :
753 126 : name(sort_name), type(datatype) {
754 126 : }
755 : std::string name;
756 : std::string type;
757 : };
758 :
759 : // this data structure is passed on to post-processing module
760 : class PostProcessingQuery: public QueryUnit {
761 : public:
762 : // Initialize with the JSON string hashset received from REDIS
763 : PostProcessingQuery(const std::map<std::string, std::string>& json_api_data,
764 : QueryUnit *main_query);
765 :
766 : virtual query_status_t process_query();
767 :
768 : // Query related fields
769 :
770 : std::string json_string_;
771 :
772 : // filter list is an OR of ANDs
773 : std::vector<std::vector<filter_match_t> > filter_list;
774 :
775 : // Whether to sort the table or not
776 : bool sorted;
777 :
778 : // Type of sorting
779 : sort_op sorting_type;
780 : // fields to sort on (these are actual Cassandra column names)
781 : std::vector<sort_field_t> sort_fields;
782 : int limit; // number of entries
783 :
784 : // result after post processing
785 :
786 : std::unique_ptr<BufT> result_;
787 : std::unique_ptr<MapBufT> mresult_;
788 :
789 : bool sort_field_comparator(const QEOpServerProxy::ResultRowT& lhs,
790 : const QEOpServerProxy::ResultRowT& rhs);
791 :
792 : bool merge_processing(
793 : const QEOpServerProxy::BufferT& input,
794 : QEOpServerProxy::BufferT& output);
795 : bool final_merge_processing(
796 : const std::vector<boost::shared_ptr<QEOpServerProxy::BufferT> >& inputs,
797 : QEOpServerProxy::BufferT& output);
798 : };
799 :
800 : class StatsQuery;
801 :
802 : class AnalyticsQuery: public QueryUnit {
803 : public:
804 : AnalyticsQuery(const std::string& qid, std::map<std::string,
805 : std::string>& json_api_data,
806 : int or_number,
807 : const std::vector<query_result_unit_t> * where_info,
808 : const TtlMap& ttlmap,
809 : EventManager *evm, std::vector<std::string> cassandra_ips,
810 : std::vector<int> cassandra_ports, int batch,
811 : int total_batches, const std::string& cassandra_user,
812 : const std::string &cassandra_password,
813 : QueryEngine *qe, void * pipeline_handle = NULL);
814 : AnalyticsQuery(const std::string& qid, GenDbIfPtr dbif,
815 : std::map<std::string, std::string> json_api_data,
816 : int or_number,
817 : const std::vector<query_result_unit_t> * where_info,
818 : const TtlMap& ttlmap, int batch, int total_batches,
819 : QueryEngine *qe, void * pipeline_handle = NULL);
820 4026 : virtual ~AnalyticsQuery() {}
821 :
822 : virtual query_status_t process_query();
823 :
824 : // Interface to Cassandra
825 : GenDbIfPtr dbif_;
826 : void db_err_handler() {};
827 :
828 : //Query related fields
829 :
830 : // Query Id
831 : std::string query_id;
832 :
833 : // if the req is to get objectids, the following will contain the key
834 : std::string object_value_key;
835 :
836 : std::string sandesh_moduleid; // module id of the query engine itself
837 : bool filter_qe_logs; // whether to filter query engine logs
838 :
839 : // final result of the query
840 : std::unique_ptr<QEOpServerProxy::BufferT> final_result;
841 : std::unique_ptr<QEOpServerProxy::OutRowMultimapT> final_mresult;
842 :
843 : std::map<std::string, std::string> json_api_data_;
844 : const std::vector<query_result_unit_t> * where_info_;
845 :
846 : TtlMap ttlmap_;
847 : uint64_t where_start_;
848 : uint64_t select_start_;
849 : uint64_t postproc_start_;
850 : QEOpServerProxy::QPerfInfo qperf_;
851 :
852 : WhereQuery *wherequery_;
853 : SelectQuery *selectquery_;
854 : PostProcessingQuery *postprocess_;
855 :
856 : // For parallelization
857 : // Values from the original query
858 : // start time of the time range for the queried records
859 : uint64_t original_from_time;
860 : // end time of the time range for the queried records
861 : uint64_t original_end_time;
862 : // whether results from parallel instances need to be merged
863 : bool merge_needed; // whether sorting/limit operations are needed
864 : // which parallel batch number is this
865 : int parallel_batch_num;
866 : // total number of parallel batches
867 : int total_parallel_batches;
868 : // whether any processing is needed
869 : bool processing_needed;
870 : // time slice for each parallel instance
871 : uint64_t time_slice;
872 : // shared ptr to query engine needed when where_query winds up
873 : QueryEngine* qe_;
874 : // outer pipeline handle, needed while calling the QEResult from
875 : // where_query context
876 : void *handle_;
877 : // this is for merge between multiple instances running on same core
878 : bool merge_processing(const QEOpServerProxy::BufferT& input,
879 : QEOpServerProxy::BufferT& output);
880 : // this is for merge between instances running on multiple cores
881 : bool final_merge_processing(
882 : const std::vector<boost::shared_ptr<QEOpServerProxy::BufferT> >& inputs,
883 : QEOpServerProxy::BufferT& output);
884 :
885 : // this is to get parallelization details once the query is parsed
886 : void get_query_details(bool& is_merge_needed, bool& is_map_output,
887 : std::vector<uint64_t>& chunk_sizes,
888 : std::string& where, uint32_t& wterms,
889 : std::string& select,
890 : std::string& post,
891 : uint64_t& time_period,
892 : int& parse_status);
893 :
894 456799 : virtual std::string table() const {
895 456799 : return table_;
896 : }
897 0 : virtual uint64_t req_from_time() const {
898 0 : return req_from_time_;
899 : }
900 0 : virtual uint64_t req_end_time() const {
901 0 : return req_end_time_;
902 : }
903 8162 : virtual uint64_t from_time() const {
904 8162 : return from_time_;
905 : }
906 8009 : virtual uint64_t end_time() const {
907 8009 : return end_time_;
908 : }
909 0 : virtual const std::vector<query_result_unit_t>& where_query_result() {
910 0 : return *where_info_;
911 : }
912 0 : virtual uint32_t direction_ing() const {
913 0 : return wherequery_->direction_ing;
914 : }
915 :
916 : // validation functions
917 : virtual bool is_object_table_query(const std::string &tname);
918 : static bool is_message_table_query(const std::string &tname);
919 : bool is_message_table_query();
920 : static bool is_stat_table_query(const std::string& tname);
921 : static bool is_stat_fieldnames_table_query(const std::string& tname);
922 : static bool is_session_query(const std::string& tname); // either flow-series or flow-records query
923 : static bool is_flow_query(const std::string& tname); // either flow-series or flow-records query
924 : bool is_valid_where_field(const std::string& where_field);
925 : bool is_valid_sort_field(const std::string& sort_field);
926 : std::string get_column_field_datatype(const std::string& col_field);
927 15 : virtual bool is_query_parallelized() { return parallelize_query_; }
928 :
929 3446 : const StatsQuery& stats(void) const { return *stats_; }
930 : std::string stat_name_attr; // will be populated only for stats query
931 : private:
932 : std::unique_ptr<StatsQuery> stats_;
933 : // Analytics table to query
934 : std::string table_;
935 : // query start time requested by the user
936 : uint64_t req_from_time_;
937 : // query end time requested by the user
938 : uint64_t req_end_time_;
939 : // start time of the time range for the queried records.
940 : // If the start time requested by the user is earlier than the
941 : // analytics start time, then this field holds the analytics start time.
942 : // Else, from_time is same as req_from_time.
943 : uint64_t from_time_;
944 : // end time of the time range for the queried records.
945 : // If the end time requested by the user is later than the time @ which the
946 : // query was received, then this field holds the time @ which the query
947 : // was received. Else, end_time is same as req_end_time.
948 : uint64_t end_time_;
949 : bool parallelize_query_;
950 : // Init function
951 : void Init(const std::string& qid,
952 : const std::map<std::string, std::string>& json_api_data,
953 : int32_t or_number);
954 : bool can_parallelize_query();
955 : void ParseStatName(std::string &stat_table_name);
956 : };
957 :
958 : // limit on the size of query result we can handle
959 : static const int query_result_size_limit = 25000000;
960 :
961 : // main class
962 : class QueryEngine {
963 : public:
964 : static const uint64_t StartTimeDiffInSec = 12*3600;
965 : static int max_slice_;
966 :
967 : struct QueryParams {
968 127 : QueryParams(std::string qi,
969 : std::map<std::string, std::string> qu,
970 127 : uint32_t ch, uint64_t tm) :
971 127 : qid(qi), terms(qu), maxChunks(ch), query_starttm(tm) {}
972 1524 : QueryParams() {}
973 : std::string qid;
974 : std::map<std::string, std::string> terms;
975 : uint32_t maxChunks;
976 : uint64_t query_starttm;
977 : };
978 :
979 : uint64_t stime;
980 : int max_tasks_;
981 :
982 : QueryEngine(EventManager *evm,
983 : std::vector<std::string> cassandra_ips,
984 : std::vector<int> cassandra_ports,
985 : std::vector<std::string> redis_ip_ports,
986 : const std::string & redis_password,
987 : const bool redis_ssl_enable,
988 : const std::string & redis_keyfile,
989 : const std::string & redis_certfile,
990 : const std::string & redis_ca_cert,
991 : int max_tasks, int max_slice,
992 : const std::string & cassandra_name,
993 : const std::string & cassandra_password,
994 : bool cassandra_use_ssl,
995 : const std::string& cassandra_ca_certs,
996 : const std::string & cluster_id,
997 : const std::string &host_ip);
998 :
999 : QueryEngine(EventManager *evm,
1000 : std::vector<std::string> redis_ip_ports,
1001 : const std::string & redis_password,
1002 : const bool redis_ssl_enable,
1003 : const std::string & redis_keyfile,
1004 : const std::string & redis_certfile,
1005 : const std::string & redis_ca_cert,
1006 : int max_tasks, int max_slice,
1007 : const std::string & cassandra_user,
1008 : const std::string & cassandra_password,
1009 : bool cassandra_use_ssl,
1010 : const std::string& cassandra_ca_certs,
1011 : const std::string &host_ip);
1012 :
1013 : // This constructor used only for test purpose
1014 5 : QueryEngine(){}
1015 :
1016 : virtual ~QueryEngine();
1017 :
1018 : int
1019 : QueryPrepare(QueryParams qp,
1020 : std::vector<uint64_t> &chunk_size,
1021 : bool & need_merge, bool & map_output,
1022 : std::string& where, uint32_t& wterms,
1023 : std::string& select, std::string& post,
1024 : uint64_t& time_period,
1025 : std::string &table);
1026 :
1027 : // Query Execution of WHERE term
1028 : bool
1029 : QueryExecWhere(void * handle, QueryParams qp, uint32_t chunk,
1030 : uint32_t or_number);
1031 :
1032 : void
1033 : WhereQueryResult(AnalyticsQuery *q);
1034 :
1035 : // Query Execution of SELECT and post-processing
1036 : bool
1037 : QueryExec(void * handle, QueryParams qp, uint32_t chunk,
1038 : const std::vector<query_result_unit_t> *wi);
1039 :
1040 : bool
1041 : QueryAccumulate(QueryParams qp,
1042 : const QEOpServerProxy::BufferT& input,
1043 : QEOpServerProxy::BufferT& output);
1044 :
1045 : bool
1046 : QueryFinalMerge(QueryParams qp,
1047 : const std::vector<boost::shared_ptr<QEOpServerProxy::BufferT> >& inputs,
1048 : QEOpServerProxy::BufferT& output);
1049 :
1050 : bool
1051 : QueryFinalMerge(QueryParams qp,
1052 : const std::vector<boost::shared_ptr<QEOpServerProxy::OutRowMultimapT> >& inputs,
1053 : QEOpServerProxy::OutRowMultimapT& output);
1054 :
1055 : // Unit test function
1056 : void QueryEngine_Test();
1057 :
1058 : void db_err_handler() {};
1059 : TtlMap& GetTTlMap() { return ttlmap_; }
1060 0 : const std::string & keyspace() { return keyspace_; }
1061 : GenDb::DbTableStatistics stable_stats_;
1062 : mutable std::mutex smutex_;
1063 : bool GetCumulativeStats(std::vector<GenDb::DbTableInfo> *vdbti,
1064 : GenDb::DbErrors *dbe, std::vector<GenDb::DbTableInfo> *vstats_dbti)
1065 : const;
1066 : bool GetDiffStats(std::vector<GenDb::DbTableInfo> *vdbti,
1067 : GenDb::DbErrors *dbe, std::vector<GenDb::DbTableInfo> *vstats_dbti);
1068 : bool GetCqlStats(cass::cql::DbStats *stats) const;
1069 : bool GetCqlMetrics(cass::cql::Metrics *matrics) const;
1070 39 : GenDbIfPtr GetDbHandler() { return dbif_; }
1071 :
1072 : private:
1073 : GenDbIfPtr dbif_;
1074 : boost::scoped_ptr<QEOpServerProxy> qosp_;
1075 : EventManager *evm_;
1076 : std::vector<int> cassandra_ports_;
1077 : std::vector<std::string> cassandra_ips_;
1078 : std::string cassandra_user_;
1079 : std::string cassandra_password_;
1080 : bool cassandra_use_ssl_;
1081 : std::string cassandra_ca_certs_;
1082 : TtlMap ttlmap_;
1083 : std::string keyspace_;
1084 : };
1085 :
1086 : #endif
|
