Line data Source code
1 : /*
2 : * Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
3 : */
4 :
5 : #include <sys/types.h>
6 : #include <sys/socket.h>
7 : #include <netdb.h>
8 : #include <fstream>
9 : #include <boost/asio/ip/host_name.hpp>
10 : #include <boost/foreach.hpp>
11 : #include <boost/program_options.hpp>
12 : #include <boost/tokenizer.hpp>
13 : #include "base/logging.h"
14 : #include "base/task.h"
15 : #include "base/task_trigger.h"
16 : #include "base/timer.h"
17 : #include "base/connection_info.h"
18 : #include "io/event_manager.h"
19 : #include "QEOpServerProxy.h"
20 : #include <boost/bind/bind.hpp>
21 : #include <boost/assign/list_of.hpp>
22 : #include <sandesh/common/vns_types.h>
23 : #include <sandesh/common/vns_constants.h>
24 : #include <analytics/analytics_types.h>
25 : #include "nodeinfo_types.h"
26 : #include "options.h"
27 : #include "query.h"
28 : #include "stats_query.h"
29 : #include "qe_sandesh.h"
30 : #include <base/misc_utils.h>
31 : #include <query_engine/buildinfo.h>
32 : #include <sandesh/sandesh_http.h>
33 : #include <io/process_signal.h>
34 : #include <malloc.h>
35 :
36 : using std::unique_ptr;
37 : using std::string;
38 : using std::vector;
39 : using std::map;
40 : using std::vector;
41 : using boost::assign::list_of;
42 : using boost::system::error_code;
43 : using namespace boost::asio;
44 : using namespace std;
45 : using process::ConnectionStateManager;
46 : using process::GetProcessStateCb;
47 : using process::ConnectionType;
48 : using process::ConnectionTypeName;
49 : using process::g_process_info_constants;
50 : using process::Signal;
51 : using namespace boost::placeholders;
52 :
53 : // This is to force qed to wait for a gdbattach
54 : // before proceeding.
55 : // It will make it easier to debug qed during systest
56 : volatile int gdbhelper = 1;
57 : int timer_count = 0;
58 :
59 : static EventManager * pevm = NULL;
60 : static Options options;
61 : static TaskTrigger *qe_dbstats_task_trigger;
62 : static Timer *qe_dbstats_timer;
63 :
64 88 : static void WaitForIdle() {
65 : static const int kTimeout = 15;
66 88 : TaskScheduler *scheduler = TaskScheduler::GetInstance();
67 :
68 88 : for (int i = 0; i < (kTimeout * 1000); i++) {
69 88 : if (scheduler->IsEmpty()) {
70 88 : break;
71 : }
72 0 : usleep(1000);
73 : }
74 88 : }
75 :
76 44 : static bool OptionsParse(Options &options, EventManager &evm,
77 : int argc, char *argv[]) {
78 : try {
79 44 : options.Parse(evm, argc, argv);
80 44 : return true;
81 0 : } catch (boost::program_options::error &e) {
82 0 : cout << "Error " << e.what() << endl;
83 0 : } catch (...) {
84 0 : cout << "Options Parser: Caught fatal unknown exception" << endl;
85 0 : }
86 :
87 0 : return false;
88 : }
89 :
90 44 : static void ShutdownQe() {
91 44 : WaitForIdle();
92 44 : if (qe_dbstats_timer) {
93 44 : TimerManager::DeleteTimer(qe_dbstats_timer);
94 44 : delete qe_dbstats_task_trigger;
95 44 : qe_dbstats_timer = NULL;
96 : }
97 44 : Sandesh::Uninit();
98 : ConnectionStateManager::
99 44 : GetInstance()->Shutdown();
100 44 : WaitForIdle();
101 44 : }
102 :
103 44 : static void TerminateQe(const boost::system::error_code &error, int sig) {
104 44 : if (!error) {
105 : static bool done;
106 44 : if (done) {
107 0 : return;
108 : }
109 44 : ShutdownQe();
110 44 : pevm->Shutdown();
111 44 : done = true;
112 : } else {
113 0 : LOG(ERROR, "SIGTERM handler ERROR: " << error);
114 : }
115 : }
116 :
117 0 : static void ReConfigQe(const boost::system::error_code &error, int sig) {
118 0 : if (!error) {
119 0 : options.ParseReConfig();
120 : } else {
121 0 : LOG(ERROR, "SIGHUP handler ERROR: " << error);
122 : }
123 0 : }
124 :
125 39 : static bool QEDbStatsTrigger() {
126 39 : qe_dbstats_task_trigger->Set();
127 39 : return false;
128 : }
129 :
130 : /*
131 : * Send the database stats to the collector periodically
132 : */
133 39 : static bool SendQEDbStats(QESandeshContext &ctx) {
134 : // DB stats
135 39 : std::vector<GenDb::DbTableInfo> vdbti, vstats_dbti;
136 39 : GenDb::DbErrors dbe;
137 39 : QueryEngine *qe = ctx.QE();
138 39 : if ((qe->GetDbHandler()).get() != NULL) {
139 23 : qe->GetDiffStats(&vdbti, &dbe, &vstats_dbti);
140 : }
141 39 : map<string,GenDb::DbTableStat> mtstat, msstat;
142 :
143 177 : for (size_t idx=0; idx<vdbti.size(); idx++) {
144 138 : GenDb::DbTableStat dtis;
145 138 : dtis.set_reads(vdbti[idx].get_reads());
146 138 : dtis.set_read_fails(vdbti[idx].get_read_fails());
147 138 : dtis.set_writes(vdbti[idx].get_writes());
148 138 : dtis.set_write_fails(vdbti[idx].get_write_fails());
149 138 : dtis.set_write_back_pressure_fails(vdbti[idx].get_write_back_pressure_fails());
150 138 : mtstat.insert(make_pair(vdbti[idx].get_table_name(), dtis));
151 138 : }
152 :
153 40 : for (size_t idx=0; idx<vstats_dbti.size(); idx++) {
154 1 : GenDb::DbTableStat dtis;
155 1 : dtis.set_reads(vstats_dbti[idx].get_reads());
156 1 : dtis.set_read_fails(vstats_dbti[idx].get_read_fails());
157 1 : dtis.set_writes(vstats_dbti[idx].get_writes());
158 1 : dtis.set_write_fails(vstats_dbti[idx].get_write_fails());
159 1 : dtis.set_write_back_pressure_fails(
160 1 : vstats_dbti[idx].get_write_back_pressure_fails());
161 1 : msstat.insert(make_pair(vstats_dbti[idx].get_table_name(), dtis));
162 1 : }
163 :
164 78 : QEDbStats qe_db_stats;
165 39 : qe_db_stats.set_table_info(mtstat);
166 39 : qe_db_stats.set_errors(dbe);
167 39 : qe_db_stats.set_stats_info(msstat);
168 :
169 39 : cass::cql::DbStats cql_stats;
170 39 : if (qe->GetCqlStats(&cql_stats)) {
171 23 : qe_db_stats.set_cql_stats(cql_stats);
172 : }
173 39 : qe_db_stats.set_name(Sandesh::source());
174 39 : QEDbStatsUve::Send(qe_db_stats);
175 39 : if (timer_count >5) {
176 0 : timer_count = 0;
177 0 : malloc_trim(0);
178 : }
179 39 : timer_count ++;
180 39 : qe_dbstats_timer->Cancel();
181 39 : qe_dbstats_timer->Start(60*1000, boost::bind(&QEDbStatsTrigger),
182 : NULL);
183 39 : return true;
184 39 : }
185 :
186 : int
187 44 : main(int argc, char *argv[]) {
188 44 : EventManager evm;
189 44 : pevm = &evm;
190 :
191 44 : srand(unsigned(time(NULL)));
192 :
193 : // Increase max number of threads available by a factor of 4
194 44 : TaskScheduler::SetThreadAmpFactor(
195 : QEOpServerProxy::nThreadCountMultFactor);
196 :
197 44 : if (!OptionsParse(options, evm, argc, argv)) {
198 0 : exit(-1);
199 : }
200 :
201 44 : while (gdbhelper==0) {
202 0 : usleep(1000);
203 : }
204 :
205 44 : Module::type module = Module::QUERY_ENGINE;
206 44 : string module_name = g_vns_constants.ModuleNames.find(module)->second;
207 :
208 44 : std::string log_property_file = options.log_property_file();
209 44 : if (log_property_file.size()) {
210 0 : LoggingInit(log_property_file);
211 : }
212 : else {
213 88 : LoggingInit(options.log_file(), options.log_file_size(),
214 44 : options.log_files_count(), options.use_syslog(),
215 88 : options.syslog_facility(), module_name,
216 : SandeshLevelTolog4Level(
217 88 : Sandesh::StringToLevel(options.log_level())));
218 : }
219 :
220 44 : int max_tasks = options.max_tasks();
221 : // Tune max_tasks
222 44 : if (max_tasks == 0)
223 : {
224 : // no command line option was specified to tune the max # of tasks
225 44 : max_tasks = QEOpServerProxy::nMaxChunks;
226 :
227 44 : if (max_tasks*2 > TaskScheduler::GetThreadCount())
228 : {
229 : // avoid creating too many tasks for one query
230 44 : max_tasks = TaskScheduler::GetThreadCount()/2;
231 : // make sure atleast we have one task
232 44 : max_tasks = (max_tasks > 1) ? max_tasks : 1;
233 : }
234 : }
235 :
236 44 : LOG(INFO, "http-server-port " << options.http_server_port());
237 44 : LOG(INFO, "Max-tasks " << max_tasks);
238 44 : LOG(INFO, "Max-slice " << options.max_slice());
239 146 : BOOST_FOREACH(std::string collector_ip, options.collector_server_list()) {
240 51 : LOG(INFO, "Collectors " << collector_ip);
241 95 : }
242 :
243 : // Initialize Sandesh
244 : NodeType::type node_type =
245 44 : g_vns_constants.Module2NodeType.find(module)->second;
246 44 : std::string instance_id(g_vns_constants.INSTANCE_ID_DEFAULT);
247 : // Determine if the number of connections is expected:
248 : // 1. Collector client
249 : // 2. Redis
250 : // 3. Cassandra
251 : std::vector<ConnectionTypeName> expected_connections =
252 : boost::assign::list_of
253 88 : (ConnectionTypeName(g_process_info_constants.ConnectionTypeNames.find(
254 44 : ConnectionType::DATABASE)->second, ""))
255 44 : (ConnectionTypeName(g_process_info_constants.ConnectionTypeNames.find(
256 88 : ConnectionType::REDIS_QUERY)->second, "Query"))
257 44 : (ConnectionTypeName(g_process_info_constants.ConnectionTypeNames.find(
258 132 : ConnectionType::COLLECTOR)->second, ""));
259 44 : bool use_collector_list = true;
260 44 : if (!options.collectors_configured()) {
261 0 : use_collector_list = false;
262 : }
263 : ConnectionStateManager::
264 176 : GetInstance()->Init(*evm.io_service(),
265 88 : options.hostname(), module_name,
266 : instance_id,
267 88 : boost::bind(&GetProcessStateCb, _1, _2, _3,
268 : expected_connections), "ObjectDatabaseInfo");
269 : bool success;
270 : // subscribe to the collector service with discovery only if the
271 : // collector list is not configured.
272 44 : if (use_collector_list) {
273 : std::vector<std::string> collectors(
274 44 : options.randomized_collector_server_list());
275 44 : if (!collectors.size()) {
276 0 : collectors = options.default_collector_server_list();
277 : }
278 88 : success = Sandesh::InitGenerator(module_name, options.hostname(),
279 44 : g_vns_constants.NodeTypeNames.find(node_type)->second,
280 44 : instance_id, &evm, options.http_server_port(),
281 88 : collectors, NULL, Sandesh::DerivedStats(),
282 : options.sandesh_config());
283 44 : } else {
284 0 : const std::vector<std::string> collectors;
285 0 : success = Sandesh::InitGenerator(module_name, options.hostname(),
286 0 : g_vns_constants.NodeTypeNames.find(node_type)->second,
287 0 : instance_id, &evm, options.http_server_port(),
288 0 : collectors, NULL, Sandesh::DerivedStats(),
289 : options.sandesh_config());
290 0 : }
291 44 : if (!success) {
292 0 : LOG(ERROR, "SANDESH: Initialization FAILED ... exiting");
293 0 : ShutdownQe();
294 0 : exit(1);
295 : }
296 :
297 44 : Sandesh::SetLoggingParams(options.log_local(), options.log_category(),
298 88 : options.log_level());
299 :
300 : // XXX Disable logging -- for test purposes only
301 44 : if (options.log_disable()) {
302 0 : SetLoggingDisabled(true);
303 : }
304 :
305 44 : vector<string> cassandra_servers(options.cassandra_server_list());
306 44 : vector<string> cassandra_ips;
307 44 : vector<int> cassandra_ports;
308 44 : for (vector<string>::const_iterator it = cassandra_servers.begin();
309 88 : it != cassandra_servers.end(); it++) {
310 44 : string cassandra_server(*it);
311 : typedef boost::tokenizer<boost::char_separator<char> > tokenizer;
312 44 : boost::char_separator<char> sep(":");
313 44 : tokenizer tokens(cassandra_server, sep);
314 44 : tokenizer::iterator tit = tokens.begin();
315 44 : string cassandra_ip(*tit);
316 44 : cassandra_ips.push_back(cassandra_ip);
317 44 : ++tit;
318 44 : string port(*tit);
319 : int cassandra_port;
320 44 : stringToInteger(port, cassandra_port);
321 44 : cassandra_ports.push_back(cassandra_port);
322 44 : }
323 44 : ostringstream css;
324 44 : copy(cassandra_servers.begin(), cassandra_servers.end(),
325 44 : ostream_iterator<string>(css, " "));
326 44 : LOG(INFO, "Cassandra Servers: " << css.str());
327 :
328 44 : boost::scoped_ptr<QueryEngine> qe;
329 :
330 44 : if (cassandra_ports.size() == 1 && cassandra_ports[0] == 0) {
331 19 : qe.reset(new QueryEngine(&evm,
332 38 : options.redis_server_list(),
333 38 : options.redis_password(),
334 19 : options.redis_ssl_enable(),
335 38 : options.redis_keyfile(),
336 38 : options.redis_certfile(),
337 38 : options.redis_ca_cert(),
338 : max_tasks,
339 19 : options.max_slice(),
340 38 : options.cassandra_user(),
341 38 : options.cassandra_password(),
342 19 : options.cassandra_use_ssl(),
343 38 : options.cassandra_ca_certs(),
344 38 : options.host_ip()));
345 : } else {
346 75 : qe.reset(new QueryEngine(&evm,
347 : cassandra_ips,
348 : cassandra_ports,
349 50 : options.redis_server_list(),
350 50 : options.redis_password(),
351 25 : options.redis_ssl_enable(),
352 50 : options.redis_keyfile(),
353 50 : options.redis_certfile(),
354 50 : options.redis_ca_cert(),
355 : max_tasks,
356 25 : options.max_slice(),
357 50 : options.cassandra_user(),
358 50 : options.cassandra_password(),
359 25 : options.cassandra_use_ssl(),
360 50 : options.cassandra_ca_certs(),
361 50 : options.cluster_id(),
362 50 : options.host_ip()));
363 : }
364 44 : QESandeshContext qec(qe.get());
365 44 : Sandesh::set_client_context(&qec);
366 44 : qe_dbstats_task_trigger =
367 88 : new TaskTrigger(boost::bind(&SendQEDbStats, qec),
368 44 : TaskScheduler::GetInstance()->GetTaskId("QE::DBStats"), 0);
369 44 : qe_dbstats_timer = TimerManager::CreateTimer(*evm.io_service(),
370 : "QE Db stats timer",
371 : TaskScheduler::GetInstance()->GetTaskId("QE::DBStats"), 0);
372 44 : qe_dbstats_timer->Start(5*1000, boost::bind(&QEDbStatsTrigger), NULL);
373 :
374 : vector<Signal::SignalHandler> sigterm_handlers = boost::assign::list_of
375 44 : (boost::bind(&TerminateQe, _1, _2));
376 : vector<Signal::SignalHandler> sighup_handlers = boost::assign::list_of
377 44 : (boost::bind(&ReConfigQe, _1, _2));
378 : Signal::SignalCallbackMap smap = boost::assign::map_list_of
379 44 : (SIGTERM, sigterm_handlers)
380 44 : (SIGHUP, sighup_handlers);
381 : // Register the signal handlers
382 44 : Signal signal(&evm, smap, vector<Signal::SignalChildHandler>(), false);
383 44 : evm.Run();
384 : // Deregister the signal handlers since we are exiting
385 44 : signal.Terminate();
386 44 : return 0;
387 44 : }
388 :
|
