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