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