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