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