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 <atomic>
7 : #include <cstdlib>
8 : #include <cerrno>
9 : #include <utility>
10 : #include <list>
11 : #include <mutex>
12 :
13 : #include <boost/bind/bind.hpp>
14 : #include <boost/asio.hpp>
15 : #include <boost/tuple/tuple.hpp>
16 : #include "base/util.h"
17 : #include "base/logging.h"
18 : #include "base/address_util.h"
19 : #include "hiredis/hiredis.h"
20 : #include "hiredis/hiredis_ssl.h"
21 : #include "hiredis/boostasio.hpp"
22 : #include <contrail-collector/redis_connection.h>
23 : #include "base/work_pipeline.h"
24 : #include "QEOpServerProxy.h"
25 : #include "rapidjson/document.h"
26 : #include "rapidjson/stringbuffer.h"
27 : #include "rapidjson/writer.h"
28 : #include "query.h"
29 : #include "stats_query.h"
30 : #include "analytics_types.h"
31 : #include "stats_select.h"
32 : #include <base/connection_info.h>
33 :
34 : using std::list;
35 : using std::string;
36 : using std::vector;
37 : using std::map;
38 : using boost::assign::list_of;
39 : using boost::ptr_map;
40 : using boost::nullable;
41 : using boost::tuple;
42 : using boost::shared_ptr;
43 : using boost::scoped_ptr;
44 : using std::pair;
45 : using std::unique_ptr;
46 : using std::make_pair;
47 : using process::ConnectionState;
48 : using process::ConnectionType;
49 : using process::ConnectionStatus;
50 : using process::Endpoint;
51 : using namespace boost::placeholders;
52 :
53 : extern RedisAsyncConnection * rac_alloc(EventManager *, const std::string & ,unsigned short,
54 : RedisAsyncConnection::ClientConnectCbFn ,
55 : RedisAsyncConnection::ClientDisconnectCbFn,
56 : const bool, const std::string &, const std::string &, const std::string &);
57 :
58 : extern RedisAsyncConnection * rac_alloc_nocheck(EventManager *, const std::string & ,unsigned short,
59 : RedisAsyncConnection::ClientConnectCbFn ,
60 : RedisAsyncConnection::ClientDisconnectCbFn,
61 : const bool, const std::string &, const std::string &, const std::string &);
62 :
63 : SandeshTraceBufferPtr QeTraceBuf(SandeshTraceBufferCreate(QE_TRACE_BUF, 10000));
64 :
65 : struct RawResultT {
66 : QEOpServerProxy::QPerfInfo perf;
67 : shared_ptr<QEOpServerProxy::BufferT> res;
68 : shared_ptr<QEOpServerProxy::OutRowMultimapT> mres;
69 : shared_ptr<WhereResultT> wres;
70 : };
71 :
72 : typedef pair<redisReply,vector<string> > RedisT;
73 :
74 3408 : bool RedisAsyncArgCommand(RedisAsyncConnection * rac,
75 : void *rpi, const vector<string>& args) {
76 :
77 3408 : return rac->RedisAsyncArgCmd(rpi, args);
78 : }
79 :
80 : class QEOpServerProxy::QEOpServerImpl {
81 : public:
82 : typedef std::vector<std::string> QEOutputT;
83 :
84 : struct Input {
85 : int redis_host_idx;
86 : int cnum;
87 : string hostname;
88 : QueryEngine::QueryParams qp;
89 : vector<uint64_t> chunk_size;
90 : uint32_t wterms;
91 : bool need_merge;
92 : bool map_output;
93 : string where;
94 : string select;
95 : string post;
96 : uint64_t time_period;
97 : string table;
98 : uint32_t max_rows;
99 : std::atomic<uint32_t> chunk_q;
100 : std::atomic<uint32_t> total_rows;
101 :
102 : // to handle absent operator= for std::atomic
103 1251 : Input& operator=(const Input& other) {
104 1251 : if (this != &other) {
105 1251 : redis_host_idx = other.redis_host_idx;
106 1251 : cnum = other.cnum;
107 1251 : hostname = other.hostname;
108 1254 : qp = other.qp;
109 1227 : chunk_size = other.chunk_size;
110 1229 : wterms = other.wterms;
111 1229 : need_merge = other.need_merge;
112 1229 : map_output = other.map_output;
113 1229 : where = other.where;
114 1235 : select = other.select;
115 1250 : post = other.post;
116 1253 : time_period = other.time_period;
117 1253 : table = other.table;
118 1257 : max_rows = other.max_rows;
119 : // these two are handled only in QueryExec and are not used in next stages - don't copy they
120 1257 : chunk_q = 0;
121 1257 : total_rows = 0;
122 : }
123 1259 : return *this;
124 : }
125 : };
126 :
127 3914 : void JsonInsert(std::vector<query_column> &columns,
128 : contrail_rapidjson::Document& dd,
129 : std::pair<const string,string> * map_it) {
130 :
131 3914 : bool found = false;
132 42568 : for (size_t j = 0; j < columns.size(); j++)
133 : {
134 38654 : if ((0 == map_it->first.compare(0,5,string("COUNT")))) {
135 0 : contrail_rapidjson::Value val(contrail_rapidjson::kNumberType);
136 0 : unsigned long num = 0;
137 0 : stringToInteger(map_it->second, num);
138 0 : val.SetUint64(num);
139 0 : contrail_rapidjson::Value vk;
140 0 : dd.AddMember(vk.SetString(map_it->first.c_str(), dd.GetAllocator()), val, dd.GetAllocator());
141 0 : found = true;
142 38654 : } else if (columns[j].name == map_it->first) {
143 3914 : if (map_it->second.length() == 0) {
144 12 : contrail_rapidjson::Value val(contrail_rapidjson::kNullType);
145 12 : contrail_rapidjson::Value vk;
146 12 : dd.AddMember(vk.SetString(map_it->first.c_str(), dd.GetAllocator()), val, dd.GetAllocator());
147 12 : found = true;
148 12 : continue;
149 12 : }
150 :
151 : // find out type and convert
152 4838 : if (columns[j].datatype == "string" ||
153 936 : columns[j].datatype == "uuid")
154 : {
155 2966 : contrail_rapidjson::Value val(contrail_rapidjson::kStringType);
156 2966 : val.SetString(map_it->second.c_str(), dd.GetAllocator());
157 2966 : contrail_rapidjson::Value vk;
158 2966 : dd.AddMember(vk.SetString(map_it->first.c_str(), dd.GetAllocator()), val, dd.GetAllocator());
159 3902 : } else if (columns[j].datatype == "ipaddr") {
160 0 : contrail_rapidjson::Value val(contrail_rapidjson::kStringType);
161 0 : val.SetString(map_it->second.c_str(),
162 : dd.GetAllocator());
163 0 : contrail_rapidjson::Value vk;
164 0 : dd.AddMember(vk.SetString(map_it->first.c_str(), dd.GetAllocator()), val, dd.GetAllocator());
165 :
166 936 : } else if (columns[j].datatype == "double") {
167 0 : contrail_rapidjson::Value val(contrail_rapidjson::kNumberType);
168 0 : double dval = (double) strtod(map_it->second.c_str(), NULL);
169 0 : val.SetDouble(dval);
170 0 : contrail_rapidjson::Value vk;
171 0 : dd.AddMember(vk.SetString(map_it->first.c_str(), dd.GetAllocator()), val, dd.GetAllocator());
172 0 : } else {
173 936 : contrail_rapidjson::Value val(contrail_rapidjson::kNumberType);
174 936 : unsigned long num = 0;
175 936 : stringToInteger(map_it->second, num);
176 936 : val.SetUint64(num);
177 936 : contrail_rapidjson::Value vk;
178 936 : dd.AddMember(vk.SetString(map_it->first.c_str(), dd.GetAllocator()), val, dd.GetAllocator());
179 936 : }
180 3902 : found = true;
181 : }
182 : }
183 3914 : assert(found);
184 3914 : }
185 :
186 126 : void QueryJsonify(const string& table, bool map_output,
187 : const BufferT* raw_res, const OutRowMultimapT* raw_mres, QEOutputT* raw_json) {
188 :
189 126 : vector<OutRowT>::iterator res_it;
190 :
191 126 : std::vector<query_column> columns;
192 :
193 126 : if (!table.size()) return;
194 126 : bool found = false;
195 882 : for(size_t i = 0; i < g_viz_constants._TABLES.size(); i++)
196 : {
197 756 : if (g_viz_constants._TABLES[i].name == table) {
198 59 : found = true;
199 59 : columns = g_viz_constants._TABLES[i].schema.columns;
200 : }
201 : }
202 126 : if (!found) {
203 67 : found = true;
204 67 : columns = g_viz_constants._OBJECT_TABLE_SCHEMA.columns;
205 : }
206 126 : assert(found || map_output);
207 :
208 126 : if (map_output) {
209 70 : OutRowMultimapT::const_iterator mres_it;
210 264 : for (mres_it = raw_mres->begin(); mres_it != raw_mres->end(); ++mres_it) {
211 194 : string jstr;
212 194 : StatsSelect::Jsonify(table, mres_it->second.first, mres_it->second.second, jstr);
213 194 : raw_json->push_back(jstr);
214 194 : }
215 : } else {
216 56 : QEOpServerProxy::BufferT* raw_result =
217 : const_cast<QEOpServerProxy::BufferT*>(raw_res);
218 56 : QEOpServerProxy::BufferT::iterator res_it;
219 3152 : for (res_it = raw_result->begin(); res_it != raw_result->end(); ++res_it) {
220 3096 : std::map<std::string, std::string>::iterator map_it;
221 3096 : contrail_rapidjson::Document dd;
222 3096 : dd.SetObject();
223 :
224 3096 : for (map_it = (*res_it).first.begin();
225 7010 : map_it != (*res_it).first.end(); ++map_it) {
226 : // search for column name in the schema
227 3914 : JsonInsert(columns, dd, &(*map_it));
228 : }
229 3096 : contrail_rapidjson::StringBuffer sb;
230 3096 : contrail_rapidjson::Writer<contrail_rapidjson::StringBuffer> writer(sb);
231 3096 : dd.Accept(writer);
232 3096 : raw_json->push_back(sb.GetString());
233 3096 : }
234 : }
235 126 : }
236 :
237 :
238 904 : void QECallback(void * qid, QPerfInfo qperf,
239 : unique_ptr<std::vector<query_result_unit_t> > res) {
240 :
241 904 : RawResultT* raw(new RawResultT);
242 904 : raw->perf = qperf;
243 904 : raw->wres = std::move(res);
244 :
245 904 : ExternalProcIf<RawResultT> * rpi = NULL;
246 904 : if (qid)
247 904 : rpi = reinterpret_cast<ExternalProcIf<RawResultT> *>(qid);
248 :
249 904 : if (rpi) {
250 904 : unique_ptr<RawResultT> rp(raw);
251 904 : rpi->Response(std::move(rp));
252 904 : }
253 904 : }
254 :
255 856 : void QECallback(void * qid, QPerfInfo qperf, unique_ptr<QEOpServerProxy::BufferT> res,
256 : unique_ptr<QEOpServerProxy::OutRowMultimapT> mres) {
257 :
258 856 : RawResultT* raw(new RawResultT);
259 856 : raw->perf = qperf;
260 856 : raw->res = std::move(res);
261 856 : raw->mres = std::move(mres);
262 :
263 856 : ExternalProcIf<RawResultT> * rpi = NULL;
264 856 : if (qid)
265 856 : rpi = reinterpret_cast<ExternalProcIf<RawResultT> *>(qid);
266 :
267 856 : if (rpi) {
268 856 : unique_ptr<RawResultT> rp(raw);
269 856 : rpi->Response(std::move(rp));
270 856 : }
271 856 : }
272 :
273 : struct Stage0Out {
274 : Input inp;
275 : bool ret_code;
276 : vector<QPerfInfo> ret_info;
277 : vector<uint32_t> chunk_merge_time;
278 : shared_ptr<BufferT> result;
279 : shared_ptr<OutRowMultimapT> mresult;
280 : vector<shared_ptr<WhereResultT> > welem;
281 : shared_ptr<WhereResultT> wresult;
282 : uint32_t current_chunk;
283 : };
284 :
285 2737 : ExternalBase::Efn QueryExec(uint32_t inst, const vector<RawResultT*> & exts,
286 : const Input & inp, Stage0Out & res) {
287 2737 : uint32_t step = exts.size();
288 :
289 2749 : if (!step) {
290 988 : res.inp = inp;
291 1007 : res.ret_code = true;
292 :
293 1007 : if (inp.map_output)
294 560 : res.mresult = shared_ptr<OutRowMultimapT>(new OutRowMultimapT());
295 : else
296 447 : res.result = shared_ptr<BufferT>(new BufferT());
297 :
298 980 : res.wresult = shared_ptr<WhereResultT>(new WhereResultT());
299 2002 : for (size_t or_idx=0; or_idx<inp.wterms; or_idx++) {
300 1022 : shared_ptr<WhereResultT> ss;
301 1022 : res.welem.push_back(ss);
302 1026 : }
303 :
304 980 : Input& cinp = const_cast<Input&>(inp);
305 980 : res.current_chunk = cinp.chunk_q.fetch_add(1);
306 980 : const uint32_t chunknum = res.current_chunk;
307 980 : if (chunknum >= inp.chunk_size.size()) {
308 154 : return NULL;
309 : }
310 :
311 : // Update query status
312 851 : RedisAsyncConnection * rac = conns_[res.inp.redis_host_idx][res.inp.cnum].get();
313 847 : string rkey = "REPLY:" + res.inp.qp.qid;
314 : char stat[40];
315 844 : uint prg = 10 + (chunknum * 75)/inp.chunk_size.size();
316 845 : sprintf(stat,"{\"progress\":%d}", prg);
317 827 : RedisAsyncArgCommand(rac, NULL,
318 1698 : list_of(string("RPUSH"))(rkey)(stat));
319 :
320 1703 : return boost::bind(&QueryEngine::QueryExecWhere, qosp_->qe_,
321 1704 : _1, inp.qp, chunknum, 0);
322 853 : }
323 :
324 1761 : res.ret_info.push_back(exts[step-1]->perf);
325 1760 : if (exts[step-1]->perf.error) {
326 0 : res.ret_code =false;
327 : }
328 1760 : if (!res.ret_code) return NULL;
329 :
330 : // Number of substeps per chunk is the number of OR terms in WHERE
331 : // plus one more substep for select and post processing
332 1760 : uint32_t substep = step % (inp.wterms + 1);
333 :
334 1760 : if (substep == inp.wterms) {
335 : // Get the result of the final WHERE
336 856 : res.welem[substep-1] = exts[step-1]->wres;
337 :
338 : // The set "OR" API needs raw pointers
339 855 : vector<WhereResultT*> oterms;
340 1759 : for (size_t or_idx = 0; or_idx < inp.wterms; or_idx++) {
341 903 : oterms.push_back(res.welem[or_idx].get());
342 : }
343 :
344 : // Do SET operations
345 856 : QE_ASSERT(res.wresult->size() == 0);
346 856 : SetOperationUnit::op_or(res.inp.qp.qid, *res.wresult, oterms);
347 :
348 1760 : for (size_t or_idx=0; or_idx<inp.wterms; or_idx++) {
349 904 : res.welem[or_idx].reset();
350 : }
351 :
352 : // Start the SELECT and POST-processing
353 2568 : return boost::bind(&QueryEngine::QueryExec, qosp_->qe_,
354 1711 : _1, inp.qp, res.current_chunk, res.wresult.get());
355 :
356 1760 : } else if (substep == 0) {
357 : // A chunk is complete. Start another one
358 856 : res.wresult->clear();
359 : uint32_t added_rows;
360 :
361 856 : if (inp.need_merge) {
362 448 : uint64_t then = UTCTimestampUsec();
363 448 : if (inp.map_output) {
364 416 : uint32_t base_rows = res.mresult->size();
365 : // TODO: This interface should not be Stats-Specific
366 416 : StatsSelect::Merge(std::string(""), *(exts[step-1]->mres), *(res.mresult));
367 : // Some rows of this chunk will merge into existing results
368 416 : added_rows = res.mresult->size() - base_rows;
369 : } else {
370 32 : uint32_t base_rows = res.result->size();
371 32 : res.ret_code =
372 32 : qosp_->qe_->QueryAccumulate(inp.qp,
373 32 : *(exts[step-1]->res), *(res.result));
374 : // Some rows of this chunk will merge into existing results
375 32 : added_rows = res.result->size() - base_rows;
376 : }
377 896 : res.chunk_merge_time.push_back(
378 448 : static_cast<uint32_t>((UTCTimestampUsec() - then)/1000));
379 :
380 : } else {
381 : // TODO : When merge is not needed, we can just send
382 : // a result upto redis at this point.
383 :
384 408 : if (inp.map_output) {
385 41 : added_rows = exts[step-1]->mres->size();
386 41 : OutRowMultimapT::iterator jt = res.mresult->begin();
387 41 : for (OutRowMultimapT::const_iterator it = exts[step-1]->mres->begin();
388 79 : it != exts[step-1]->mres->end(); it++ ) {
389 :
390 76 : jt = res.mresult->insert(jt,
391 76 : std::make_pair(it->first, it->second));
392 : }
393 : } else {
394 367 : added_rows = exts[step-1]->res->size();
395 1101 : res.result->insert(res.result->begin(),
396 367 : exts[step-1]->res->begin(),
397 367 : exts[step-1]->res->end());
398 : }
399 : }
400 856 : Input& cinp = const_cast<Input&>(inp);
401 1712 : if (cinp.total_rows.fetch_add(added_rows) + added_rows > cinp.max_rows) {
402 0 : QE_LOG_NOQID(ERROR, "QueryExec Max Rows Exceeded " <<
403 : cinp.total_rows << " chunk " << cinp.chunk_q);
404 0 : return NULL;
405 : }
406 :
407 856 : res.current_chunk = cinp.chunk_q.fetch_add(1);
408 856 : const uint32_t chunknum = res.current_chunk;
409 856 : if (chunknum < inp.chunk_size.size()) {
410 2 : string key = "QUERY:" + res.inp.qp.qid;
411 :
412 : // Update query status
413 2 : RedisAsyncConnection * rac = conns_[res.inp.redis_host_idx][res.inp.cnum].get();
414 2 : string rkey = "REPLY:" + res.inp.qp.qid;
415 : char stat[40];
416 2 : uint prg = 10 + (chunknum * 75)/inp.chunk_size.size();
417 2 : sprintf(stat,"{\"progress\":%d}", prg);
418 2 : RedisAsyncArgCommand(rac, NULL,
419 4 : list_of(string("RPUSH"))(rkey)(stat));
420 4 : return boost::bind(&QueryEngine::QueryExecWhere, qosp_->qe_,
421 4 : _1, inp.qp, chunknum, 0);
422 2 : } else {
423 854 : return NULL;
424 : }
425 : } else {
426 : // We are in the middle of doing WHERE processing for a chunk
427 :
428 48 : res.welem[substep-1] = exts[step-1]->wres;
429 :
430 96 : return boost::bind(&QueryEngine::QueryExecWhere, qosp_->qe_,
431 96 : _1, inp.qp, res.current_chunk, substep);
432 : }
433 : return NULL;
434 : }
435 :
436 : struct Stage0Merge {
437 : Input inp;
438 : bool ret_code;
439 : bool overflow;
440 : uint32_t fm_time;
441 : vector<vector<QPerfInfo> > ret_info;
442 : vector<vector<uint32_t> > chunk_merge_time;
443 : BufferT result;
444 : OutRowMultimapT mresult;
445 : };
446 126 : bool QueryMerge(const std::vector<boost::shared_ptr<Stage0Out> > & subs,
447 : const boost::shared_ptr<Input> & inp, Stage0Merge & res) {
448 :
449 126 : res.ret_code = true;
450 126 : res.overflow = false;
451 126 : res.inp = subs[0]->inp;
452 126 : res.fm_time = 0;
453 :
454 126 : uint32_t rows = 0;
455 126 : for (vector<shared_ptr<Stage0Out> >::const_iterator it = subs.begin() ;
456 1134 : it!=subs.end(); it++) {
457 1008 : if (res.inp.map_output)
458 560 : rows += (*it)->mresult->size();
459 : else
460 448 : rows += (*it)->result->size();
461 : }
462 :
463 : // If max_rows have been exceeded, don't do any more processing
464 126 : if (rows > res.inp.max_rows) {
465 0 : res.overflow = true;
466 0 : return true;
467 : }
468 :
469 126 : std::vector<boost::shared_ptr<OutRowMultimapT> > mqsubs;
470 126 : std::vector<boost::shared_ptr<QEOpServerProxy::BufferT> > qsubs;
471 126 : for (vector<shared_ptr<Stage0Out> >::const_iterator it = subs.begin() ;
472 1134 : it!=subs.end(); it++) {
473 :
474 1008 : res.ret_info.push_back((*it)->ret_info);
475 1008 : res.chunk_merge_time.push_back((*it)->chunk_merge_time);
476 :
477 1008 : if ((*it)->ret_code == false) {
478 0 : res.ret_code = false;
479 : } else {
480 1008 : if (res.inp.map_output)
481 560 : mqsubs.push_back((*it)->mresult);
482 : else
483 448 : qsubs.push_back((*it)->result);
484 : }
485 : }
486 :
487 126 : if (!res.ret_code) return true;
488 :
489 126 : if (res.inp.need_merge) {
490 67 : uint64_t then = UTCTimestampUsec();
491 :
492 67 : if (res.inp.map_output) {
493 63 : res.ret_code =
494 63 : qosp_->qe_->QueryFinalMerge(res.inp.qp, mqsubs, res.mresult);
495 :
496 : } else {
497 4 : res.ret_code =
498 4 : qosp_->qe_->QueryFinalMerge(res.inp.qp, qsubs, res.result);
499 : }
500 :
501 67 : uint64_t now = UTCTimestampUsec();
502 67 : res.fm_time = static_cast<uint32_t>((now - then)/1000);
503 : } else {
504 : // TODO : If a merge was not needed, results have been sent to
505 : // redis already. The only thing still needed is the status
506 59 : for (vector<shared_ptr<Stage0Out> >::const_iterator it = subs.begin() ;
507 531 : it!=subs.end(); it++) {
508 :
509 472 : if (res.inp.map_output) {
510 56 : OutRowMultimapT::iterator jt = res.mresult.begin();
511 56 : for (OutRowMultimapT::const_iterator kt = (*it)->mresult->begin();
512 94 : kt != (*it)->mresult->end(); kt++) {
513 38 : jt = res.mresult.insert(jt,
514 76 : std::make_pair(kt->first, kt->second));
515 : }
516 : } else {
517 832 : res.result.insert(res.result.begin(),
518 416 : (*it)->result->begin(),
519 416 : (*it)->result->end());
520 : }
521 : }
522 : }
523 :
524 126 : return true;
525 126 : }
526 :
527 : struct Output {
528 : Input inp;
529 : uint32_t redis_time;
530 : bool ret_code;
531 : };
532 503 : ExternalBase::Efn QueryResp(uint32_t inst, const vector<RedisT*> & exts,
533 : const Stage0Merge & inp, Output & ret) {
534 503 : uint32_t step = exts.size();
535 503 : switch (inst) {
536 252 : case 0: {
537 252 : if (!step) {
538 126 : ret.inp = inp.inp;
539 126 : RedisAsyncConnection * rac = conns_[ret.inp.redis_host_idx][ret.inp.cnum].get();
540 126 : std::stringstream keystr;
541 126 : unique_ptr<QEOutputT> jsonresult(new QEOutputT);
542 :
543 126 : QE_LOG_NOQID(INFO, "Will Jsonify #rows " <<
544 : inp.result.size() + inp.mresult.size());
545 126 : QueryJsonify(inp.inp.table, inp.inp.map_output,
546 : &inp.result, &inp.mresult, jsonresult.get());
547 :
548 126 : vector<string> const * const res = jsonresult.get();
549 126 : vector<string>::size_type idx = 0;
550 126 : uint32_t rownum = 0;
551 :
552 126 : QE_LOG_NOQID(INFO, "Did Jsonify #rows " << res->size());
553 :
554 126 : uint64_t then = UTCTimestampUsec();
555 : char stat[80];
556 126 : string key = "REPLY:" + ret.inp.qp.qid;
557 :
558 126 : if (inp.overflow) {
559 0 : sprintf(stat,"{\"progress\":%d}", - ENOBUFS);
560 126 : } else if (!inp.ret_code) {
561 0 : sprintf(stat,"{\"progress\":%d}", - EIO);
562 : } else {
563 700 : while (idx < res->size()) {
564 574 : uint32_t rowsize = 0;
565 574 : keystr.str(string());
566 574 : keystr << "RESULT:" << ret.inp.qp.qid << ":" << rownum;
567 1148 : vector<string> command = list_of(string("RPUSH"))(keystr.str());
568 3864 : while ((idx < res->size()) && (((int)rowsize) < kMaxRowThreshold)) {
569 3290 : command.push_back(res->at(idx));
570 3290 : rowsize += res->at(idx).size();
571 3290 : idx++;
572 : }
573 574 : RedisAsyncArgCommand(rac, NULL, command);
574 574 : RedisAsyncArgCommand(rac, NULL,
575 1148 : list_of(string("EXPIRE"))(keystr.str())("300"));
576 574 : sprintf(stat,"{\"progress\":90, \"lines\":%d}",
577 : (int)rownum);
578 574 : RedisAsyncArgCommand(rac, NULL,
579 1148 : list_of(string("RPUSH"))(key)(stat));
580 574 : rownum++;
581 574 : }
582 126 : sprintf(stat,"{\"progress\":100, \"lines\":%d, \"count\":%d}",
583 126 : (int)rownum, (int)res->size());
584 : }
585 126 : uint64_t now = UTCTimestampUsec();
586 126 : ret.redis_time = static_cast<uint32_t>((now - then)/1000);
587 126 : QE_LOG_NOQID(DEBUG, "QE Query Result is " << stat);
588 252 : return boost::bind(&RedisAsyncArgCommand,
589 126 : conns_[ret.inp.redis_host_idx][ret.inp.cnum].get(), _1,
590 378 : list_of(string("RPUSH"))(key)(stat));
591 126 : } else {
592 126 : RedisAsyncConnection * rac = conns_[ret.inp.redis_host_idx][ret.inp.cnum].get();
593 126 : string key = "REPLY:" + ret.inp.qp.qid;
594 126 : RedisAsyncArgCommand(rac, NULL,
595 252 : list_of(string("EXPIRE"))(key)("300"));
596 :
597 126 : key = "QUERY:" + ret.inp.qp.qid;
598 126 : RedisAsyncArgCommand(rac, NULL,
599 252 : list_of(string("EXPIRE"))(key)("300"));
600 :
601 126 : uint64_t now = UTCTimestampUsec();
602 126 : uint32_t qtime = static_cast<uint32_t>(
603 126 : (now - ret.inp.qp.query_starttm)/1000);
604 :
605 126 : uint64_t enqtm = atol(ret.inp.qp.terms["enqueue_time"].c_str());
606 126 : uint32_t enq_delay = static_cast<uint32_t>(
607 126 : (ret.inp.qp.query_starttm - enqtm)/1000);
608 :
609 126 : QueryStats qs;
610 : size_t outsize;
611 :
612 126 : if (ret.inp.map_output)
613 70 : outsize = inp.mresult.size();
614 : else
615 56 : outsize = inp.result.size();
616 :
617 126 : qs.set_rows(static_cast<uint32_t>(outsize));
618 126 : qs.set_time(qtime);
619 126 : qs.set_qid(ret.inp.qp.qid);
620 126 : qs.set_chunks(inp.inp.chunk_size.size());
621 126 : std::ostringstream wherestr, selstr, poststr;
622 1134 : for (size_t i=0; i < inp.ret_info.size(); i++) {
623 2768 : for (size_t j=0; j < inp.ret_info[i].size(); j++) {
624 1760 : wherestr << inp.ret_info[i][j].chunk_where_time << ",";
625 1760 : selstr << inp.ret_info[i][j].chunk_select_time << ",";
626 1760 : poststr << inp.ret_info[i][j].chunk_postproc_time << ",";
627 : }
628 1008 : wherestr << " ";
629 1008 : selstr << " ";
630 1008 : poststr << " ";
631 : }
632 126 : if (inp.overflow) {
633 0 : qs.set_error("ERROR-ENOBUFS");
634 126 : } else if (!inp.ret_code) {
635 0 : qs.set_error("ERROR-EIO");
636 : } else {
637 126 : qs.set_error("None");
638 : }
639 126 : qs.set_chunk_where_time(wherestr.str());
640 126 : qs.set_chunk_select_time(selstr.str());
641 126 : qs.set_chunk_postproc_time(poststr.str());
642 :
643 126 : std::ostringstream mergestr;
644 1134 : for (size_t i=0; i < inp.chunk_merge_time.size(); i++) {
645 1456 : for (size_t j=0; j < inp.chunk_merge_time[i].size(); j++) {
646 448 : mergestr << inp.chunk_merge_time[i][j] << ",";
647 : }
648 1008 : mergestr << " ";
649 : }
650 :
651 126 : qs.set_chunk_merge_time(mergestr.str());
652 126 : qs.set_final_merge_time(inp.fm_time);
653 126 : qs.set_where(inp.inp.where);
654 126 : qs.set_select(inp.inp.select);
655 126 : qs.set_post(inp.inp.post);
656 126 : qs.set_time_span(static_cast<uint32_t>(inp.inp.time_period));
657 126 : qs.set_enq_delay(enq_delay);
658 126 : QUERY_PERF_INFO_SEND(Sandesh::source(), // name
659 : inp.inp.table, // table
660 : qs);
661 :
662 126 : QE_LOG_NOQID(INFO, "Finished: QID " << ret.inp.qp.qid <<
663 : " Table " << inp.inp.table <<
664 : " Time(ms) " << qtime <<
665 : " RedisTime(ms) " << ret.redis_time <<
666 : " MergeTime(ms) " << inp.fm_time <<
667 : " Rows " << outsize <<
668 : " EnQ-delay" << enq_delay);
669 :
670 126 : ret.ret_code = true;
671 126 : }
672 : }
673 126 : break;
674 252 : case 1: {
675 252 : if (!step) {
676 126 : string key = "ENGINE:" + inp.inp.hostname;
677 252 : return boost::bind(&RedisAsyncArgCommand,
678 126 : conns_[inp.inp.redis_host_idx][inp.inp.cnum].get(), _1,
679 378 : list_of(string("LREM"))(key)("0")(inp.inp.qp.qid));
680 126 : } else {
681 126 : ret.ret_code = true;
682 : }
683 : }
684 126 : break;
685 : }
686 251 : return NULL;
687 : }
688 :
689 :
690 : typedef WorkPipeline<Input, Stage0Merge, Output> QEPipeT;
691 :
692 126 : void QEPipeCb(QEPipeT *wp, bool ret_code) {
693 126 : std::scoped_lock lock(mutex_);
694 :
695 126 : boost::shared_ptr<Output> res = wp->Result();
696 126 : assert(pipes_.find(res->inp.qp.qid)->second == wp);
697 126 : pipes_.erase(res->inp.qp.qid);
698 126 : m_analytics_queries.erase(res->inp.qp.qid);
699 126 : npipes_[res->inp.redis_host_idx][res->inp.cnum-1]--;
700 126 : QE_LOG_NOQID(DEBUG, " Result " << res->ret_code << " , " << res->inp.cnum << " conn");
701 126 : delete wp;
702 126 : }
703 :
704 356 : void ConnUpPostProcess(int redis_host_idx, uint8_t cnum) {
705 356 : if (!connState_[redis_host_idx][cnum]) {
706 230 : QE_LOG_NOQID(DEBUG, "ConnUp SetCB" << (uint32_t)cnum);
707 230 : cb_proc_fn_[redis_host_idx][cnum] = boost::bind(&QEOpServerImpl::CallbackProcess,
708 230 : this, redis_host_idx, cnum, _1, _2, _3);
709 230 : conns_[redis_host_idx][cnum].get()->SetClientAsyncCmdCb(cb_proc_fn_[redis_host_idx][cnum]);
710 230 : connState_[redis_host_idx][cnum] = true;
711 : }
712 :
713 356 : bool isConnected = true;
714 2136 : for(int i=0; i<kConnections+1; i++)
715 1780 : if (!connState_[redis_host_idx][i]) isConnected = false;
716 356 : if (isConnected) {
717 172 : string key = "ENGINE:" + hostname_;
718 344 : conns_[redis_host_idx][0].get()->RedisAsyncArgCmd(0,
719 344 : list_of(string("BRPOPLPUSH"))("QUERYQ")(key)("0"));
720 172 : }
721 356 : }
722 :
723 126 : int LeastLoadedConnection(int redis_host_idx) {
724 126 : int minvalue = npipes_[redis_host_idx][0];
725 126 : int minindex = 0;
726 504 : for (int i = 1; i < kConnections; i++) {
727 378 : if (npipes_[redis_host_idx][i] < minvalue) {
728 0 : minvalue = npipes_[redis_host_idx][i];
729 0 : minindex = i;
730 : }
731 : }
732 126 : return minindex;
733 : }
734 :
735 0 : void QueryError(int redis_host_idx, string qid, int ret_code) {
736 0 : string redis_host = redis_host_port_pairs_[redis_host_idx].first;
737 0 : int redis_port = redis_host_port_pairs_[redis_host_idx].second;
738 0 : redisContext *c = redisConnect(redis_host.c_str(), redis_port);
739 0 : if (c->err) {
740 0 : QE_LOG_NOQID(ERROR, "Cannot report query error for " << qid <<
741 : " . No Redis Connection");
742 0 : redisFree(c);
743 0 : return;
744 : }
745 :
746 : /* Secure the connection if SSL enabled */
747 0 : if (redis_ssl_enable_) {
748 0 : redisSSLContextError ssl_error = REDIS_SSL_CTX_NONE;
749 0 : redisSSLContext *ssl_ctx = redisCreateSSLContext(
750 : redis_ca_cert_.c_str(),
751 : NULL,
752 : redis_certfile_.c_str(),
753 : redis_keyfile_.c_str(),
754 : "sni",
755 : &ssl_error);
756 0 : if (!ssl_ctx || ssl_error != REDIS_SSL_CTX_NONE) {
757 0 : QE_LOG_NOQID(ERROR, "Cannot report query error for " << qid <<
758 : " SSL Context Error: " << redisSSLContextGetError(ssl_error));
759 0 : if (ssl_ctx) redisFreeSSLContext(ssl_ctx);
760 0 : redisFree(c);
761 0 : return;
762 : }
763 0 : int rc = redisInitiateSSLWithContext(c, ssl_ctx);
764 0 : if (rc != REDIS_OK) {
765 0 : QE_LOG_NOQID(ERROR, "Cannot report query error for " << qid <<
766 : " SSL Connection Error: " << c->errstr);
767 0 : redisFreeSSLContext(ssl_ctx);
768 0 : redisFree(c);
769 0 : return;
770 : }
771 0 : redisFreeSSLContext(ssl_ctx);
772 : }
773 :
774 : //Authenticate the context with password
775 0 : if (!redis_password_.empty()) {
776 0 : redisReply * reply = (redisReply *) redisCommand(c, "AUTH %s",
777 : redis_password_.c_str());
778 0 : if (reply->type == REDIS_REPLY_ERROR) {
779 0 : QE_LOG_NOQID(ERROR, "Authentication to redis error");
780 0 : freeReplyObject(reply);
781 0 : redisFree(c);
782 0 : return;
783 : }
784 0 : freeReplyObject(reply);
785 : }
786 :
787 : char stat[80];
788 0 : string key = "REPLY:" + qid;
789 0 : sprintf(stat,"{\"progress\":%d}", - ret_code);
790 :
791 0 : redisReply * reply = (redisReply *) redisCommand(c, "RPUSH %s %s",
792 : key.c_str(), stat);
793 :
794 0 : freeReplyObject(reply);
795 0 : redisFree(c);
796 0 : }
797 :
798 :
799 126 : void StartPipeline(const string qid, int redis_host_idx) {
800 126 : QueryStats qs;
801 126 : qs.set_qid(qid);
802 126 : qs.set_rows(0);
803 126 : qs.set_time(0);
804 126 : qs.set_final_merge_time(0);
805 126 : qs.set_enq_delay(0);
806 :
807 126 : uint64_t now = UTCTimestampUsec();
808 :
809 126 : std::scoped_lock lock(mutex_);
810 :
811 126 : string redis_host = redis_host_port_pairs_[redis_host_idx].first;
812 126 : int redis_port = redis_host_port_pairs_[redis_host_idx].second;
813 126 : QE_LOG_NOQID(INFO, "StartPipeline on :" << " Redis:" << redis_host << " Port:" << redis_port);
814 126 : redisContext *c = redisConnect(redis_host.c_str(), redis_port);
815 :
816 126 : if (c->err) {
817 0 : QE_LOG_NOQID(ERROR, "Cannot start Pipleline for " << qid <<
818 : " . No Redis Connection");
819 0 : redisFree(c);
820 0 : qs.set_error("No Redis Connection");
821 0 : QUERY_PERF_INFO_SEND(Sandesh::source(), // name
822 : "__UNKNOWN__", // table
823 : qs);
824 0 : return;
825 : }
826 :
827 : /* Secure the connection if SSL enabled */
828 126 : if (redis_ssl_enable_) {
829 0 : redisSSLContextError ssl_error = REDIS_SSL_CTX_NONE;
830 0 : redisSSLContext *ssl_ctx = redisCreateSSLContext(
831 : redis_ca_cert_.c_str(),
832 : NULL,
833 : redis_certfile_.c_str(),
834 : redis_keyfile_.c_str(),
835 : "sni",
836 : &ssl_error);
837 0 : if (!ssl_ctx || ssl_error != REDIS_SSL_CTX_NONE) {
838 0 : QE_LOG_NOQID(ERROR, "Cannot start pipeline for " << qid <<
839 : " SSL Context Error: " << redisSSLContextGetError(ssl_error));
840 0 : if (ssl_ctx) redisFreeSSLContext(ssl_ctx);
841 0 : redisFree(c);
842 0 : qs.set_error("Redis SSL Context Error");
843 0 : QUERY_PERF_INFO_SEND(Sandesh::source(), "__UNKNOWN__", qs);
844 0 : return;
845 : }
846 0 : int rc = redisInitiateSSLWithContext(c, ssl_ctx);
847 0 : if (rc != REDIS_OK) {
848 0 : QE_LOG_NOQID(ERROR, "Cannot start pipeline for " << qid <<
849 : " SSL Connection Error: " << c->errstr);
850 0 : redisFreeSSLContext(ssl_ctx);
851 0 : redisFree(c);
852 0 : qs.set_error("Redis SSL Connection Error");
853 0 : QUERY_PERF_INFO_SEND(Sandesh::source(), "__UNKNOWN__", qs);
854 0 : return;
855 : }
856 0 : redisFreeSSLContext(ssl_ctx);
857 : }
858 :
859 : //Authenticate the context with password
860 126 : if ( !redis_password_.empty()) {
861 0 : redisReply * reply = (redisReply *) redisCommand(c, "AUTH %s",
862 : redis_password_.c_str());
863 0 : if (reply->type == REDIS_REPLY_ERROR) {
864 0 : QE_LOG_NOQID(ERROR, "Authentication to redis error");
865 0 : freeReplyObject(reply);
866 0 : redisFree(c);
867 0 : qs.set_error("Redis Auth Failed");
868 0 : QUERY_PERF_INFO_SEND(Sandesh::source(), // name
869 : "__UNKNOWN__", // table
870 : qs);
871 0 : return;
872 : }
873 0 : freeReplyObject(reply);
874 : }
875 :
876 126 : string key = "QUERY:" + qid;
877 126 : redisReply * reply = (redisReply *) redisCommand(c, "hgetall %s",
878 : key.c_str());
879 :
880 126 : map<string,string> terms;
881 126 : if (!(c->err) && (reply->type == REDIS_REPLY_ARRAY)) {
882 1200 : for (uint32_t i=0; i<reply->elements; i+=2) {
883 1074 : string idx(reply->element[i]->str);
884 1074 : string val(reply->element[i+1]->str);
885 1074 : terms[idx] = val;
886 1074 : }
887 126 : } else {
888 0 : QE_LOG_NOQID(ERROR, "Cannot start Pipleline for " << qid <<
889 : ". Could not read query input");
890 0 : freeReplyObject(reply);
891 0 : redisFree(c);
892 0 : QueryError(redis_host_idx, qid, 5);
893 0 : qs.set_error("Could not read query input");
894 0 : QUERY_PERF_INFO_SEND(Sandesh::source(), // name
895 : "__UNKNOWN__", // table
896 : qs);
897 0 : return;
898 : }
899 :
900 126 : freeReplyObject(reply);
901 126 : redisFree(c);
902 :
903 126 : QueryEngine::QueryParams qp(qid, terms, max_tasks_,
904 252 : UTCTimestampUsec());
905 :
906 126 : vector<uint64_t> chunk_size;
907 : bool need_merge;
908 : bool map_output;
909 126 : string table;
910 126 : string where;
911 : uint32_t wterms;
912 126 : string select;
913 126 : string post;
914 : uint64_t time_period;
915 :
916 126 : int ret = qosp_->qe_->QueryPrepare(qp, chunk_size, need_merge, map_output,
917 : where, wterms, select, post, time_period, table);
918 :
919 126 : qs.set_where(where);
920 126 : qs.set_select(select);
921 126 : qs.set_post(post);
922 126 : qs.set_time_span(time_period);
923 126 : uint64_t enqtm = atol(terms["enqueue_time"].c_str());
924 126 : uint32_t enq_delay = static_cast<uint32_t>((now - enqtm)/1000);
925 126 : qs.set_enq_delay(enq_delay);
926 :
927 126 : if (ret!=0) {
928 0 : QueryError(redis_host_idx, qid, ret);
929 0 : QE_LOG_NOQID(ERROR, "Cannot start Pipleline for " << qid <<
930 : ". Query Parsing Error " << ret);
931 0 : qs.set_error("Query Parsing Error");
932 0 : QUERY_PERF_INFO_SEND(Sandesh::source(), // name
933 : table, // table
934 : qs);
935 0 : return;
936 : } else {
937 126 : QE_LOG_NOQID(INFO, "Chunks: " << chunk_size.size() <<
938 : " Need Merge: " << need_merge);
939 : }
940 :
941 126 : if (pipes_.size() >= 32) {
942 0 : QueryError(redis_host_idx, qid, EMFILE);
943 0 : QE_LOG_NOQID(ERROR, "Cannot start Pipleline for " << qid <<
944 : ". Too many queries : " << pipes_.size());
945 0 : qs.set_error("EMFILE");
946 0 : QUERY_PERF_INFO_SEND(Sandesh::source(), // name
947 : table, // table
948 : qs);
949 0 : return;
950 : }
951 :
952 126 : shared_ptr<Input> inp(new Input());
953 126 : inp.get()->hostname = hostname_;
954 126 : inp.get()->qp = qp;
955 126 : inp.get()->map_output = map_output;
956 126 : inp.get()->need_merge = need_merge;
957 126 : inp.get()->chunk_size = chunk_size;
958 126 : inp.get()->where = where;
959 126 : inp.get()->select = select;
960 126 : inp.get()->post = post;
961 126 : inp.get()->time_period = time_period;
962 126 : inp.get()->table = table;
963 126 : inp.get()->chunk_q = 0;
964 126 : inp.get()->total_rows = 0;
965 126 : inp.get()->max_rows = max_rows_;
966 126 : inp.get()->wterms = wterms;
967 126 : vector<pair<int,int> > tinfo;
968 1134 : for (uint idx=0; idx<(uint)max_tasks_; idx++) {
969 1008 : tinfo.push_back(make_pair(0, -1));
970 : }
971 :
972 : QEPipeT *wp = new QEPipeT(
973 252 : new WorkStage<Input, Stage0Merge, RawResultT, Stage0Out>(
974 : tinfo,
975 : boost::bind(&QEOpServerImpl::QueryExec, this, _1,_2,_3,_4),
976 126 : boost::bind(&QEOpServerImpl::QueryMerge, this, _1,_2,_3)),
977 252 : new WorkStage<Stage0Merge, Output, RedisT>(
978 252 : list_of(make_pair(0,-1))(make_pair(0,-1)),
979 126 : boost::bind(&QEOpServerImpl::QueryResp, this, _1,_2,_3,_4)));
980 :
981 126 : pipes_.insert(make_pair(qid, wp));
982 :
983 : // Initialize the m_analytics_queries for this qid
984 126 : m_analytics_queries[qid] = std::vector<boost::shared_ptr<AnalyticsQuery> > ();
985 :
986 126 : int conn = LeastLoadedConnection(redis_host_idx);
987 126 : QE_LOG_NOQID(DEBUG, "Getting Least Loaded Conn as :" << conn);
988 126 : npipes_[redis_host_idx][conn]++;
989 :
990 : // The cnum with index 0 is only used for receiving new queries
991 126 : inp.get()->cnum = conn+1;
992 126 : inp.get()->redis_host_idx = redis_host_idx;
993 :
994 126 : wp->Start(boost::bind(&QEOpServerImpl::QEPipeCb, this, wp, _1), inp);
995 126 : QE_LOG_NOQID(DEBUG, "Starting Pipeline for " << qid << " , " << conn+1 <<
996 : " conn, " << tinfo.size() << " tasks");
997 :
998 : // Update query status
999 126 : RedisAsyncConnection * rac = conns_[redis_host_idx][inp.get()->cnum].get();
1000 126 : string rkey = "REPLY:" + qid;
1001 : char stat[40];
1002 126 : sprintf(stat,"{\"progress\":15}");
1003 126 : RedisAsyncArgCommand(rac, NULL,
1004 252 : list_of(string("RPUSH"))(rkey)(stat));
1005 :
1006 :
1007 126 : }
1008 :
1009 230 : void ConnUpPrePostProcess(int redis_host_idx, uint8_t cnum) {
1010 : //Assign callback for AUTH command
1011 230 : cb_proc_fn_[redis_host_idx][cnum] = boost::bind(&QEOpServerImpl::ConnectCallbackProcess,
1012 230 : this, redis_host_idx, cnum, _1, _2, _3);
1013 230 : conns_[redis_host_idx][cnum].get()->SetClientAsyncCmdCb(cb_proc_fn_[redis_host_idx][cnum]);
1014 : //Send AUTH command
1015 230 : RedisAsyncConnection * rac = conns_[redis_host_idx][cnum].get();
1016 230 : if (!redis_password_.empty()) {
1017 20 : RedisAsyncArgCommand(rac, NULL,
1018 40 : list_of(string("AUTH"))(redis_password_.c_str()));
1019 : } else {
1020 210 : RedisAsyncArgCommand(rac, NULL,
1021 420 : list_of(string("PING")));
1022 : }
1023 230 : }
1024 :
1025 230 : void ConnUp(int redis_host_idx, uint8_t cnum) {
1026 230 : string redis_host = redis_host_port_pairs_[redis_host_idx].first;
1027 230 : std::ostringstream ostr;
1028 230 : ostr << "ConnUp.. UP " << (uint32_t)cnum << " With Redis:" << redis_host;
1029 230 : QE_LOG_NOQID(DEBUG, ostr.str());
1030 460 : qosp_->evm_->io_service()->post(
1031 230 : boost::bind(&QEOpServerImpl::ConnUpPrePostProcess,
1032 : this, redis_host_idx, cnum));
1033 230 : }
1034 :
1035 245 : void UpdateRedisConnectionStatus () {
1036 245 : std::ostringstream ostr;
1037 245 : map<string, string>::iterator it;
1038 245 : vector<string> redis_endpoints;
1039 245 : int redis_hosts_cnt = redis_host_port_pairs_.size();
1040 245 : int redis_down_count = 0;
1041 245 : int redis_idx = 0;
1042 245 : int conn_idx = 0;
1043 245 : vector<Endpoint> redis_endpoint_list;
1044 490 : for (redis_idx = 0; redis_idx < redis_hosts_cnt; redis_idx++) {
1045 245 : string redis_host(redis_host_port_pairs_[redis_idx].first);
1046 245 : redis_endpoint_list.push_back(conns_[redis_idx][0]->Endpoint());
1047 541 : for (conn_idx = 0; conn_idx < kConnections + 1; conn_idx++) {
1048 486 : if (true == connState_[redis_idx][conn_idx]) {
1049 190 : break;
1050 : }
1051 : }
1052 245 : if (conn_idx == kConnections + 1) {
1053 55 : ostr << conns_[redis_idx][0]->Endpoint() << "::Down.";
1054 55 : redis_down_count++;
1055 : }
1056 245 : }
1057 245 : if (redis_down_count < redis_hosts_cnt) {
1058 : /* Then QE should be UP, but display if any redis connection is down */
1059 380 : ConnectionState::GetInstance()->Update(ConnectionType::REDIS_QUERY,
1060 380 : "Query", ConnectionStatus::UP, redis_endpoint_list, ostr.str());
1061 : } else {
1062 55 : std::ostringstream empty_ostr;
1063 110 : ConnectionState::GetInstance()->Update(ConnectionType::REDIS_QUERY,
1064 110 : "Query", ConnectionStatus::DOWN, redis_endpoint_list, empty_ostr.str());
1065 55 : }
1066 245 : }
1067 :
1068 15 : void ConnDown(int redis_host_idx, uint8_t cnum) {
1069 15 : string redis_host = redis_host_port_pairs_[redis_host_idx].first;
1070 15 : std::ostringstream ostr;
1071 15 : ostr << "ConnDown.. DOWN.. Reconnect.." << (uint32_t)cnum << " With Redis:" << redis_host;
1072 15 : QE_LOG_NOQID(DEBUG, ostr.str());
1073 15 : connState_[redis_host_idx][cnum] = false;
1074 30 : qosp_->evm_->io_service()->post(
1075 15 : boost::bind(&QEOpServerImpl::UpdateRedisConnectionStatus, this));
1076 30 : qosp_->evm_->io_service()->post(boost::bind(&RedisAsyncConnection::RAC_Connect,
1077 15 : conns_[redis_host_idx][cnum].get()));
1078 15 : }
1079 :
1080 230 : void ConnectCallbackProcess(int redis_host_idx, uint8_t cnum, const redisAsyncContext *c, void *r, void *privdata) {
1081 230 : QE_LOG_NOQID(DEBUG, "UP ConnectCallbackProcess.." << conns_[redis_host_idx][cnum]->Endpoint());
1082 230 : if (r == NULL) {
1083 0 : QE_LOG_NOQID(DEBUG, "In ConnectCallbackProcess.. NULL Reply");
1084 0 : return;
1085 : }
1086 230 : redisReply reply = *reinterpret_cast<redisReply*>(r);
1087 230 : if (reply.type != REDIS_REPLY_ERROR) {
1088 230 : QE_LOG_NOQID(DEBUG, "In ConnectCallbackProcess.." << conns_[redis_host_idx][cnum]->Endpoint());
1089 460 : qosp_->evm_->io_service()->post(
1090 230 : boost::bind(&QEOpServerImpl::UpdateRedisConnectionStatus, this));
1091 460 : qosp_->evm_->io_service()->post(
1092 460 : boost::bind(&QEOpServerImpl::ConnUpPostProcess,
1093 : this, redis_host_idx, cnum));
1094 : } else {
1095 0 : QE_LOG_NOQID(ERROR,"In connectCallbackProcess.. Error");
1096 0 : QE_ASSERT(reply.type != REDIS_REPLY_ERROR);
1097 : }
1098 : }
1099 :
1100 3337 : void CallbackProcess(int redis_host_idx, uint8_t cnum, const redisAsyncContext *c, void *r, void *privdata) {
1101 :
1102 : //QE_TRACE_NOQID(DEBUG, "Redis CB" << cnum);
1103 3337 : if (0 == cnum) {
1104 129 : if (r == NULL) {
1105 3 : QE_LOG_NOQID(DEBUG, __func__ << ": received NULL reply from redis");
1106 3 : return;
1107 : }
1108 :
1109 126 : redisReply reply = *reinterpret_cast<redisReply*>(r);
1110 126 : if (reply.type != REDIS_REPLY_STRING) {
1111 0 : QE_LOG_NOQID(ERROR, __func__ << " Bad Redis reply on control connection: " << reply.type);
1112 0 : if (reply.type == REDIS_REPLY_ERROR) {
1113 0 : string errstr(reply.str);
1114 0 : QE_LOG_NOQID(ERROR, __func__ << " Redis Error: " << reply.str);
1115 0 : sleep(1000);
1116 0 : }
1117 : }
1118 126 : QE_ASSERT(reply.type == REDIS_REPLY_STRING);
1119 126 : string qid(reply.str);
1120 :
1121 126 : StartPipeline(qid, redis_host_idx);
1122 :
1123 252 : qosp_->evm_->io_service()->post(
1124 126 : boost::bind(&QEOpServerImpl::ConnUpPostProcess,
1125 : this, redis_host_idx, cnum));
1126 126 : return;
1127 126 : }
1128 :
1129 3208 : unique_ptr<RedisT> fullReply;
1130 3208 : vector<string> elements;
1131 :
1132 3208 : if (r == NULL) {
1133 : //QE_TRACE_NOQID(DEBUG, "NULL Reply...\n");
1134 : } else {
1135 3208 : redisReply reply = *reinterpret_cast<redisReply*>(r);
1136 3208 : fullReply.reset(new RedisT);
1137 3208 : fullReply.get()->first = reply;
1138 3208 : if (reply.type == REDIS_REPLY_ARRAY) {
1139 0 : for (uint32_t i=0; i<reply.elements; i++) {
1140 0 : string element(reply.element[i]->str);
1141 0 : fullReply.get()->second.push_back(element);
1142 0 : }
1143 3208 : } else if (reply.type == REDIS_REPLY_STRING) {
1144 0 : fullReply.get()->second.push_back(string(reply.str));
1145 : }
1146 : }
1147 3208 : if (!privdata) {
1148 : //QE_TRACE_NOQID(DEBUG, "Ignoring redis reply");
1149 2956 : return;
1150 : }
1151 252 : ExternalProcIf<RedisT> * rpi =
1152 : reinterpret_cast<ExternalProcIf<RedisT> *>(privdata);
1153 252 : QE_TRACE_NOQID(DEBUG, " Rx data from REDIS for " << rpi->Key());
1154 :
1155 252 : rpi->Response(std::move(fullReply));
1156 :
1157 6164 : }
1158 :
1159 44 : void BuildRedisIPPort(vector<string> redis_ip_port_list) {
1160 : typedef boost::tokenizer<boost::char_separator<char> > tokenizer;
1161 44 : for (vector<string>::const_iterator it = redis_ip_port_list.begin();
1162 88 : it != redis_ip_port_list.end(); it++) {
1163 44 : string redis_ip_port(*it);
1164 44 : boost::char_separator<char> sep(":");
1165 44 : tokenizer tokens(redis_ip_port, sep);
1166 44 : tokenizer::iterator tit = tokens.begin();
1167 44 : string redis_ip(*tit);
1168 44 : ++tit;
1169 44 : string redis_port(*tit);
1170 44 : redis_host_port_pairs_.push_back(make_pair(redis_ip, atoi(redis_port.c_str())));
1171 44 : }
1172 44 : }
1173 :
1174 44 : QEOpServerImpl(vector<string> redis_ip_ports,
1175 : const string & redis_password,
1176 : const bool redis_ssl_enable,
1177 : const string & redis_keyfile,
1178 : const string & redis_certfile,
1179 : const string & redis_ca_cert,
1180 : QEOpServerProxy * qosp, int max_tasks,
1181 44 : int max_rows, const string &host_ip) :
1182 44 : hostname_(ResolveCanonicalName(host_ip)),
1183 44 : redis_password_(redis_password),
1184 44 : redis_ssl_enable_(redis_ssl_enable),
1185 44 : redis_keyfile_(redis_keyfile),
1186 44 : redis_certfile_(redis_certfile),
1187 44 : redis_ca_cert_(redis_ca_cert),
1188 44 : qosp_(qosp),
1189 44 : max_tasks_(max_tasks),
1190 88 : max_rows_(max_rows) {
1191 44 : int redis_host_idx = 0;
1192 : /* Initialize */
1193 44 : BuildRedisIPPort(redis_ip_ports);
1194 44 : int redis_host_count = redis_host_port_pairs_.size();
1195 44 : cb_proc_fn_ = new RedisAsyncConnection::ClientAsyncCmdCbFn* [redis_host_count];
1196 44 : conns_ = new boost::shared_ptr<RedisAsyncConnection>* [redis_host_count];
1197 44 : npipes_ = new int* [redis_host_count];
1198 44 : connState_ = new bool* [redis_host_count];
1199 88 : for (int i = 0; i < redis_host_count; i++) {
1200 264 : cb_proc_fn_[i] = new RedisAsyncConnection::ClientAsyncCmdCbFn[kConnections + 1];
1201 264 : conns_[i] = new boost::shared_ptr<RedisAsyncConnection>[kConnections + 1];
1202 44 : npipes_[i] = new int[kConnections];
1203 44 : connState_[i] = new bool[kConnections + 1];
1204 : }
1205 88 : for (redis_host_idx = 0; redis_host_idx < int(redis_host_count); redis_host_idx++) {
1206 44 : string redis_host = redis_host_port_pairs_[redis_host_idx].first;
1207 44 : int redis_port = redis_host_port_pairs_[redis_host_idx].second;
1208 264 : for (int i = 0; i < kConnections + 1; i++) {
1209 220 : cb_proc_fn_[redis_host_idx][i] = boost::bind(&QEOpServerImpl::CallbackProcess,
1210 220 : this, redis_host_idx, i, _1, _2, _3);
1211 220 : connState_[redis_host_idx][i] = false;
1212 220 : if (i) {
1213 176 : conns_[redis_host_idx][i].reset(rac_alloc(qosp->evm_, redis_host, redis_port,
1214 : boost::bind(&QEOpServerImpl::ConnUp, this, redis_host_idx, i),
1215 : boost::bind(&QEOpServerImpl::ConnDown, this, redis_host_idx, i),
1216 : redis_ssl_enable, redis_keyfile, redis_certfile, redis_ca_cert));
1217 : } else {
1218 44 : conns_[redis_host_idx][i].reset(rac_alloc_nocheck(qosp->evm_, redis_host, redis_port,
1219 : boost::bind(&QEOpServerImpl::ConnUp, this, redis_host_idx, i),
1220 : boost::bind(&QEOpServerImpl::ConnDown, this, redis_host_idx, i),
1221 : redis_ssl_enable, redis_keyfile, redis_certfile, redis_ca_cert));
1222 : }
1223 : // The cnum with index 0 is only used for receiving new queries
1224 : // It does not host any pipelines
1225 220 : if (i) npipes_[redis_host_idx][i-1] = 0;
1226 : }
1227 44 : }
1228 44 : }
1229 :
1230 264 : ~QEOpServerImpl() {
1231 44 : }
1232 :
1233 904 : void AddAnalyticsQuery(const std::string &qid,
1234 : boost::shared_ptr<AnalyticsQuery> q) {
1235 904 : std::scoped_lock lock(mutex_);
1236 904 : m_analytics_queries[qid].push_back(q);
1237 904 : }
1238 :
1239 : private:
1240 :
1241 : static const int kMaxRowThreshold = 10000;
1242 :
1243 : // We always have one connection to receive new queries from OpServer
1244 : // This is the number of addition connections, which will be
1245 : // used to read query parameters and write query results
1246 : static const uint8_t kConnections = 4;
1247 :
1248 : const string hostname_;
1249 : map<string, string> redis_status_maps;
1250 : const string redis_password_;
1251 : const bool redis_ssl_enable_;
1252 : const string redis_keyfile_;
1253 : const string redis_certfile_;
1254 : const string redis_ca_cert_;
1255 : QEOpServerProxy * const qosp_;
1256 : boost::shared_ptr<RedisAsyncConnection> **conns_;
1257 : RedisAsyncConnection::ClientAsyncCmdCbFn **cb_proc_fn_;
1258 : bool **connState_;
1259 :
1260 : std::mutex mutex_;
1261 : map<string,QEPipeT*> pipes_;
1262 : vector<pair<string, int> > redis_host_port_pairs_;
1263 : int **npipes_;
1264 : int max_tasks_;
1265 : int max_rows_;
1266 : std::map<std::string, std::vector<boost::shared_ptr<AnalyticsQuery> > >
1267 : m_analytics_queries;
1268 :
1269 : };
1270 :
1271 44 : QEOpServerProxy::QEOpServerProxy(EventManager *evm, QueryEngine *qe,
1272 : vector<string> redis_ip_ports,
1273 : const string & redis_password,
1274 : const bool redis_ssl_enable,
1275 : const string & redis_keyfile,
1276 : const string & redis_certfile,
1277 : const string & redis_ca_cert,
1278 44 : const std::string &host_ip, int max_tasks, int max_rows) :
1279 44 : evm_(evm),
1280 44 : qe_(qe),
1281 44 : impl_(new QEOpServerImpl(redis_ip_ports, redis_password, redis_ssl_enable,
1282 : redis_keyfile, redis_certfile, redis_ca_cert,
1283 88 : this, max_tasks, max_rows, host_ip)) {}
1284 :
1285 88 : QEOpServerProxy::~QEOpServerProxy() {}
1286 :
1287 : void
1288 856 : QEOpServerProxy::QueryResult(void * qid, QPerfInfo qperf,
1289 : std::auto_ptr<BufferT> res, std::auto_ptr<OutRowMultimapT> mres) {
1290 :
1291 856 : impl_->QECallback(qid, qperf, std::unique_ptr<BufferT>(res.release()), std::unique_ptr<OutRowMultimapT>(mres.release()));
1292 856 : }
1293 :
1294 : void
1295 904 : QEOpServerProxy::QueryResult(void * qid, QPerfInfo qperf,
1296 : std::auto_ptr<std::vector<query_result_unit_t> > res) {
1297 :
1298 904 : impl_->QECallback(qid, qperf, std::unique_ptr<std::vector<query_result_unit_t>>(res.release()));
1299 904 : }
1300 :
1301 903 : void QEOpServerProxy::AddAnalyticsQuery(const std::string &qid,
1302 : boost::shared_ptr<AnalyticsQuery> q) {
1303 903 : impl_->AddAnalyticsQuery(qid, q);
1304 904 : }
|
