Line data Source code
1 : /*
2 : * Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
3 : * Copyright Contributors to the OpenSDN project
4 : */
5 :
6 : //
7 : // sandesh.cc
8 : //
9 : // Sandesh Implementation
10 : //
11 :
12 : #include <boost/filesystem.hpp>
13 : #include <boost/bind/bind.hpp>
14 : #include <boost/foreach.hpp>
15 : #include <boost/format.hpp>
16 : #include <boost/algorithm/string/predicate.hpp>
17 :
18 : #include <base/logging.h>
19 : #include <base/parse_object.h>
20 : #include <base/queue_task.h>
21 : #include <base/address_util.h>
22 : #include <http/http_session.h>
23 : #include <io/tcp_session.h>
24 :
25 : #include <sandesh/transport/TBufferTransports.h>
26 : #include <sandesh/transport/TSimpleFileTransport.h>
27 : #include <sandesh/protocol/TBinaryProtocol.h>
28 : #include <sandesh/protocol/TProtocol.h>
29 :
30 : #include <sandesh/sandesh_types.h>
31 : #include <sandesh/sandesh.h>
32 : #include <sandesh/sandesh_trace.h>
33 : #include <sandesh/sandesh_uve_types.h>
34 : #include "sandesh_statistics.h"
35 : #include "sandesh_uve.h"
36 : #include "sandesh_session.h"
37 : #include "sandesh_http.h"
38 : #include "sandesh_client.h"
39 : #include "sandesh_connection.h"
40 : #include "sandesh_state_machine.h"
41 :
42 : #include <log4cplus/helpers/pointer.h>
43 : #include <log4cplus/configurator.h>
44 : #include <log4cplus/fileappender.h>
45 : #include <log4cplus/syslogappender.h>
46 :
47 : using boost::asio::ip::tcp;
48 : using boost::asio::ip::address;
49 :
50 : using namespace contrail::sandesh::protocol;
51 : using namespace contrail::sandesh::transport;
52 : using namespace log4cplus;
53 : using namespace boost::placeholders;
54 :
55 : // Statics
56 : Sandesh::SandeshRole::type Sandesh::role_ = SandeshRole::Invalid;
57 : bool Sandesh::enable_local_log_ = false;
58 : bool Sandesh::enable_flow_log_ = false;
59 : bool Sandesh::enable_session_syslog_ = false;
60 : int Sandesh::http_port_ = 0;
61 : bool Sandesh::enable_trace_print_ = false;
62 : bool Sandesh::send_queue_enabled_ = true;
63 : bool Sandesh::connect_to_collector_ = false;
64 : bool Sandesh::disable_flow_collection_ = false;
65 : bool Sandesh::disable_sending_all_ = false;
66 : bool Sandesh::disable_sending_object_logs_ = false;
67 : bool Sandesh::disable_sending_flows_ = false;
68 : bool Sandesh::slo_to_collector_ = false;
69 : bool Sandesh::sampled_to_collector_ = false;
70 : bool Sandesh::slo_to_logger_ = false;
71 : bool Sandesh::sampled_to_logger_ = false;
72 : SandeshClient *Sandesh::client_ = NULL;
73 : SandeshConfig Sandesh::config_;
74 : std::unique_ptr<Sandesh::SandeshRxQueue> Sandesh::recv_queue_;
75 : std::string Sandesh::module_;
76 : std::string Sandesh::source_;
77 : std::string Sandesh::node_type_;
78 : std::string Sandesh::instance_id_;
79 : int Sandesh::recv_task_id_ = -1;
80 : SandeshContext* Sandesh::client_context_ = NULL;
81 : Sandesh::SandeshCallback Sandesh::response_callback_ = 0;
82 : SandeshLevel::type Sandesh::logging_level_ = SandeshLevel::INVALID;
83 : SandeshLevel::type Sandesh::logging_ut_level_ =
84 : getenv("SANDSH_UT_DEBUG") ? SandeshLevel::SYS_DEBUG : SandeshLevel::UT_DEBUG;
85 : std::string Sandesh::logging_category_;
86 : EventManager* Sandesh::event_manager_ = NULL;
87 : SandeshMessageStatistics Sandesh::msg_stats_;
88 : std::mutex Sandesh::stats_mutex_;
89 : log4cplus::Logger Sandesh::logger_ =
90 : log4cplus::Logger::getInstance(LOG4CPLUS_TEXT("SANDESH"));
91 : log4cplus::Logger Sandesh::slo_logger_ =
92 : log4cplus::Logger::getInstance(LOG4CPLUS_TEXT("SLO_SESSION"));
93 : log4cplus::Logger Sandesh::sampled_logger_ =
94 : log4cplus::Logger::getInstance(LOG4CPLUS_TEXT("SAMPLED_SESSION"));
95 :
96 : Sandesh::ModuleContextMap Sandesh::module_context_;
97 : std::atomic<uint32_t> Sandesh::sandesh_send_ratelimit_;
98 :
99 : const char *loggingPattern = "%D{%Y-%m-%d %a %H:%M:%S:%Q %Z} "
100 : " %h [Thread %t, Pid %i]: %m%n";
101 :
102 85 : const char * Sandesh::SandeshRoleToString(SandeshRole::type role) {
103 85 : switch (role) {
104 83 : case SandeshRole::Generator:
105 83 : return "Generator";
106 0 : case SandeshRole::Collector:
107 0 : return "Collector";
108 2 : case SandeshRole::Test:
109 2 : return "Test";
110 0 : case SandeshRole::Invalid:
111 0 : return "Invalid";
112 0 : default:
113 0 : return "Unknown";
114 : }
115 : }
116 :
117 86 : void Sandesh::InitReceive(int recv_task_inst) {
118 86 : assert(recv_task_id_ == -1);
119 86 : TaskScheduler *scheduler = TaskScheduler::GetInstance();
120 86 : recv_task_id_ = scheduler->GetTaskId("sandesh::RecvQueue");
121 172 : recv_queue_.reset(new SandeshRxQueue(recv_task_id_, recv_task_inst,
122 86 : &Sandesh::ProcessRecv));
123 86 : }
124 :
125 37 : void Sandesh::InitClient(EventManager *evm, Endpoint server,
126 : const SandeshConfig &config, bool periodicuve) {
127 37 : connect_to_collector_ = true;
128 37 : SANDESH_LOG(INFO, "SANDESH: CONNECT TO COLLECTOR: " <<
129 : connect_to_collector_);
130 : // Create and initialize the client
131 37 : assert(client_ == NULL);
132 37 : std::vector<Endpoint> collector_endpoints = boost::assign::list_of(server);
133 37 : client_ = new SandeshClient(evm, collector_endpoints, config,
134 37 : periodicuve);
135 37 : client_->Initiate();
136 37 : }
137 :
138 10 : static int32_t SandeshHttpCallback(SandeshRequest *rsnh) {
139 10 : return rsnh->Enqueue(Sandesh::recv_queue());
140 : }
141 :
142 : extern int PullSandeshGenStatsReq;
143 : extern int PullSandeshUVE;
144 : extern int PullSandeshTraceReq;
145 :
146 88 : bool Sandesh::Initialize(SandeshRole::type role,
147 : const std::string &module,
148 : const std::string &source,
149 : const std::string &node_type,
150 : const std::string &instance_id,
151 : EventManager *evm,
152 : unsigned short http_port,
153 : SandeshContext *client_context,
154 : const SandeshConfig &config) {
155 88 : PullSandeshGenStatsReq = 1;
156 88 : PullSandeshUVE = 1;
157 88 : PullSandeshTraceReq = 1;
158 :
159 88 : if (role_ != SandeshRole::Invalid || role == SandeshRole::Invalid) {
160 2 : return true;
161 : }
162 :
163 86 : SANDESH_LOG(INFO, "SANDESH: ROLE : " << SandeshRoleToString(role));
164 86 : SANDESH_LOG(INFO, "SANDESH: MODULE : " << module);
165 86 : SANDESH_LOG(INFO, "SANDESH: SOURCE : " << source);
166 86 : SANDESH_LOG(INFO, "SANDESH: NODE TYPE : " << node_type);
167 86 : SANDESH_LOG(INFO, "SANDESH: INSTANCE ID : " << instance_id);
168 86 : SANDESH_LOG(INFO, "SANDESH: HTTP SERVER PORT : " << http_port);
169 :
170 86 : role_ = role;
171 86 : module_ = module;
172 86 : source_ = source;
173 86 : node_type_ = node_type;
174 86 : instance_id_ = instance_id;
175 86 : client_context_ = client_context;
176 86 : config_ = config;
177 86 : event_manager_ = evm;
178 :
179 86 : set_send_rate_limit(config.system_logs_rate_limit);
180 86 : DisableSendingObjectLogs(config.disable_object_logs);
181 86 : InitReceive(Task::kTaskInstanceAny);
182 86 : bool success(SandeshHttp::Init(evm, module, http_port,
183 : &SandeshHttpCallback, &http_port_, config_));
184 86 : if (!success) {
185 0 : SANDESH_LOG(ERROR, "SANDESH: HTTP INIT FAILED (PORT " <<
186 : http_port << ")");
187 0 : return false;
188 : }
189 86 : RecordPort("http", module_, http_port_);
190 86 : return true;
191 : }
192 :
193 86 : void Sandesh::RecordPort(const std::string& name, const std::string& module, unsigned short port) {
194 : int fd;
195 86 : std::ostringstream myfifoss;
196 86 : myfifoss << "/tmp/" << module << "." << getppid() << "." << name << "_port";
197 86 : std::string myfifo = myfifoss.str();
198 86 : std::ostringstream hss;
199 86 : hss << port << "\n";
200 86 : std::string hstr = hss.str();
201 :
202 86 : fd = open(myfifo.c_str(), O_WRONLY | O_NONBLOCK);
203 86 : if (fd != -1) {
204 44 : SANDESH_LOG(INFO, "SANDESH: Write " << name << "_port " << port <<
205 : "TO : " << myfifo);
206 44 : write(fd, hstr.c_str(), hstr.length());
207 44 : close(fd);
208 : } else {
209 42 : SANDESH_LOG(INFO, "SANDESH: NOT Writing " << name << "_port " << port <<
210 : "TO : " << myfifo);
211 : }
212 86 : }
213 :
214 37 : bool Sandesh::ConnectToCollector(const std::string &collector_ip,
215 : int collector_port, bool periodicuve) {
216 37 : boost::system::error_code ec;
217 37 : address collector_addr = AddressFromString(collector_ip, &ec);
218 37 : if (ec) {
219 0 : SANDESH_LOG(ERROR, __func__ << ": Invalid collector address: " <<
220 : collector_ip << " Error: " << ec);
221 0 : return false;
222 : }
223 :
224 37 : SANDESH_LOG(INFO, "SANDESH: COLLECTOR : " << collector_ip);
225 37 : SANDESH_LOG(INFO, "SANDESH: COLLECTOR PORT : " << collector_port);
226 :
227 37 : tcp::endpoint collector(collector_addr, collector_port);
228 37 : InitClient(event_manager_, collector, Sandesh::config(), periodicuve);
229 37 : return true;
230 : }
231 :
232 0 : void Sandesh::ReConfigCollectors(const std::vector<std::string>& collector_list) {
233 0 : if (client_) {
234 0 : client_->ReConfigCollectors(collector_list);
235 : }
236 0 : }
237 :
238 46 : bool Sandesh::InitClient(EventManager *evm,
239 : const std::vector<std::string> &collectors,
240 : const SandeshConfig &config) {
241 46 : connect_to_collector_ = true;
242 46 : SANDESH_LOG(INFO, "SANDESH: CONNECT TO COLLECTOR: " <<
243 : connect_to_collector_);
244 46 : std::vector<Endpoint> collector_endpoints;
245 152 : BOOST_FOREACH(const std::string &collector, collectors) {
246 53 : Endpoint ep;
247 53 : if (!MakeEndpoint(&ep, collector)) {
248 0 : SANDESH_LOG(ERROR, __func__ << ": Invalid collector address: " <<
249 : collector);
250 0 : return false;
251 : }
252 53 : collector_endpoints.push_back(ep);
253 : }
254 46 : client_ = new SandeshClient(evm, collector_endpoints, config, true);
255 46 : client_->Initiate();
256 46 : return true;
257 46 : }
258 :
259 40 : bool Sandesh::InitGenerator(const std::string &module,
260 : const std::string &source,
261 : const std::string &node_type,
262 : const std::string &instance_id,
263 : EventManager *evm,
264 : unsigned short http_port,
265 : SandeshContext *client_context,
266 : DerivedStats ds,
267 : const SandeshConfig &config) {
268 40 : assert(SandeshUVETypeMaps::InitDerivedStats(ds));
269 40 : return Initialize(SandeshRole::Generator, module, source, node_type,
270 40 : instance_id, evm, http_port, client_context, config);
271 : }
272 :
273 46 : bool Sandesh::InitGenerator(const std::string &module,
274 : const std::string &source,
275 : const std::string &node_type,
276 : const std::string &instance_id,
277 : EventManager *evm,
278 : unsigned short http_port,
279 : const std::vector<std::string> &collectors,
280 : SandeshContext *client_context,
281 : DerivedStats ds,
282 : const SandeshConfig &config) {
283 46 : assert(SandeshUVETypeMaps::InitDerivedStats(ds));
284 46 : bool success(Initialize(SandeshRole::Generator, module, source, node_type,
285 : instance_id, evm, http_port, client_context,
286 : config));
287 46 : if (!success) {
288 0 : return false;
289 : }
290 46 : return InitClient(evm, collectors, config);
291 : }
292 :
293 : // Collector
294 0 : bool Sandesh::InitCollector(const std::string &module,
295 : const std::string &source,
296 : const std::string &node_type,
297 : const std::string &instance_id,
298 : EventManager *evm,
299 : const std::string &collector_ip, int collector_port,
300 : unsigned short http_port,
301 : SandeshContext *client_context,
302 : const SandeshConfig &config) {
303 0 : bool success(Initialize(SandeshRole::Collector, module, source, node_type,
304 : instance_id, evm, http_port, client_context,
305 : config));
306 0 : if (!success) {
307 0 : return false;
308 : }
309 0 : return ConnectToCollector(collector_ip, collector_port, true);
310 : }
311 :
312 2 : bool Sandesh::InitGeneratorTest(const std::string &module,
313 : const std::string &source,
314 : const std::string &node_type,
315 : const std::string &instance_id,
316 : EventManager *evm,
317 : unsigned short http_port,
318 : SandeshContext *client_context,
319 : const SandeshConfig &config) {
320 2 : return Initialize(SandeshRole::Test, module, source, node_type,
321 2 : instance_id, evm, http_port, client_context, config);
322 : }
323 :
324 83 : static void WaitForIdle() {
325 : static const int kTimeout = 15;
326 83 : TaskScheduler *scheduler = TaskScheduler::GetInstance();
327 :
328 167 : for (int i = 0; i < (kTimeout * 1000); i++) {
329 167 : if (scheduler->IsEmpty()) {
330 83 : break;
331 : }
332 84 : usleep(1000);
333 : }
334 83 : }
335 :
336 9 : void Sandesh::SetDscpValue(uint8_t value) {
337 9 : SandeshClient *client = Sandesh::client();
338 9 : if (client) {
339 0 : client->SetDscpValue(value);
340 : }
341 9 : }
342 :
343 86 : void Sandesh::Uninit() {
344 :
345 : // Wait until all pending http session based tasks are cleaned up.
346 86 : long count = 60000;
347 86 : while (count--) {
348 86 : if (HttpSession::GetPendingTaskCount() == 0) break;
349 0 : usleep(1000);
350 : }
351 86 : SandeshHttp::Uninit();
352 86 : role_ = SandeshRole::Invalid;
353 86 : if (recv_queue_.get() != NULL) {
354 86 : recv_queue_->Shutdown();
355 86 : recv_queue_.reset(NULL);
356 86 : assert(recv_task_id_ != -1);
357 86 : recv_task_id_ = -1;
358 : } else {
359 0 : assert(recv_task_id_ == -1);
360 : }
361 86 : if (client_ != NULL) {
362 83 : client_->Shutdown();
363 170 : while (client()->IsSession()) usleep(100);
364 83 : WaitForIdle();
365 83 : client_->ClearSessions();
366 83 : TcpServerManager::DeleteServer(client_);
367 83 : client_ = NULL;
368 : }
369 86 : }
370 :
371 60 : void Sandesh::SetLoggingParams(bool enable_local_log, std::string category,
372 : std::string level, bool enable_trace_print, bool enable_flow_log,
373 : bool enable_session_syslog) {
374 60 : SetLocalLogging(enable_local_log);
375 60 : SetLoggingCategory(category);
376 60 : SetLoggingLevel(level);
377 60 : SetTracePrint(enable_trace_print);
378 60 : SetFlowLogging(enable_flow_log);
379 60 : SetSessionSyslogging(enable_session_syslog);
380 60 : }
381 :
382 273 : void Sandesh::SetLoggingParams(bool enable_local_log, std::string category,
383 : SandeshLevel::type level, bool enable_trace_print,
384 : bool enable_flow_log) {
385 273 : SetLocalLogging(enable_local_log);
386 273 : SetLoggingCategory(category);
387 273 : SetLoggingLevel(level);
388 273 : SetTracePrint(enable_trace_print);
389 273 : SetFlowLogging(enable_flow_log);
390 273 : }
391 :
392 60 : void Sandesh::SetLoggingLevel(std::string level) {
393 60 : SandeshLevel::type nlevel = StringToLevel(level);
394 60 : SetLoggingLevel(nlevel);
395 60 : }
396 :
397 12362651 : log4cplus::LogLevel SandeshLevelTolog4Level(
398 : SandeshLevel::type slevel) {
399 12362651 : switch (slevel) {
400 24 : case SandeshLevel::SYS_EMERG:
401 : case SandeshLevel::SYS_ALERT:
402 : case SandeshLevel::SYS_CRIT:
403 24 : return log4cplus::FATAL_LOG_LEVEL;
404 5264 : case SandeshLevel::SYS_ERR:
405 5264 : return log4cplus::ERROR_LOG_LEVEL;
406 17480 : case SandeshLevel::SYS_WARN:
407 : case SandeshLevel::SYS_NOTICE:
408 17480 : return log4cplus::WARN_LOG_LEVEL;
409 795260 : case SandeshLevel::SYS_INFO:
410 795260 : return log4cplus::INFO_LOG_LEVEL;
411 4875652 : case SandeshLevel::SYS_DEBUG:
412 4875652 : return log4cplus::DEBUG_LOG_LEVEL;
413 6668971 : default:
414 : case SandeshLevel::INVALID:
415 6668971 : return log4cplus::ALL_LOG_LEVEL;
416 : }
417 : }
418 :
419 334 : void Sandesh::SetLoggingLevel(SandeshLevel::type level) {
420 334 : log4cplus::LogLevel log4_new_level(SandeshLevelTolog4Level(level));
421 334 : log4cplus::LogLevel log4_old_level(logger_.getLogLevel());
422 334 : if (logging_level_ != level ||
423 : log4_old_level != log4_new_level) {
424 : const log4cplus::LogLevelManager &log4level_manager(
425 240 : log4cplus::getLogLevelManager());
426 240 : SANDESH_LOG(INFO, "SANDESH: Logging: LEVEL: " << "[ " <<
427 : LevelToString(logging_level_) << " ] -> [ " <<
428 : LevelToString(level) << " ] log4level: [ " <<
429 : log4level_manager.toString(log4_old_level) <<
430 : " ] -> [ " <<
431 : log4level_manager.toString(log4_new_level) <<
432 : " ]");
433 240 : logging_level_ = level;
434 240 : logger_.setLogLevel(log4_new_level);
435 : // Set the LogLevel on rootLogger
436 240 : ::SetLoggingLevel(log4_new_level);
437 : }
438 334 : }
439 :
440 333 : void Sandesh::SetLoggingCategory(std::string category) {
441 333 : if (logging_category_ != category) {
442 0 : SANDESH_LOG(INFO, "SANDESH: Logging: CATEGORY: " <<
443 : (logging_category_.empty() ? "*" : logging_category_) << " -> " <<
444 : (category.empty() ? "*" : category));
445 0 : logging_category_ = category;
446 : }
447 333 : }
448 :
449 340 : void Sandesh::SetLocalLogging(bool enable_local_log) {
450 340 : if (enable_local_log_ != enable_local_log) {
451 206 : SANDESH_LOG(INFO, "SANDESH: Logging: " <<
452 : (enable_local_log_ ? "ENABLED" : "DISABLED") << " -> " <<
453 : (enable_local_log ? "ENABLED" : "DISABLED"));
454 206 : enable_local_log_ = enable_local_log;
455 : }
456 340 : }
457 :
458 333 : void Sandesh::SetTracePrint(bool enable_trace_print) {
459 333 : if (enable_trace_print_ != enable_trace_print) {
460 0 : SANDESH_LOG(INFO, "SANDESH: Trace: PRINT: " <<
461 : (enable_trace_print_ ? "ENABLED" : "DISABLED") << " -> " <<
462 : (enable_trace_print ? "ENABLED" : "DISABLED"));
463 0 : enable_trace_print_ = enable_trace_print;
464 : }
465 333 : }
466 :
467 333 : void Sandesh::SetFlowLogging(bool enable_flow_log) {
468 333 : if (enable_flow_log_ != enable_flow_log) {
469 0 : SANDESH_LOG(INFO, "SANDESH: Flow Logging: " <<
470 : (enable_flow_log_ ? "ENABLED" : "DISABLED") << " -> " <<
471 : (enable_flow_log ? "ENABLED" : "DISABLED"));
472 0 : enable_flow_log_ = enable_flow_log;
473 : }
474 333 : }
475 :
476 60 : void Sandesh::SetSessionSyslogging(bool enable_session_syslog) {
477 60 : if (enable_session_syslog_ != enable_session_syslog) {
478 0 : SANDESH_LOG(INFO, "SANDESH: Flow Logging: " <<
479 : (enable_session_syslog_ ? "ENABLED" : "DISABLED") << " -> " <<
480 : (enable_session_syslog ? "ENABLED" : "DISABLED"));
481 0 : enable_session_syslog_ = enable_session_syslog;
482 : }
483 :
484 60 : }
485 :
486 0 : void Sandesh::DisableFlowCollection(bool disable) {
487 0 : if (disable_flow_collection_ != disable) {
488 0 : SANDESH_LOG(INFO, "SANDESH: Disable Flow Collection: " <<
489 : disable_flow_collection_ << " -> " << disable);
490 0 : disable_flow_collection_ = disable;
491 : }
492 0 : }
493 :
494 0 : void Sandesh::DisableSendingAllMessages(bool disable) {
495 0 : if (disable_sending_all_ != disable) {
496 0 : SANDESH_LOG(INFO, "SANDESH: Disable Sending ALL Messages: " <<
497 : disable_sending_all_ << " -> " << disable);
498 0 : disable_sending_all_ = disable;
499 : }
500 0 : }
501 :
502 198774 : bool Sandesh::IsSendingAllMessagesDisabled() {
503 198774 : return disable_sending_all_;
504 : }
505 :
506 86 : void Sandesh::DisableSendingObjectLogs(bool disable) {
507 86 : if (disable_sending_object_logs_ != disable) {
508 0 : SANDESH_LOG(INFO, "SANDESH: Disable Sending Object Logs: " <<
509 : disable_sending_object_logs_ << " -> " << disable);
510 0 : disable_sending_object_logs_ = disable;
511 : }
512 86 : }
513 :
514 9595 : bool Sandesh::IsSendingObjectLogsDisabled() {
515 9595 : return disable_sending_object_logs_;
516 : }
517 :
518 0 : void Sandesh::DisableSendingFlows(bool disable) {
519 0 : if (disable_sending_flows_ != disable) {
520 0 : SANDESH_LOG(INFO, "SANDESH: Disable Sending Flows: " <<
521 : disable_sending_flows_ << " -> " << disable);
522 0 : disable_sending_flows_ = disable;
523 : }
524 0 : }
525 :
526 0 : bool Sandesh::IsSendingFlowsDisabled() {
527 0 : return disable_sending_flows_;
528 : }
529 :
530 184573 : bool Sandesh::IsSendingSystemLogsDisabled() {
531 184573 : return sandesh_send_ratelimit_ == 0;
532 : }
533 :
534 86 : void Sandesh::set_send_rate_limit(int rate_limit) {
535 86 : if (rate_limit >= 0) {
536 86 : SANDESH_LOG(INFO, "SANDESH: System Log Send Rate Limit: " <<
537 : sandesh_send_ratelimit_ << " -> " << rate_limit);
538 86 : sandesh_send_ratelimit_ = rate_limit;
539 : }
540 86 : }
541 :
542 29565 : uint32_t Sandesh::get_send_rate_limit() {
543 29565 : return sandesh_send_ratelimit_;
544 : }
545 :
546 12936 : bool Sandesh::Enqueue(SandeshQueue *queue) {
547 12936 : if (!queue) {
548 0 : if (IsLoggingDroppedAllowed(type())) {
549 0 : SANDESH_LOG(ERROR, __func__ << ": SandeshQueue NULL : Dropping Message: "
550 : << ToString());
551 : }
552 0 : UpdateTxMsgFailStats(name_, 0, SandeshTxDropReason::NoQueue);
553 0 : Release();
554 0 : return false;
555 : }
556 : //Frame an elemet object and enqueue it
557 12936 : SandeshElement elem(this);
558 12936 : if (!queue->Enqueue(elem)) {
559 : // XXX Change when WorkQueue implements bounded queues
560 0 : return true;
561 : }
562 12936 : return true;
563 : }
564 :
565 10 : bool Sandesh::ProcessRecv(SandeshRequest *rsnh) {
566 10 : rsnh->HandleRequest();
567 10 : rsnh->Release();
568 10 : return true;
569 : }
570 :
571 2280 : void SandeshRequest::Release() { self_.reset(); }
572 :
573 22880 : int32_t Sandesh::WriteBinary(u_int8_t *buf, u_int32_t buf_len,
574 : int *error) {
575 : int32_t xfer;
576 : boost::shared_ptr<TMemoryBuffer> btrans =
577 : boost::shared_ptr<TMemoryBuffer>(
578 22880 : new TMemoryBuffer(buf, buf_len));
579 22871 : btrans->setWriteBuffer(buf, buf_len);
580 : boost::shared_ptr<TBinaryProtocol> prot =
581 22867 : boost::shared_ptr<TBinaryProtocol>(new TBinaryProtocol(btrans));
582 22868 : xfer = Write(prot);
583 22868 : if (xfer < 0) {
584 2 : SANDESH_LOG(DEBUG, __func__ << "Write sandesh to " << buf_len <<
585 : " bytes FAILED" << std::endl);
586 0 : *error = EINVAL;
587 0 : return xfer;
588 : }
589 22866 : return xfer;
590 22866 : }
591 :
592 0 : int32_t Sandesh::ReadBinary(u_int8_t *buf, u_int32_t buf_len,
593 : int *error) {
594 0 : int32_t xfer = 0;
595 : boost::shared_ptr<TMemoryBuffer> btrans =
596 : boost::shared_ptr<TMemoryBuffer>(
597 0 : new TMemoryBuffer(buf, buf_len));
598 : boost::shared_ptr<TBinaryProtocol> prot =
599 0 : boost::shared_ptr<TBinaryProtocol>(new TBinaryProtocol(btrans));
600 0 : xfer = Read(prot);
601 0 : if (xfer < 0) {
602 0 : SANDESH_LOG(DEBUG, __func__ << "Read sandesh from " << buf_len <<
603 : " bytes FAILED" << std::endl);
604 0 : *error = EINVAL;
605 0 : return xfer;
606 : }
607 0 : return xfer;
608 0 : }
609 :
610 0 : int32_t Sandesh::WriteBinaryToFile(const std::string& path, int *error) {
611 : int32_t xfer;
612 : boost::shared_ptr<TSimpleFileTransport> btrans =
613 : boost::shared_ptr<TSimpleFileTransport>(
614 0 : new TSimpleFileTransport(path, false, true));
615 : boost::shared_ptr<TBinaryProtocol> prot =
616 0 : boost::shared_ptr<TBinaryProtocol>(new TBinaryProtocol(btrans));
617 0 : xfer = Write(prot);
618 0 : if (xfer < 0) {
619 0 : SANDESH_LOG(DEBUG, __func__ << "Write sandesh to file FAILED"
620 : << std::endl);
621 0 : *error = EINVAL;
622 0 : return xfer;
623 : }
624 0 : return xfer;
625 0 : }
626 :
627 0 : int32_t Sandesh::ReadBinaryFromFile(const std::string& path, int *error) {
628 : int32_t xfer;
629 : boost::shared_ptr<TSimpleFileTransport> btrans =
630 : boost::shared_ptr<TSimpleFileTransport>(
631 0 : new TSimpleFileTransport(path));
632 : boost::shared_ptr<TBinaryProtocol> prot =
633 0 : boost::shared_ptr<TBinaryProtocol>(new TBinaryProtocol(btrans));
634 0 : xfer = Read(prot);
635 0 : if (xfer < 0) {
636 0 : SANDESH_LOG(DEBUG, __func__ << "Read sandesh from file FAILED"
637 : << std::endl);
638 0 : *error = EINVAL;
639 0 : return xfer;
640 : }
641 0 : return xfer;
642 0 : }
643 :
644 22882 : int32_t Sandesh::ReceiveBinaryMsgOne(u_int8_t *buf, u_int32_t buf_len,
645 : int *error, SandeshContext *client_context) {
646 : int32_t xfer;
647 22882 : std::string sandesh_name;
648 : boost::shared_ptr<TMemoryBuffer> btrans =
649 : boost::shared_ptr<TMemoryBuffer>(
650 22882 : new TMemoryBuffer(buf, buf_len));
651 : boost::shared_ptr<TBinaryProtocol> prot =
652 22881 : boost::shared_ptr<TBinaryProtocol>(new TBinaryProtocol(btrans));
653 : // Extract sandesh name
654 22881 : xfer = prot->readSandeshBegin(sandesh_name);
655 22879 : if (xfer < 0) {
656 0 : SANDESH_LOG(DEBUG, __func__ << "Read sandesh begin from " << buf_len <<
657 : " bytes FAILED" << std::endl);
658 0 : *error = EINVAL;
659 0 : return xfer;
660 : }
661 : // Create and process the sandesh
662 22879 : Sandesh *sandesh = SandeshBaseFactory::CreateInstance(sandesh_name);
663 22879 : if (sandesh == NULL) {
664 0 : SANDESH_LOG(DEBUG, __func__ << " Unknown sandesh:" <<
665 : sandesh_name << std::endl);
666 0 : *error = EINVAL;
667 0 : return -1;
668 : }
669 : // Reinitialize buffer and protocol
670 45760 : btrans = boost::shared_ptr<TMemoryBuffer>(
671 45756 : new TMemoryBuffer(buf, buf_len));
672 22880 : prot = boost::shared_ptr<TBinaryProtocol>(new TBinaryProtocol(btrans));
673 22880 : xfer = sandesh->Read(prot);
674 22881 : if (xfer < 0) {
675 0 : SANDESH_LOG(DEBUG, __func__ << " Decoding " << sandesh_name << " FAILED" <<
676 : std::endl);
677 0 : *error = EINVAL;
678 0 : return xfer;
679 : }
680 22881 : SandeshBuffer * bsnh = dynamic_cast<SandeshBuffer *>(sandesh);
681 22881 : if (bsnh) bsnh->Process(client_context);
682 22882 : sandesh->Release();
683 22882 : return xfer;
684 22882 : }
685 :
686 0 : int32_t Sandesh::ReceiveBinaryMsg(u_int8_t *buf, u_int32_t buf_len,
687 : int *error, SandeshContext *client_context) {
688 0 : u_int32_t xfer = 0;
689 : int ret;
690 0 : while (xfer < buf_len) {
691 0 : ret = ReceiveBinaryMsgOne(buf + xfer, buf_len - xfer, error,
692 : client_context);
693 0 : if (ret < 0) {
694 0 : SANDESH_LOG(DEBUG, __func__ << "Read sandesh from " << buf_len <<
695 : " bytes at offset " << xfer << " FAILED (" <<
696 : error << ")");
697 0 : return ret;
698 : }
699 0 : xfer += ret;
700 : }
701 0 : return xfer;
702 : }
703 :
704 11621 : bool Sandesh::SendEnqueue() {
705 11621 : if (!client_) {
706 2236 : if (IsLoggingDroppedAllowed(type())) {
707 2236 : if (IsConnectToCollectorEnabled()) {
708 0 : SANDESH_LOG(ERROR, "SANDESH: No client: " << ToString());
709 : } else {
710 2236 : Log();
711 : }
712 : }
713 2236 : UpdateTxMsgFailStats(name_, 0, SandeshTxDropReason::NoClient);
714 2236 : Release();
715 2236 : return false;
716 : }
717 9385 : if (!client_->SendSandesh(this)) {
718 0 : if (IsLoggingDroppedAllowed(type())) {
719 0 : SANDESH_LOG(ERROR, "SANDESH: Send FAILED: " << ToString());
720 : }
721 0 : UpdateTxMsgFailStats(name_, 0,
722 : SandeshTxDropReason::ClientSendFailed);
723 0 : Release();
724 0 : return false;
725 : }
726 9386 : return true;
727 : }
728 :
729 11658 : bool Sandesh::Dispatch(SandeshConnection * sconn) {
730 : // Sandesh client does not have a connection
731 11658 : if (sconn) {
732 37 : return sconn->SendSandesh(this);
733 : } else {
734 11621 : return SendEnqueue();
735 : }
736 : }
737 :
738 1940 : bool SandeshResponse::Dispatch(SandeshConnection * sconn) {
739 1940 : assert(sconn == NULL);
740 5810 : if ((context().find("http%") == 0) ||
741 3870 : (context().find("https%") == 0)) {
742 10 : SandeshHttp::Response(this, context());
743 10 : return true;
744 : }
745 1930 : if (response_callback_) {
746 1930 : response_callback_(this);
747 : }
748 1930 : return Sandesh::Dispatch(sconn);
749 : }
750 :
751 0 : bool SandeshTrace::Dispatch(SandeshConnection * sconn) {
752 0 : assert(sconn == NULL);
753 0 : if ((0 == context().find("http%")) ||
754 0 : (0 == context().find("https%"))) {
755 0 : SandeshHttp::Response(this, context());
756 0 : return true;
757 : }
758 0 : return Sandesh::Dispatch();
759 : }
760 :
761 184324 : bool SandeshUVE::Dispatch(SandeshConnection * sconn) {
762 184324 : assert(sconn == NULL);
763 552984 : if ((0 == context().find("http%")) ||
764 368664 : (0 == context().find("https%"))) {
765 0 : SandeshHttp::Response(this, context());
766 0 : return true;
767 : }
768 184336 : if (client_) {
769 4610 : if (IsSendingAllMessagesDisabled()) {
770 0 : Log();
771 0 : UpdateTxMsgFailStats(Name(), 0,
772 : SandeshTxDropReason::SendingDisabled);
773 0 : Release();
774 0 : return false;
775 : }
776 : // SandeshUVE has an implicit send level of SandeshLevel::SYS_UVE
777 : // which is irrespective of the level set by the user in the Send.
778 : // This is needed so that the send queue does not grow unbounded.
779 : // Once the send queue's sending level reaches SandeshLevel::SYS_UVE
780 : // we will reset the connection to the collector to initiate resync
781 : // of the UVE cache
782 4610 : if (SandeshLevel::SYS_UVE >= SendingLevel()) {
783 0 : client_->CloseSMSession();
784 : }
785 4610 : if (!client_->SendSandeshUVE(this)) {
786 0 : SANDESH_LOG(ERROR, "SandeshUVE : Send FAILED: " << ToString());
787 0 : UpdateTxMsgFailStats(Name(), 0,
788 : SandeshTxDropReason::ClientSendFailed);
789 0 : Release();
790 0 : return false;
791 : }
792 4610 : return true;
793 : }
794 179726 : if (IsConnectToCollectorEnabled()) {
795 83 : SANDESH_LOG(ERROR, "SANDESH: No Client: " << ToString());
796 : } else {
797 179646 : Log();
798 : }
799 179708 : UpdateTxMsgFailStats(Name(), 0, SandeshTxDropReason::NoClient);
800 179805 : Release();
801 179804 : return false;
802 : }
803 :
804 10 : bool SandeshRequest::Enqueue(SandeshRxQueue *queue) {
805 10 : if (!queue) {
806 0 : SANDESH_LOG(ERROR, "SandeshRequest: No RxQueue: " << ToString());
807 0 : UpdateRxMsgFailStats(Name(), 0, SandeshRxDropReason::NoQueue);
808 0 : Release();
809 0 : return false;
810 : }
811 10 : if (!queue->Enqueue(this)) {
812 : // XXX Change when WorkQueue implements bounded queues
813 0 : return true;
814 : }
815 10 : return true;
816 : }
817 :
818 236178 : bool Sandesh::IsLevelUT(SandeshLevel::type level) {
819 236178 : return level >= SandeshLevel::UT_START &&
820 236178 : level <= SandeshLevel::UT_END;
821 : }
822 :
823 20788195 : bool Sandesh::IsLevelCategoryLoggingAllowed(SandeshType::type type,
824 : SandeshLevel::type level,
825 : const std::string& category) {
826 : // Do not log UVEs unless explicitly configured via setting log_level
827 : // to INVALID. This is to avoid flooding the log files with UVEs
828 20788195 : if (type == SandeshType::UVE) {
829 183239 : level = SandeshLevel::INVALID;
830 : }
831 20788195 : bool level_allowed = logging_level_ >= level;
832 20788195 : bool category_allowed = !logging_category_.empty() ?
833 199 : logging_category_ == category : true;
834 20788795 : return level_allowed && category_allowed;
835 : }
836 :
837 12973 : bool Sandesh::IsLoggingAllowed() const {
838 12973 : if (type_ == SandeshType::FLOW || type_ == SandeshType::SESSION) {
839 0 : return enable_flow_log_;
840 : } else {
841 25897 : return IsLocalLoggingEnabled() &&
842 25897 : IsLevelCategoryLoggingAllowed(type_, level_, category_);
843 : }
844 : }
845 :
846 3296 : bool Sandesh::IsLoggingDroppedAllowed(SandeshType::type type) {
847 3296 : if (type == SandeshType::FLOW || type == SandeshType::SESSION) {
848 0 : return enable_flow_log_;
849 : } else {
850 3296 : return true;
851 : }
852 : }
853 :
854 12133711 : const char * Sandesh::LevelToString(SandeshLevel::type level) {
855 12133711 : std::map<int, const char*>::const_iterator it = _SandeshLevel_VALUES_TO_NAMES.find(level);
856 12132380 : if (it != _SandeshLevel_VALUES_TO_NAMES.end()) {
857 12132245 : return it->second;
858 : } else {
859 0 : return "UNKNOWN";
860 : }
861 : }
862 :
863 106 : SandeshLevel::type Sandesh::StringToLevel(std::string level) {
864 106 : std::map<int, const char*>::const_iterator it = _SandeshLevel_VALUES_TO_NAMES.begin();
865 907 : while (it != _SandeshLevel_VALUES_TO_NAMES.end()) {
866 907 : if (strncmp(level.c_str(), it->second, strlen(it->second)) == 0) {
867 106 : return static_cast<SandeshLevel::type>(it->first);
868 : }
869 801 : it++;
870 : }
871 0 : return SandeshLevel::INVALID;
872 : }
873 :
874 86 : void Sandesh::UpdateRxMsgStats(const std::string &msg_name,
875 : uint64_t bytes) {
876 86 : std::scoped_lock lock(stats_mutex_);
877 86 : msg_stats_.UpdateRecv(msg_name, bytes);
878 86 : }
879 :
880 0 : void Sandesh::UpdateRxMsgFailStats(const std::string &msg_name,
881 : uint64_t bytes, SandeshRxDropReason::type dreason) {
882 0 : std::scoped_lock lock(stats_mutex_);
883 0 : msg_stats_.UpdateRecvFailed(msg_name, bytes, dreason);
884 0 : }
885 :
886 12973 : void Sandesh::UpdateTxMsgStats(const std::string &msg_name,
887 : uint64_t bytes) {
888 12973 : std::scoped_lock lock(stats_mutex_);
889 12973 : msg_stats_.UpdateSend(msg_name, bytes);
890 12973 : }
891 :
892 367551 : void Sandesh::UpdateTxMsgFailStats(const std::string &msg_name,
893 : uint64_t bytes, SandeshTxDropReason::type dreason) {
894 367551 : std::scoped_lock lock(stats_mutex_);
895 367681 : msg_stats_.UpdateSendFailed(msg_name, bytes, dreason);
896 367681 : }
897 :
898 659 : void Sandesh::GetMsgStats(
899 : std::vector<SandeshMessageTypeStats> *mtype_stats,
900 : SandeshMessageStats *magg_stats) {
901 659 : std::scoped_lock lock(stats_mutex_);
902 659 : msg_stats_.Get(mtype_stats, magg_stats);
903 659 : }
904 :
905 0 : void Sandesh::GetMsgStats(
906 : boost::ptr_map<std::string, SandeshMessageTypeStats> *mtype_stats,
907 : SandeshMessageStats *magg_stats) {
908 0 : std::scoped_lock lock(stats_mutex_);
909 0 : msg_stats_.Get(mtype_stats, magg_stats);
910 0 : }
911 :
912 0 : void Sandesh::SetSendQueue(bool enable) {
913 0 : if (send_queue_enabled_ != enable) {
914 0 : SANDESH_LOG(INFO, "SANDESH: CLIENT: SEND QUEUE: " <<
915 : (send_queue_enabled_ ? "ENABLED" : "DISABLED") << " -> " <<
916 : (enable ? "ENABLED" : "DISABLED"));
917 0 : send_queue_enabled_ = enable;
918 0 : if (enable) {
919 0 : if (client_ && client_->IsSession()) {
920 0 : client_->session()->send_queue()->MayBeStartRunner();
921 : }
922 : }
923 : }
924 0 : }
925 :
926 : // In sandesh state machine on collector, we can only drop systemlog,
927 : // objectlog, and flow. UVEs can only be dropped after being dequeued from
928 : // the state machine and published on redis/kafka
929 12449 : bool DoDropSandeshMessage(const SandeshHeader &header,
930 : const SandeshLevel::type drop_level) {
931 12449 : SandeshType::type stype(header.get_Type());
932 12449 : if (stype == SandeshType::SYSTEM ||
933 3395 : stype == SandeshType::OBJECT ||
934 3395 : stype == SandeshType::FLOW ||
935 : stype == SandeshType::SESSION) {
936 : // Is level above drop level?
937 : SandeshLevel::type slevel(
938 9054 : static_cast<SandeshLevel::type>(header.get_Level()));
939 9054 : if (slevel >= drop_level) {
940 0 : return true;
941 : }
942 : }
943 : // Drop flow message if flow collection is disabled
944 12449 : if (Sandesh::IsFlowCollectionDisabled() &&
945 0 : (stype == SandeshType::FLOW || stype == SandeshType::SESSION)) {
946 0 : return true;
947 : }
948 12449 : return false;
949 : }
950 :
951 359 : SandeshContext *Sandesh::module_context(const std::string &module_name) {
952 359 : ModuleContextMap::const_iterator loc = module_context_.find(module_name);
953 359 : if (loc != module_context_.end()) {
954 359 : return loc->second;
955 : }
956 0 : return NULL;
957 : }
958 :
959 217 : void Sandesh::set_module_context(const std::string &module_name,
960 : SandeshContext *context) {
961 : std::pair<ModuleContextMap::iterator, bool> result =
962 217 : module_context_.insert(std::make_pair(module_name, context));
963 217 : if (!result.second) {
964 203 : result.first->second = context;
965 : }
966 217 : }
967 :
968 12135077 : bool Sandesh::HandleTest(SandeshLevel::type level,
969 : const std::string& category) {
970 : // Handle unit test scenario
971 12135077 : if (IsUnitTest() || IsLevelUT(level)) {
972 11941134 : return true;
973 : }
974 194185 : return false;
975 : }
976 :
977 14205 : SandeshLevel::type Sandesh::SendingLevel() {
978 14205 : if (client_) {
979 13889 : SandeshSession *sess = client_->session();
980 13889 : if (sess) {
981 13449 : return sess->SendingLevel();
982 : }
983 : }
984 756 : return SandeshLevel::INVALID;
985 : }
986 :
987 : template<>
988 12973 : size_t Sandesh::SandeshQueue::AtomicIncrementQueueCount(
989 : SandeshElement *element)
990 : {
991 12973 : size_t sandesh_size = element->GetSize();
992 25946 : return count_.fetch_add(sandesh_size) + sandesh_size;
993 : }
994 :
995 : template<>
996 12973 : size_t Sandesh::SandeshQueue::AtomicDecrementQueueCount(
997 : SandeshElement *element) {
998 12973 : size_t sandesh_size = element->GetSize();
999 25946 : return count_.fetch_sub(sandesh_size) - sandesh_size;
1000 : }
1001 :
1002 : /*
1003 : * Add the configured appenders for the sample logger
1004 : */
1005 4 : void Sandesh::set_logger_appender(const std::string &file_name, long max_file_size,
1006 : int max_backup_index,
1007 : const std::string &syslog_facility,
1008 : const std::vector<std::string> &destn,
1009 : const std::string &ident,
1010 : bool is_sampled_logger) {
1011 4 : log4cplus::Logger logger;
1012 4 : if (is_sampled_logger) {
1013 2 : logger = Sandesh::sampled_logger();
1014 : } else {
1015 2 : logger = Sandesh::slo_logger();
1016 : }
1017 4 : logger.setAdditivity(false);
1018 : // Session messages log level is
1019 4 : logger.setLogLevel(SandeshLevel::SYS_NOTICE);
1020 : // Local logging for SLO logger
1021 4 : if (std::find(destn.begin(), destn.end(), "file") !=
1022 8 : destn.end()) {
1023 : // Append file appender to SLO logger
1024 : SharedAppenderPtr fileappender(new RollingFileAppender(file_name,
1025 0 : max_file_size, max_backup_index));
1026 0 : logger.addAppender(fileappender);
1027 0 : if (is_sampled_logger) {
1028 0 : Sandesh::sampled_to_logger_ = true;
1029 : } else {
1030 0 : Sandesh::slo_to_logger_ = true;
1031 : }
1032 0 : }
1033 : // SYSLOG appender for SLO logger
1034 4 : if (std::find(destn.begin(), destn.end(), "syslog") !=
1035 8 : destn.end()) {
1036 0 : helpers::Properties props;
1037 : std::string syslogident = boost::str(
1038 0 : boost::format("%1%[%2%]") % ident % getpid());
1039 0 : props.setProperty(LOG4CPLUS_TEXT("facility"),
1040 0 : boost::starts_with(syslog_facility, "LOG_")
1041 0 : ? syslog_facility.substr(4)
1042 : : syslog_facility);
1043 0 : props.setProperty(LOG4CPLUS_TEXT("ident"), syslogident);
1044 0 : props.setProperty(LOG4CPLUS_TEXT("additivity"), "false");
1045 0 : SharedAppenderPtr syslogappender(new SysLogAppender(props));
1046 : std::unique_ptr<Layout> syslog_layout_ptr(new PatternLayout(
1047 0 : loggingPattern));
1048 0 : syslogappender->setLayout(std::move(syslog_layout_ptr));
1049 0 : logger.addAppender(syslogappender);
1050 0 : if (is_sampled_logger) {
1051 0 : Sandesh::sampled_to_logger_ = true;
1052 : } else {
1053 0 : Sandesh::slo_to_logger_ = true;
1054 : }
1055 0 : }
1056 4 : }
1057 :
1058 2 : void Sandesh::set_send_to_collector_flags(
1059 : const std::vector<std::string> &sampled_destn,
1060 : const std::vector<std::string> &slo_destn) {
1061 2 : if (std::find(slo_destn.begin(), slo_destn.end(), "collector") !=
1062 4 : slo_destn.end()) {
1063 2 : Sandesh::slo_to_collector_ = true;
1064 : }
1065 2 : if (std::find(sampled_destn.begin(), sampled_destn.end(), "collector") !=
1066 4 : sampled_destn.end()) {
1067 2 : Sandesh::sampled_to_collector_ = true;
1068 : }
1069 2 : }
|
