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