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_client.cc
8 : //
9 : // Sandesh Client
10 : //
11 :
12 : #include <boost/bind/bind.hpp>
13 : #include <boost/assign.hpp>
14 : #include <boost/foreach.hpp>
15 :
16 : #include <base/task_annotations.h>
17 : #include <base/time_util.h>
18 : #include <io/event_manager.h>
19 : #include <io/tcp_session.h>
20 : #include <io/tcp_server.h>
21 :
22 : #include <sandesh/sandesh_constants.h>
23 : #include <sandesh/sandesh_types.h>
24 : #include <sandesh/sandesh.h>
25 : #include <sandesh/sandesh_trace.h>
26 : #include <sandesh/sandesh_session.h>
27 : #include <sandesh/sandesh_http.h>
28 :
29 : #include "sandesh_state_machine.h"
30 :
31 : #include <sandesh/protocol/TXMLProtocol.h>
32 : #include <sandesh/sandesh_ctrl_types.h>
33 : #include <sandesh/sandesh_uve_types.h>
34 : #include <sandesh/derived_stats_results_types.h>
35 : #include <sandesh/common/vns_constants.h>
36 : #include "sandesh_client.h"
37 : #include "sandesh_uve.h"
38 : #include "sandesh_util.h"
39 :
40 : using boost::asio::ip::address;
41 : using namespace boost::asio;
42 : using boost::system::error_code;
43 : using std::string;
44 : using std::map;
45 : using std::make_pair;
46 : using std::vector;
47 : using namespace boost::placeholders;
48 :
49 : const std::string SandeshClient::kSMTask = "sandesh::SandeshClientSM";
50 : const std::string SandeshClient::kSessionWriterTask = "sandesh::SandeshClientSession";
51 : const std::string SandeshClient::kSessionReaderTask = "sandesh::SandeshClientReader";
52 : bool SandeshClient::task_policy_set_ = false;
53 : const std::vector<Sandesh::QueueWaterMarkInfo>
54 : SandeshClient::kSessionWaterMarkInfo = boost::assign::tuple_list_of
55 : (50*1024*1024, SandeshLevel::SYS_UVE, true, false)
56 : (30*1024*1024, SandeshLevel::SYS_EMERG, true, false)
57 : (20*1024*1024, SandeshLevel::SYS_ERR, true, false)
58 : (1*1024*1024, SandeshLevel::SYS_DEBUG, true, false)
59 : (35*1024*1024, SandeshLevel::SYS_EMERG, false, false)
60 : (25*1024*1024, SandeshLevel::SYS_ERR, false, false)
61 : (15*1024*1024, SandeshLevel::SYS_DEBUG, false, false)
62 : (2*1024, SandeshLevel::INVALID, false, false);
63 :
64 83 : SandeshClient::SandeshClient(EventManager *evm,
65 : const std::vector<Endpoint> &collectors,
66 : const SandeshConfig &config,
67 83 : bool periodicuve)
68 : : SslServer(evm, boost::asio::ssl::context::tlsv12_client,
69 83 : config.sandesh_ssl_enable),
70 83 : sm_task_instance_(kSMTaskInstance),
71 83 : sm_task_id_(TaskScheduler::GetInstance()->GetTaskId(kSMTask)),
72 83 : session_task_instance_(kSessionTaskInstance),
73 83 : session_writer_task_id_(TaskScheduler::GetInstance()->GetTaskId(kSessionWriterTask)),
74 83 : session_reader_task_id_(TaskScheduler::GetInstance()->GetTaskId(kSessionReaderTask)),
75 83 : dscp_value_(0),
76 83 : collectors_(collectors),
77 83 : stats_collector_(config.stats_collector),
78 83 : sm_(SandeshClientSM::CreateClientSM(evm, this, sm_task_instance_, sm_task_id_,
79 : periodicuve)),
80 83 : session_wm_info_(kSessionWaterMarkInfo),
81 83 : session_close_interval_msec_(0),
82 166 : session_close_time_usec_(0) {
83 : // Set task policy for exclusion between state machine and session tasks since
84 : // session delete happens in state machine task
85 83 : if (!task_policy_set_) {
86 : TaskPolicy sm_task_policy = boost::assign::list_of
87 49 : (TaskExclusion(session_writer_task_id_))
88 49 : (TaskExclusion(session_reader_task_id_));
89 49 : TaskScheduler::GetInstance()->SetPolicy(sm_task_id_, sm_task_policy);
90 49 : task_policy_set_ = true;
91 49 : }
92 83 : if (config.sandesh_ssl_enable) {
93 1 : boost::asio::ssl::context *ctx = context();
94 1 : boost::system::error_code ec;
95 1 : ctx->set_options(boost::asio::ssl::context::default_workarounds |
96 : boost::asio::ssl::context::no_tlsv1 |
97 : boost::asio::ssl::context::no_sslv3 |
98 : boost::asio::ssl::context::no_sslv2 |
99 : boost::asio::ssl::context::no_tlsv1_1, ec);
100 1 : if (ec.value() != 0) {
101 0 : SANDESH_LOG(ERROR, "Error setting ssl options: " << ec.message());
102 0 : exit(EINVAL);
103 : }
104 : // CA certificate
105 1 : if (!config.ca_cert.empty()) {
106 : // Verify that the peer certificate is signed by a trusted CA
107 1 : ctx->set_verify_mode(boost::asio::ssl::verify_peer |
108 : boost::asio::ssl::verify_fail_if_no_peer_cert,
109 : ec);
110 1 : if (ec.value() != 0) {
111 0 : SANDESH_LOG(ERROR, "Error setting verification mode: " <<
112 : ec.message());
113 0 : exit(EINVAL);
114 : }
115 1 : ctx->load_verify_file(config.ca_cert, ec);
116 1 : if (ec.value() != 0) {
117 0 : SANDESH_LOG(ERROR, "Error loading CA certificate: " <<
118 : ec.message());
119 0 : exit(EINVAL);
120 : }
121 : }
122 : // Server certificate
123 1 : ctx->use_certificate_chain_file(config.certfile, ec);
124 1 : if (ec.value() != 0) {
125 0 : SANDESH_LOG(ERROR, "Error using server certificate: " <<
126 : ec.message());
127 0 : exit(EINVAL);
128 : }
129 : // Server private key
130 1 : ctx->use_private_key_file(config.keyfile,
131 : boost::asio::ssl::context::pem, ec);
132 1 : if (ec.value() != 0) {
133 0 : SANDESH_LOG(ERROR, "Error using server private key file: " <<
134 : ec.message());
135 0 : exit(EINVAL);
136 : }
137 : }
138 83 : if (stats_collector_ != "") {
139 0 : UdpServer::Endpoint stats_server;
140 0 : size_t found = stats_collector_.find(":");
141 0 : if (found != std::string::npos) {
142 0 : stats_client_.reset(new StatsClientRemote(*evm->io_service(), stats_collector_));
143 : } else {
144 : #if defined(BOOST_ASIO_HAS_LOCAL_SOCKETS)
145 0 : stats_client_.reset(new StatsClientLocal(*evm->io_service(), stats_collector_));
146 : #else
147 : SANDESH_LOG(ERROR, "Unix Domain Sockets are not supported on this platform");
148 : #endif
149 : }
150 : }
151 83 : }
152 :
153 166 : SandeshClient::~SandeshClient() {}
154 :
155 0 : void SandeshClient::ReConfigCollectors(
156 : const std::vector<std::string>& collector_list) {
157 0 : std::vector<Endpoint> collector_endpoints;
158 :
159 0 : BOOST_FOREACH(const std::string& collector, collector_list) {
160 0 : Endpoint ep;
161 0 : if (!MakeEndpoint(&ep, collector)) {
162 0 : SANDESH_LOG(ERROR, __func__ << ": Invalid collector address: " <<
163 : collector);
164 0 : return;
165 : }
166 0 : collector_endpoints.push_back(ep);
167 : }
168 0 : sm_->SetCollectors(collector_endpoints);
169 0 : }
170 :
171 83 : void SandeshClient::Initiate() {
172 83 : sm_->SetAdminState(false);
173 83 : if (collectors_.size())
174 83 : sm_->SetCollectors(collectors_);
175 83 : if (stats_collector_ != "") {
176 0 : stats_client_->Initialize();
177 : }
178 83 : }
179 :
180 83 : void SandeshClient::Shutdown() {
181 83 : sm_->SetAdminState(true);
182 83 : }
183 :
184 9386 : bool SandeshClient::SendSandesh(Sandesh *snh) {
185 9386 : return sm_->SendSandesh(snh);
186 : }
187 :
188 4610 : bool SandeshClient::SendSandeshUVE(Sandesh *snh) {
189 4610 : return sm_->SendSandeshUVE(snh);
190 : }
191 :
192 86 : bool SandeshClient::ReceiveCtrlMsg(const std::string &msg,
193 : const SandeshHeader &header, const std::string &sandesh_name,
194 : const uint32_t header_offset) {
195 :
196 86 : Sandesh * sandesh = SandeshSession::DecodeCtrlSandesh(msg, header, sandesh_name, header_offset);
197 :
198 86 : const SandeshCtrlServerToClient * snh = dynamic_cast<const SandeshCtrlServerToClient *>(sandesh);
199 86 : if (!snh) {
200 0 : SANDESH_LOG(ERROR, "Received Ctrl Message with wrong type " << sandesh->Name());
201 0 : sandesh->Release();
202 0 : return false;
203 : }
204 86 : if (!snh->get_success()) {
205 0 : SANDESH_LOG(ERROR, "Received Ctrl Message : Connection with server has failed");
206 0 : sandesh->Release();
207 0 : return false;
208 : }
209 86 : SANDESH_LOG(DEBUG, "Received Ctrl Message with size " << snh->get_type_info().size());
210 :
211 86 : map<string,uint32_t> sMap;
212 86 : const vector<UVETypeInfo> & vu = snh->get_type_info();
213 86 : for(uint32_t i = 0; i < vu.size(); i++) {
214 0 : sMap.insert(std::make_pair(vu[i].get_type_name(), vu[i].get_seq_num()));
215 : }
216 86 : SandeshUVETypeMaps::SyncAllMaps(sMap);
217 :
218 86 : sandesh->Release();
219 86 : return true;
220 86 : }
221 :
222 :
223 86 : bool SandeshClient::ReceiveMsg(const std::string& msg,
224 : const SandeshHeader &header, const std::string &sandesh_name,
225 : const uint32_t header_offset) {
226 :
227 : namespace sandesh_prot = contrail::sandesh::protocol;
228 : namespace sandesh_trans = contrail::sandesh::transport;
229 :
230 86 : if (header.get_Hints() & g_sandesh_constants.SANDESH_CONTROL_HINT) {
231 86 : bool success = ReceiveCtrlMsg(msg, header, sandesh_name, header_offset);
232 86 : if (success) {
233 86 : Sandesh::UpdateRxMsgStats(sandesh_name, msg.size());
234 : } else {
235 0 : Sandesh::UpdateRxMsgFailStats(sandesh_name, msg.size(),
236 : SandeshRxDropReason::ControlMsgFailed);
237 : }
238 86 : return success;
239 : }
240 :
241 : // Create and process the sandesh
242 0 : Sandesh *sandesh = SandeshBaseFactory::CreateInstance(sandesh_name);
243 0 : if (sandesh == NULL) {
244 0 : SANDESH_LOG(ERROR, __func__ << ": Unknown sandesh: " << sandesh_name);
245 0 : Sandesh::UpdateRxMsgFailStats(sandesh_name, msg.size(),
246 : SandeshRxDropReason::CreateFailed);
247 0 : return true;
248 : }
249 : boost::shared_ptr<sandesh_trans::TMemoryBuffer> btrans =
250 : boost::shared_ptr<sandesh_trans::TMemoryBuffer>(
251 0 : new sandesh_trans::TMemoryBuffer((uint8_t *)msg.c_str() + header_offset,
252 0 : msg.size() - header_offset));
253 : boost::shared_ptr<sandesh_prot::TXMLProtocol> prot =
254 0 : boost::shared_ptr<sandesh_prot::TXMLProtocol>(new sandesh_prot::TXMLProtocol(btrans));
255 0 : int32_t xfer = sandesh->Read(prot);
256 0 : if (xfer < 0) {
257 0 : SANDESH_LOG(ERROR, __func__ << ": Decoding " << sandesh_name << " FAILED");
258 0 : Sandesh::UpdateRxMsgFailStats(sandesh_name, msg.size(),
259 : SandeshRxDropReason::DecodingFailed);
260 0 : return false;
261 : }
262 :
263 0 : Sandesh::UpdateRxMsgStats(sandesh_name, msg.size());
264 0 : SandeshRequest *sr = dynamic_cast<SandeshRequest *>(sandesh);
265 0 : assert(sr);
266 0 : sr->Enqueue(Sandesh::recv_queue());
267 0 : return true;
268 0 : }
269 :
270 : void
271 0 : SandeshCtrlServerToClient::HandleRequest() const { }
272 :
273 : void
274 0 : SandeshCtrlClientToServer::HandleRequest() const { }
275 :
276 :
277 105 : SandeshSession *SandeshClient::CreateSMSession(
278 : TcpSession::EventObserver eocb,
279 : SandeshReceiveMsgCb rmcb,
280 : TcpServer::Endpoint ep) {
281 105 : TcpSession *session = SslServer::CreateSession();
282 105 : Socket *socket = session->socket();
283 :
284 105 : error_code ec;
285 105 : socket->open(ip::tcp::v4(), ec);
286 105 : if (ec) {
287 0 : SANDESH_LOG(ERROR, __func__ << " Open FAILED: " << ec.message());
288 0 : DeleteSession(session);
289 0 : return NULL;
290 : }
291 105 : ec = session->SetSocketOptions();
292 105 : if (ec) {
293 0 : SANDESH_LOG(ERROR, __func__ << " Unable to set socket options: " << ec.message());
294 0 : DeleteSession(session);
295 0 : return NULL;
296 : }
297 105 : if (dscp_value_) {
298 0 : session->SetDscpSocketOption(dscp_value_);
299 : }
300 105 : SandeshSession *sandesh_session =
301 : static_cast<SandeshSession *>(session);
302 105 : sandesh_session->SetReceiveMsgCb(rmcb);
303 105 : sandesh_session->SetConnection(NULL);
304 105 : sandesh_session->set_observer(eocb);
305 : // Set watermarks
306 945 : for (size_t i = 0; i < session_wm_info_.size(); i++) {
307 840 : sandesh_session->SetSendQueueWaterMark(session_wm_info_[i]);
308 : }
309 105 : TcpServer::Connect(sandesh_session, ep);
310 :
311 105 : return sandesh_session;
312 : }
313 :
314 96 : void SandeshClient::InitializeSMSession(int count) {
315 96 : std::vector<string> stv;
316 :
317 : SandeshUVETypeMaps::uve_global_map::const_iterator it =
318 96 : SandeshUVETypeMaps::Begin();
319 555 : for(; it!= SandeshUVETypeMaps::End(); it++) {
320 459 : stv.push_back(it->first);
321 : }
322 96 : SANDESH_LOG(DEBUG, "Sending Ctrl Message for " << Sandesh::source() << ":" <<
323 : Sandesh::module() << ":" << Sandesh::instance_id() << ":" <<
324 : Sandesh::node_type() << " count " << count);
325 :
326 288 : SandeshCtrlClientToServer::Request(Sandesh::source(), Sandesh::module(),
327 96 : count, stv, getpid(), Sandesh::http_port(),
328 192 : Sandesh::node_type(), Sandesh::instance_id(), "ctrl");
329 :
330 96 : }
331 :
332 0 : bool SandeshClient::CloseSMSessionInternal() {
333 0 : SandeshSession *session(sm_->session());
334 0 : if (session) {
335 0 : session->EnqueueClose();
336 0 : return true;
337 : }
338 0 : return false;
339 : }
340 :
341 0 : bool DoCloseSMSession(uint64_t now_usec, uint64_t last_close_usec,
342 : uint64_t last_close_interval_usec, int *close_interval_msec) {
343 : // If this is the first time, we will accept the next close
344 : // only after the initial close interval time
345 0 : if (last_close_interval_usec == 0 || last_close_usec == 0) {
346 0 : *close_interval_msec =
347 : SandeshClient::kInitialSMSessionCloseIntervalMSec;
348 0 : return true;
349 : }
350 0 : assert(now_usec >= last_close_usec);
351 0 : uint64_t time_since_close_usec(now_usec - last_close_usec);
352 : // We will ignore close events receive before the last close
353 : // interval is finished
354 0 : if (time_since_close_usec <= last_close_interval_usec) {
355 0 : *close_interval_msec = 0;
356 0 : return false;
357 : }
358 : // We will double the close interval time if we get a close
359 : // event between last close interval and 2 * last close interval.
360 : // If the close event is between 2 * last close interval and
361 : // 4 * last close interval, then the close interval will be
362 : // same as the current close interval. If the close event is
363 : // after 4 * last close interval, then we will reset the close
364 : // interval to the initial close interval
365 0 : if (time_since_close_usec > last_close_interval_usec &&
366 0 : time_since_close_usec <= 2 * last_close_interval_usec) {
367 0 : uint64_t nclose_interval_msec((2 * last_close_interval_usec)/1000);
368 0 : *close_interval_msec = std::min(nclose_interval_msec,
369 0 : static_cast<uint64_t>(
370 : SandeshClient::kMaxSMSessionCloseIntervalMSec));
371 0 : return true;
372 0 : } else if ((2 * last_close_interval_usec <= time_since_close_usec) &&
373 0 : (time_since_close_usec <= 4 * last_close_interval_usec)) {
374 0 : *close_interval_msec = last_close_interval_usec/1000;
375 0 : return true;
376 : } else {
377 0 : *close_interval_msec =
378 : SandeshClient::kInitialSMSessionCloseIntervalMSec;
379 0 : return true;
380 : }
381 : }
382 :
383 0 : bool SandeshClient::CloseSMSession() {
384 0 : uint64_t now_usec(UTCTimestampUsec());
385 0 : int close_interval_msec(0);
386 0 : bool close(DoCloseSMSession(now_usec, session_close_time_usec_,
387 0 : session_close_interval_msec_ * 1000, &close_interval_msec));
388 0 : if (close) {
389 0 : session_close_time_usec_ = now_usec;
390 0 : session_close_interval_msec_ = close_interval_msec;
391 0 : return CloseSMSessionInternal();
392 : }
393 0 : return false;
394 : }
395 :
396 : static bool client_start = false;
397 : static uint64_t client_start_time;
398 :
399 659 : void SandeshClient::SendUVE(int count,
400 : const string & stateName, const string & server,
401 : const Endpoint & server_ip,
402 : const std::vector<TcpServer::Endpoint> & collector_eps) {
403 1318 : ModuleClientState mcs;
404 1977 : mcs.set_name(Sandesh::source() + ":" + Sandesh::node_type() +
405 2636 : ":" + Sandesh::module() + ":" + Sandesh::instance_id());
406 659 : SandeshClientInfo sci;
407 659 : if (!client_start) {
408 49 : client_start_time = UTCTimestampUsec();
409 49 : client_start = true;
410 : }
411 659 : sci.set_start_time(client_start_time);
412 659 : sci.set_successful_connections(count);
413 659 : sci.set_pid(getpid());
414 659 : sci.set_http_port(Sandesh::http_port());
415 659 : sci.set_status(stateName);
416 659 : sci.set_collector_name(server);
417 659 : std::ostringstream collector_ip;
418 659 : collector_ip << server_ip;
419 659 : sci.set_collector_ip(collector_ip.str());
420 659 : std::vector<std::string> collectors;
421 1827 : BOOST_FOREACH(const TcpServer::Endpoint& ep, collector_eps) {
422 584 : std::ostringstream collector_ip;
423 584 : collector_ip << ep;
424 584 : collectors.push_back(collector_ip.str());
425 584 : }
426 659 : sci.set_collector_list(collectors);
427 : // Sandesh client socket statistics
428 659 : SocketIOStats rx_stats;
429 659 : GetRxSocketStats(rx_stats);
430 659 : sci.set_rx_socket_stats(rx_stats);
431 659 : SocketIOStats tx_stats;
432 659 : GetTxSocketStats(tx_stats);
433 659 : sci.set_tx_socket_stats(tx_stats);
434 :
435 659 : mcs.set_client_info(sci);
436 :
437 659 : std::vector<SandeshMessageTypeStats> mtype_stats;
438 659 : SandeshMessageStats magg_stats;
439 659 : Sandesh::GetMsgStats(&mtype_stats, &magg_stats);
440 :
441 659 : map<string,uint64_t> csev;
442 659 : csev.insert(make_pair("sent", magg_stats.get_messages_sent()));
443 659 : csev.insert(make_pair("dropped_no_queue",
444 659 : magg_stats.get_messages_sent_dropped_no_queue()));
445 659 : csev.insert(make_pair("dropped_no_client",
446 659 : magg_stats.get_messages_sent_dropped_no_client()));
447 659 : csev.insert(make_pair("dropped_no_session",
448 659 : magg_stats.get_messages_sent_dropped_no_session()));
449 659 : csev.insert(make_pair("dropped_queue_level",
450 659 : magg_stats.get_messages_sent_dropped_queue_level()));
451 659 : csev.insert(make_pair("dropped_client_send_failed",
452 659 : magg_stats.get_messages_sent_dropped_client_send_failed()));
453 659 : csev.insert(make_pair("dropped_session_not_connected",
454 659 : magg_stats.get_messages_sent_dropped_session_not_connected()));
455 659 : csev.insert(make_pair("dropped_header_write_failed",
456 659 : magg_stats.get_messages_sent_dropped_header_write_failed()));
457 659 : csev.insert(make_pair("dropped_write_failed",
458 659 : magg_stats.get_messages_sent_dropped_write_failed()));
459 659 : csev.insert(make_pair("dropped_wrong_client_sm_state",
460 659 : magg_stats.get_messages_sent_dropped_wrong_client_sm_state()));
461 659 : csev.insert(make_pair("dropped_validation_failed",
462 659 : magg_stats.get_messages_sent_dropped_validation_failed()));
463 659 : csev.insert(make_pair("dropped_rate_limited",
464 659 : magg_stats.get_messages_sent_dropped_rate_limited()));
465 659 : csev.insert(make_pair("dropped_sending_disabled",
466 659 : magg_stats.get_messages_sent_dropped_sending_disabled()));
467 659 : csev.insert(make_pair("dropped_sending_to_syslog",
468 659 : magg_stats.get_messages_sent_dropped_sending_to_syslog()));
469 659 : mcs.set_tx_msg_agg(csev);
470 :
471 659 : map <string,SandeshMessageStats> csevm;
472 659 : for (vector<SandeshMessageTypeStats>::const_iterator smit = mtype_stats.begin();
473 15972 : smit != mtype_stats.end(); smit++) {
474 15313 : SandeshMessageStats res_sms;
475 15313 : const SandeshMessageStats& src_sms = smit->get_stats();
476 15313 : res_sms.set_messages_sent(src_sms.get_messages_sent());
477 15313 : res_sms.set_messages_sent_dropped_no_queue(
478 : src_sms.get_messages_sent_dropped_no_queue());
479 15313 : res_sms.set_messages_sent_dropped_no_client(
480 : src_sms.get_messages_sent_dropped_no_client());
481 15313 : res_sms.set_messages_sent_dropped_no_session(
482 : src_sms.get_messages_sent_dropped_no_session());
483 15313 : res_sms.set_messages_sent_dropped_queue_level(
484 : src_sms.get_messages_sent_dropped_queue_level());
485 15313 : res_sms.set_messages_sent_dropped_client_send_failed(
486 : src_sms.get_messages_sent_dropped_client_send_failed());
487 15313 : res_sms.set_messages_sent_dropped_session_not_connected(
488 : src_sms.get_messages_sent_dropped_session_not_connected());
489 15313 : res_sms.set_messages_sent_dropped_header_write_failed(
490 : src_sms.get_messages_sent_dropped_header_write_failed());
491 15313 : res_sms.set_messages_sent_dropped_write_failed(
492 : src_sms.get_messages_sent_dropped_write_failed());
493 15313 : res_sms.set_messages_sent_dropped_wrong_client_sm_state(
494 : src_sms.get_messages_sent_dropped_wrong_client_sm_state());
495 15313 : res_sms.set_messages_sent_dropped_validation_failed(
496 : src_sms.get_messages_sent_dropped_validation_failed());
497 15313 : res_sms.set_messages_sent_dropped_rate_limited(
498 : src_sms.get_messages_sent_dropped_rate_limited());
499 15313 : res_sms.set_messages_sent_dropped_sending_disabled(
500 : src_sms.get_messages_sent_dropped_sending_disabled());
501 15313 : res_sms.set_messages_sent_dropped_sending_to_syslog(
502 : src_sms.get_messages_sent_dropped_sending_to_syslog());
503 15313 : csevm.insert(make_pair(smit->get_message_type(), res_sms));
504 15313 : }
505 659 : mcs.set_msg_type_agg(csevm);
506 :
507 659 : SandeshModuleClientTrace::Send(mcs);
508 659 : }
509 :
510 0 : void SandeshClient::SetSessionWaterMarkInfo(
511 : Sandesh::QueueWaterMarkInfo &scwm) {
512 0 : SandeshSession *session = sm_->session();
513 0 : if (session) {
514 0 : session->SetSendQueueWaterMark(scwm);
515 : }
516 0 : session_wm_info_.push_back(scwm);
517 0 : }
518 :
519 0 : void SandeshClient::ResetSessionWaterMarkInfo() {
520 0 : SandeshSession *session = sm_->session();
521 0 : if (session) {
522 0 : session->ResetSendQueueWaterMark();
523 : }
524 0 : session_wm_info_.clear();
525 0 : }
526 :
527 0 : void SandeshClient::GetSessionWaterMarkInfo(
528 : std::vector<Sandesh::QueueWaterMarkInfo> &scwm_info) const {
529 0 : scwm_info = session_wm_info_;
530 0 : }
531 :
532 105 : SslSession *SandeshClient::AllocSession(SslSocket *socket) {
533 : return new SandeshSession(this, socket, session_task_instance_,
534 : session_writer_task_id_,
535 105 : session_reader_task_id_);
536 : }
537 :
538 0 : void SandeshClient::SetDscpValue(uint8_t value) {
539 0 : if (value == dscp_value_)
540 0 : return;
541 :
542 0 : dscp_value_ = value;
543 0 : SandeshSession *sess = session();
544 0 : if (sess) {
545 0 : sess->SetDscpSocketOption(value);
546 : }
547 0 : SandeshHttp::UpdateDscp(value);
548 : }
|
