Line data Source code
1 : /*
2 : * Copyright (c) 2017 Juniper Networks, Inc. All rights reserved.
3 : * Copyright Contributors to the OpenSDN project
4 : */
5 :
6 : #include <bitset>
7 : #include <pkt/flow_table.h>
8 : #include <pkt/flow_mgmt/flow_mgmt_request.h>
9 : #include <uve/stats_collector.h>
10 : #include <uve/interface_uve_stats_table.h>
11 : #include <sandesh/common/flow_types.h>
12 : #include <init/agent_param.h>
13 : #include <vrouter/flow_stats/session_stats_collector.h>
14 : #include <vrouter/ksync/ksync_init.h>
15 : #include <oper/tag.h>
16 : #include <oper/security_logging_object.h>
17 : #include <oper/global_vrouter.h>
18 : #include <vrouter/flow_stats/flow_stats_collector.h>
19 : #include <vrouter/flow_stats/flow_stats_types.h>
20 : #include <cmn/agent_factory.h>
21 :
22 : // setting work queue max size as 4M
23 : #define DEFAULT_SSC_REQUEST_QUEUE_SIZE 4*1024*1024
24 : #define MAX_SSC_REQUEST_QUEUE_ITERATIONS 256
25 :
26 : SandeshTraceBufferPtr SessionStatsTraceBuf(SandeshTraceBufferCreate(
27 : "SessionStats", 4000));
28 :
29 : bool session_debug_ = false;
30 2 : SessionStatsCollector::SessionStatsCollector(boost::asio::io_context &io,
31 : AgentUveBase *uve,
32 : uint32_t instance_id,
33 : FlowStatsManager *aging_module,
34 2 : SessionStatsCollectorObject *obj) :
35 : StatsCollector(TaskScheduler::GetInstance()->GetTaskId
36 : (kTaskSessionStatsCollector), instance_id,
37 : io, kSessionStatsTimerInterval, "Session stats collector"),
38 2 : agent_uve_(uve),
39 2 : task_id_(uve->agent()->task_scheduler()->GetTaskId
40 2 : (kTaskSessionStatsCollector)),
41 2 : session_ep_iteration_key_(), session_agg_iteration_key_(),
42 2 : session_iteration_key_(),
43 2 : request_queue_(agent_uve_->agent()->task_scheduler()->
44 : GetTaskId(kTaskSessionStatsCollectorEvent),
45 : instance_id,
46 : boost::bind(&SessionStatsCollector::RequestHandler,
47 : this, _1),
48 : DEFAULT_SSC_REQUEST_QUEUE_SIZE,
49 : MAX_SSC_REQUEST_QUEUE_ITERATIONS ),
50 4 : session_msg_list_(agent_uve_->agent()->params()->max_endpoints_per_session_msg(),
51 4 : SessionEndpoint()),
52 2 : session_msg_index_(0), instance_id_(instance_id),
53 2 : flow_stats_manager_(aging_module), parent_(obj), session_task_(NULL),
54 6 : current_time_(GetCurrentTime()), session_task_starts_(0) {
55 2 : request_queue_.set_name("Session stats collector event queue");
56 : request_queue_.set_measure_busy_time
57 2 : (agent_uve_->agent()->MeasureQueueDelay());
58 : request_queue_.SetEntryCallback
59 2 : (boost::bind(&SessionStatsCollector::RequestHandlerEntry, this));
60 : request_queue_.SetExitCallback
61 2 : (boost::bind(&SessionStatsCollector::RequestHandlerExit, this, _1));
62 2 : request_queue_.SetBounded(true);
63 2 : InitDone();
64 2 : }
65 :
66 4 : SessionStatsCollector::~SessionStatsCollector() {
67 2 : flow_stats_manager_->FreeIndex(instance_id_);
68 4 : }
69 :
70 555 : uint64_t SessionStatsCollector::GetCurrentTime() {
71 555 : return UTCTimestampUsec();
72 : }
73 :
74 2 : void SessionStatsCollector::Shutdown() {
75 2 : StatsCollector::Shutdown();
76 2 : request_queue_.Shutdown();
77 2 : }
78 :
79 2 : void SessionStatsCollector::RegisterDBClients() {
80 2 : if (agent_uve_->agent()->slo_table()) {
81 2 : slo_listener_id_ = agent_uve_->agent()->slo_table()->Register(
82 : boost::bind(&SessionStatsCollector::SloNotify, this,
83 : _1, _2));
84 : }
85 2 : }
86 :
87 6 : bool SessionStatsCollector::Run() {
88 6 : if (session_endpoint_map_.size() == 0) {
89 0 : return true;
90 : }
91 :
92 : // Start task to scan the entries
93 6 : if (session_task_ == NULL) {
94 6 : session_task_starts_++;
95 :
96 6 : if (session_debug_) {
97 0 : LOG(DEBUG,
98 : UTCUsecToString(ClockMonotonicUsec())
99 : << " SessionTasks Num " << session_task_starts_
100 : << " session_ep visited " << session_ep_visited_
101 : << " Request count " << request_queue_.Length());
102 : }
103 6 : session_ep_visited_ = 0;
104 6 : session_task_ = new SessionTask(this);
105 6 : agent_uve_->agent()->task_scheduler()->Enqueue(session_task_);
106 : }
107 6 : return true;
108 : }
109 :
110 :
111 3 : bool FlowStatsManager::UpdateSessionThreshold() {
112 3 : uint64_t curr_time = FlowStatsCollector::GetCurrentTime();
113 3 : bool export_rate_calculated = false;
114 3 : uint32_t exp_rate_without_sampling = 0;
115 :
116 : /* If flows are not being exported, no need to update threshold */
117 3 : if (!session_export_count_) {
118 3 : return true;
119 : }
120 :
121 : // Calculate Flow Export rate
122 0 : if (prev_flow_export_rate_compute_time_) {
123 0 : uint64_t diff_secs = 0;
124 0 : uint64_t diff_micro_secs = curr_time -
125 0 : prev_flow_export_rate_compute_time_;
126 0 : if (diff_micro_secs) {
127 0 : diff_secs = diff_micro_secs/1000000;
128 : }
129 0 : if (diff_secs) {
130 0 : uint32_t session_export_count = session_export_count_reset();
131 0 : session_export_rate_ = session_export_count/diff_secs;
132 0 : exp_rate_without_sampling =
133 0 : session_export_without_sampling_reset()/diff_secs;
134 0 : prev_flow_export_rate_compute_time_ = curr_time;
135 0 : export_rate_calculated = true;
136 : }
137 : } else {
138 0 : prev_flow_export_rate_compute_time_ = curr_time;
139 0 : session_export_count_ = 0;
140 0 : return true;
141 : }
142 :
143 0 : uint32_t cfg_rate = agent_->oper_db()->global_vrouter()->
144 0 : flow_export_rate();
145 : /* No need to update threshold when flow_export_rate is NOT calculated
146 : * and configured flow export rate has not changed */
147 0 : if (!export_rate_calculated &&
148 0 : (cfg_rate == prev_cfg_flow_export_rate_)) {
149 0 : return true;
150 : }
151 0 : uint64_t cur_t = threshold(), new_t = 0;
152 : // Update sampling threshold based on flow_export_rate_
153 0 : if (session_export_rate_ < ((double)cfg_rate) * 0.8) {
154 : /* There are two reasons why we can be here.
155 : * 1. None of the flows were sampled because we never crossed
156 : * 80% of configured flow-export-rate.
157 : * 2. In scale setups, the threshold was updated to high value because
158 : * of which flow-export-rate has dropped drastically.
159 : * Threshold should be updated here depending on which of the above two
160 : * situations we are in. */
161 0 : if (!sessions_sampled_atleast_once_) {
162 0 : UpdateThreshold(kDefaultFlowSamplingThreshold, false);
163 : } else {
164 0 : if (session_export_rate_ < ((double)cfg_rate) * 0.5) {
165 0 : UpdateThreshold((threshold_ / 4), false);
166 : } else {
167 0 : UpdateThreshold((threshold_ / 2), false);
168 : }
169 : }
170 0 : } else if (session_export_rate_ > (cfg_rate * 3)) {
171 0 : UpdateThreshold((threshold_ * 4), true);
172 0 : } else if (session_export_rate_ > (cfg_rate * 2)) {
173 0 : UpdateThreshold((threshold_ * 3), true);
174 0 : } else if (session_export_rate_ > ((double)cfg_rate) * 1.25) {
175 0 : UpdateThreshold((threshold_ * 2), true);
176 : }
177 0 : prev_cfg_flow_export_rate_ = cfg_rate;
178 0 : new_t = threshold();
179 0 : FLOW_EXPORT_STATS_TRACE(session_export_rate_, exp_rate_without_sampling,
180 : cur_t, new_t);
181 0 : return true;
182 : }
183 :
184 : /////////////////////////////////////////////////////////////////////////////
185 : // Utility methods to enqueue events into work-queue
186 : /////////////////////////////////////////////////////////////////////////////
187 230 : void SessionStatsCollector::AddEvent(const FlowEntryPtr &flow) {
188 : boost::shared_ptr<SessionStatsReq>
189 : req(new SessionStatsReq(SessionStatsReq::ADD_SESSION, flow,
190 230 : GetCurrentTime()));
191 230 : request_queue_.Enqueue(req);
192 230 : }
193 :
194 144 : void SessionStatsCollector::DeleteEvent(const FlowEntryPtr &flow,
195 : const RevFlowDepParams ¶ms) {
196 : boost::shared_ptr<SessionStatsReq>
197 : req(new SessionStatsReq(SessionStatsReq::DELETE_SESSION, flow,
198 144 : GetCurrentTime(), params));
199 144 : request_queue_.Enqueue(req);
200 144 : }
201 :
202 0 : void SessionStatsCollector::UpdateSessionStatsEvent(const FlowEntryPtr &flow,
203 : uint32_t bytes,
204 : uint32_t packets,
205 : uint32_t oflow_bytes,
206 : const boost::uuids::uuid &u) {
207 : boost::shared_ptr<SessionStatsReq>
208 : req(new SessionStatsReq(SessionStatsReq::UPDATE_SESSION_STATS, flow,
209 0 : bytes, packets, oflow_bytes, u));
210 0 : request_queue_.Enqueue(req);
211 0 : }
212 :
213 0 : void SessionStatsCollector::DispatchSessionMsg(const std::vector<SessionEndpoint> &lst) {
214 0 : flow_stats_manager_->UpdateSessionMsgExportStats(1);
215 0 : flow_stats_manager_->UpdateSessionSampleExportStats(lst.size());
216 0 : SESSION_ENDPOINT_OBJECT_LOG("", SandeshLevel::SYS_INFO, lst);
217 0 : }
218 :
219 0 : void SessionStatsCollector::EnqueueSessionMsg() {
220 0 : session_msg_index_++;
221 0 : if (session_msg_index_ ==
222 0 : agent_uve_->agent()->params()->max_endpoints_per_session_msg()) {
223 0 : DispatchSessionMsg(session_msg_list_);
224 0 : session_msg_index_ = 0;
225 : }
226 0 : }
227 :
228 6 : void SessionStatsCollector::DispatchPendingSessionMsg() {
229 6 : if (session_msg_index_ == 0) {
230 6 : return;
231 : }
232 :
233 0 : vector<SessionEndpoint>::const_iterator first = session_msg_list_.begin();
234 0 : vector<SessionEndpoint>::const_iterator last = session_msg_list_.begin() +
235 0 : session_msg_index_;
236 0 : vector<SessionEndpoint> new_list(first, last);
237 0 : DispatchSessionMsg(new_list);
238 0 : session_msg_index_ = 0;
239 0 : }
240 :
241 26 : uint8_t SessionStatsCollector::GetSessionMsgIdx() {
242 26 : SessionEndpoint &obj = session_msg_list_[session_msg_index_];
243 26 : obj = SessionEndpoint();
244 26 : return session_msg_index_;
245 : }
246 :
247 155 : bool SessionStatsCollector::RequestHandlerEntry() {
248 155 : current_time_ = GetCurrentTime();
249 155 : return true;
250 : }
251 :
252 155 : void SessionStatsCollector::RequestHandlerExit(bool done) {
253 155 : }
254 :
255 374 : bool SessionStatsCollector::RequestHandler(boost::shared_ptr<SessionStatsReq> req) {
256 374 : FlowEntry *flow = req->flow();
257 374 : FlowEntry *rflow = req->reverse_flow();
258 518 : FLOW_LOCK(flow, rflow, FlowEvent::FLOW_MESSAGE);
259 :
260 374 : switch (req->event()) {
261 230 : case SessionStatsReq::ADD_SESSION: {
262 230 : AddSession(flow, req->time());
263 230 : break;
264 : }
265 :
266 144 : case SessionStatsReq::DELETE_SESSION: {
267 144 : DeleteSession(flow, flow->uuid(), req->time(), &req->params());
268 144 : break;
269 : }
270 :
271 0 : case SessionStatsReq::UPDATE_SESSION_STATS: {
272 0 : EvictedSessionStatsUpdate(req->flow(), req->bytes(), req->packets(),
273 0 : req->oflow_bytes(), req->uuid());
274 0 : break;
275 : }
276 :
277 0 : default:
278 0 : assert(0);
279 : }
280 :
281 374 : return true;
282 374 : }
283 :
284 1278 : bool SessionEndpointKey::IsLess(const SessionEndpointKey &rhs) const {
285 1278 : if (vmi_cfg_name != rhs.vmi_cfg_name) {
286 289 : return vmi_cfg_name < rhs.vmi_cfg_name;
287 : }
288 989 : if (local_vn != rhs.local_vn) {
289 28 : return local_vn < rhs.local_vn;
290 : }
291 961 : if (remote_vn != rhs.remote_vn) {
292 145 : return remote_vn < rhs.remote_vn;
293 : }
294 816 : if (local_tagset != rhs.local_tagset) {
295 0 : return local_tagset < rhs.local_tagset;
296 : }
297 816 : if (remote_tagset != rhs.remote_tagset) {
298 0 : return remote_tagset < rhs.remote_tagset;
299 : }
300 816 : if (remote_prefix != rhs.remote_prefix) {
301 7 : return remote_prefix < rhs.remote_prefix;
302 : }
303 809 : if (match_policy != rhs.match_policy) {
304 109 : return match_policy < rhs.match_policy;
305 : }
306 700 : if (is_client_session != rhs.is_client_session) {
307 116 : return is_client_session < rhs.is_client_session;
308 : }
309 584 : return is_si < rhs.is_si;
310 : }
311 :
312 116 : bool SessionEndpointKey::IsEqual(const SessionEndpointKey &rhs) const {
313 116 : if (vmi_cfg_name != rhs.vmi_cfg_name) {
314 0 : return false;
315 : }
316 116 : if (local_vn != rhs.local_vn) {
317 2 : return false;
318 : }
319 114 : if (remote_vn != rhs.remote_vn) {
320 5 : return false;
321 : }
322 109 : if (local_tagset != rhs.local_tagset) {
323 0 : return false;
324 : }
325 109 : if (remote_tagset != rhs.remote_tagset) {
326 0 : return false;
327 : }
328 109 : if (remote_prefix != rhs.remote_prefix) {
329 0 : return false;
330 : }
331 109 : if (match_policy != rhs.match_policy) {
332 14 : return false;
333 : }
334 95 : if (is_client_session != rhs.is_client_session) {
335 0 : return false;
336 : }
337 95 : if (is_si != rhs.is_si) {
338 0 : return false;
339 : }
340 95 : return true;
341 : }
342 :
343 406 : void SessionEndpointKey::Reset() {
344 406 : vmi_cfg_name = "";
345 406 : local_vn = "";
346 406 : remote_vn = "";
347 406 : local_tagset.clear();
348 406 : remote_tagset.clear();
349 406 : remote_prefix = "";
350 406 : match_policy = "";
351 406 : is_client_session = false;
352 406 : is_si = false;
353 406 : }
354 :
355 1198 : bool SessionAggKey::IsLess(const SessionAggKey &rhs) const {
356 1198 : if (local_ip != rhs.local_ip) {
357 97 : return local_ip < rhs.local_ip;
358 : }
359 1101 : if (server_port != rhs.server_port) {
360 610 : return server_port < rhs.server_port;
361 : }
362 491 : return proto < rhs.proto;
363 : }
364 :
365 116 : bool SessionAggKey::IsEqual(const SessionAggKey &rhs) const {
366 116 : if (local_ip != rhs.local_ip) {
367 0 : return false;
368 : }
369 116 : if (server_port != rhs.server_port) {
370 0 : return false;
371 : }
372 116 : if (proto != rhs.proto) {
373 0 : return false;
374 : }
375 116 : return true;
376 : }
377 :
378 423 : void SessionAggKey::Reset() {
379 423 : local_ip = IpAddress();
380 423 : server_port = 0;
381 423 : proto = 0;
382 423 : }
383 :
384 565 : bool SessionKey::IsLess(const SessionKey &rhs) const {
385 565 : if (remote_ip != rhs.remote_ip) {
386 59 : return remote_ip < rhs.remote_ip;
387 : }
388 506 : if (client_port != rhs.client_port) {
389 0 : return client_port < rhs.client_port;
390 : }
391 506 : return uuid < rhs.uuid;
392 : }
393 :
394 119 : bool SessionKey::IsEqual(const SessionKey &rhs) const {
395 119 : if (remote_ip != rhs.remote_ip) {
396 0 : return false;
397 : }
398 119 : if (client_port != rhs.client_port) {
399 0 : return false;
400 : }
401 119 : if (uuid != rhs.uuid) {
402 3 : return false;
403 : }
404 116 : return true;
405 : }
406 :
407 478 : void SessionKey::Reset() {
408 478 : remote_ip = IpAddress();
409 478 : client_port = 0;
410 478 : uuid = boost::uuids::nil_uuid();
411 478 : }
412 :
413 230 : bool SessionStatsCollector::GetSessionKey(FlowEntry* fe,
414 : SessionAggKey &session_agg_key,
415 : SessionKey &session_key,
416 : SessionEndpointKey &session_endpoint_key) {
417 : /*
418 : * For non local flows always get the key for forward flow entry
419 : * If vms are in same compute node (local route)
420 : * Vrouter A (Client) <==> Vrouter B (Server)
421 : * A->B (Ingress + Forwarding + Local)
422 : * B->A (Ingress + Reverse + Local)
423 : *
424 : * If vms are in different compute nodes
425 : * Vrouter A (Client) <==> Vrouter B (Server)
426 : * A->B (Ingress + Forwarding) A->B (Egresss + Forward)
427 : * B->A (Egress + Reverse) B->A (Ingress + Reverse)
428 : */
429 :
430 : /*
431 : * If it is non local reverse flow then NOP, actual
432 : * config will be handled in fwd flow
433 : */
434 410 : if ((!(fe->is_flags_set(FlowEntry::LocalFlow))) &&
435 410 : (fe->is_flags_set(FlowEntry::ReverseFlow))) {
436 0 : return false;
437 : }
438 :
439 230 : const Interface *itf = fe->intf_entry();
440 230 : if (!itf) {
441 2 : return false;
442 : }
443 :
444 228 : if (itf->type() != Interface::VM_INTERFACE) {
445 2 : return false;
446 : }
447 :
448 226 : const VmInterface *vmi = static_cast<const VmInterface *>(itf);
449 226 : if (vmi->cfg_name().empty()) {
450 0 : return false;
451 : }
452 226 : const string &src_vn = !fe->data().origin_vn_src.empty() ?
453 18 : fe->data().origin_vn_src :
454 244 : fe->data().source_vn_match;
455 226 : const string &dst_vn = !fe->data().origin_vn_dst.empty() ?
456 10 : fe->data().origin_vn_dst :
457 236 : fe->data().dest_vn_match;
458 :
459 226 : session_endpoint_key.vmi_cfg_name = vmi->cfg_name();
460 226 : session_endpoint_key.local_tagset = fe->local_tagset();
461 226 : session_endpoint_key.remote_tagset = fe->remote_tagset();
462 226 : session_endpoint_key.remote_prefix = fe->RemotePrefix();
463 226 : session_endpoint_key.match_policy = fe->fw_policy_name_uuid();
464 226 : if (vmi->service_intf_type().empty()) {
465 226 : session_endpoint_key.is_si = false;
466 : } else {
467 0 : session_endpoint_key.is_si= true;
468 : }
469 :
470 226 : if (fe->IsClientFlow()) {
471 63 : session_agg_key.local_ip = fe->key().src_addr;
472 63 : session_agg_key.server_port = fe->key().dst_port;
473 63 : session_key.remote_ip = fe->key().dst_addr;
474 63 : session_key.client_port = fe->key().src_port;
475 63 : session_endpoint_key.local_vn = src_vn;
476 63 : session_endpoint_key.remote_vn = dst_vn;
477 63 : session_endpoint_key.is_client_session = true;
478 163 : } else if (fe->IsServerFlow()) {
479 : /*
480 : * If it is local flow, then reverse flow
481 : * (Ingress + Reverse) will be used to create
482 : * the server side session (Egress + Forward)
483 : */
484 163 : if (fe->is_flags_set(FlowEntry::LocalFlow)) {
485 25 : session_agg_key.local_ip = fe->key().src_addr;
486 25 : session_agg_key.server_port = fe->key().src_port;
487 25 : session_key.remote_ip = fe->key().dst_addr;
488 25 : session_key.client_port = fe->key().dst_port;
489 25 : session_endpoint_key.local_vn = src_vn;
490 25 : session_endpoint_key.remote_vn = dst_vn;
491 : } else {
492 138 : session_agg_key.local_ip = fe->key().dst_addr;
493 138 : session_agg_key.server_port = fe->key().dst_port;
494 138 : session_key.remote_ip = fe->key().src_addr;
495 138 : session_key.client_port = fe->key().src_port;
496 138 : session_endpoint_key.local_vn = dst_vn;
497 138 : session_endpoint_key.remote_vn = src_vn;
498 : }
499 163 : session_endpoint_key.is_client_session = false;
500 : } else {
501 0 : return false;
502 : }
503 226 : session_agg_key.proto = fe->key().protocol;
504 226 : session_key.uuid = fe->uuid();
505 226 : return true;
506 : }
507 :
508 452 : void SessionStatsCollector::UpdateSessionFlowStatsInfo(FlowEntry *fe,
509 : SessionFlowStatsInfo *session_flow) const {
510 452 : session_flow->flow = fe;
511 452 : session_flow->gen_id= fe->gen_id();
512 452 : session_flow->flow_handle = fe->flow_handle();
513 452 : session_flow->uuid = fe->uuid();
514 452 : session_flow->total_bytes = 0;
515 452 : session_flow->total_packets = 0;
516 452 : }
517 :
518 226 : void SessionStatsCollector::UpdateSessionStatsInfo(FlowEntry* fe,
519 : uint64_t setup_time, SessionStatsInfo *session) const {
520 226 : FlowEntry *rfe = fe->reverse_flow_entry();
521 :
522 226 : session->setup_time = setup_time;
523 226 : session->teardown_time = 0;
524 226 : session->exported_atleast_once = false;
525 226 : UpdateSessionFlowStatsInfo(fe, &session->fwd_flow);
526 226 : UpdateSessionFlowStatsInfo(rfe, &session->rev_flow);
527 226 : }
528 :
529 654 : static void BuildTraceTagList(const TagList &slist, vector<string> *dlist) {
530 654 : TagList::const_iterator it = slist.begin();
531 654 : while (it != slist.end()) {
532 0 : dlist->push_back(integerToString(*it));
533 0 : ++it;
534 : }
535 654 : }
536 :
537 327 : static void TraceSession(const string &op, const SessionEndpointKey &ep,
538 : const SessionAggKey &agg, const SessionKey &session,
539 : bool rev_flow_params) {
540 327 : SessionTraceInfo info;
541 327 : info.vmi = ep.vmi_cfg_name;
542 327 : info.local_vn = ep.local_vn;
543 327 : info.remote_vn = ep.remote_vn;
544 327 : BuildTraceTagList(ep.local_tagset, &info.local_tagset);
545 327 : BuildTraceTagList(ep.remote_tagset, &info.remote_tagset);
546 327 : info.remote_prefix = ep.remote_prefix;
547 327 : info.match_policy = ep.match_policy;
548 327 : info.is_si = ep.is_si;
549 327 : info.is_client = ep.is_client_session;
550 327 : info.local_ip = agg.local_ip.to_string();
551 327 : info.server_port = agg.server_port;
552 327 : info.protocol = agg.proto;
553 327 : info.remote_ip = session.remote_ip.to_string();
554 327 : info.client_port = session.client_port;
555 327 : info.flow_uuid = to_string(session.uuid);
556 327 : SESSION_STATS_TRACE(Trace, op, info, rev_flow_params);
557 327 : }
558 :
559 230 : void SessionStatsCollector::AddSession(FlowEntry* fe, uint64_t setup_time) {
560 230 : SessionAggKey session_agg_key;
561 230 : SessionEndpointInfo::SessionAggMap::iterator session_agg_map_iter;
562 230 : SessionPreAggInfo session_agg_info;
563 230 : SessionStatsInfo session;
564 230 : SessionKey session_key;
565 230 : SessionPreAggInfo::SessionMap::iterator session_map_iter;
566 230 : SessionEndpointInfo session_endpoint_info = {};
567 230 : SessionEndpointKey session_endpoint_key;
568 230 : SessionEndpointMap::iterator session_endpoint_map_iter;
569 230 : FlowEntry *fe_fwd = fe;
570 : bool success;
571 :
572 230 : if (NULL == fe->reverse_flow_entry()) {
573 0 : return;
574 : }
575 :
576 230 : if (!(fe->is_flags_set(FlowEntry::LocalFlow))) {
577 180 : if (fe->is_flags_set(FlowEntry::ReverseFlow)) {
578 90 : fe_fwd = fe->reverse_flow_entry();
579 : }
580 : }
581 :
582 230 : success = GetSessionKey(fe_fwd, session_agg_key, session_key,
583 : session_endpoint_key);
584 230 : if (!success) {
585 4 : return;
586 : }
587 :
588 : /*
589 : * If the flow is part of session endpoint DB then
590 : * delete the existing one
591 : * Flow add comes for the existing flow for the following cases
592 : * - flow uuid changes (add-delete-add : will be compressed)
593 : * - any key changes from session endpoint, aggregate or session
594 : */
595 226 : FlowSessionMap::iterator flow_session_map_iter;
596 226 : flow_session_map_iter = flow_session_map_.find(fe_fwd);
597 226 : if (flow_session_map_iter != flow_session_map_.end()) {
598 :
599 119 : FlowToSessionMap &flow_to_session_map = flow_session_map_iter->second;
600 : FlowToSessionMap rhs_flow_to_session_map(session_key,
601 : session_agg_key,
602 119 : session_endpoint_key);
603 119 : if (!(flow_to_session_map.IsEqual(rhs_flow_to_session_map))) {
604 24 : DeleteSession(fe_fwd, flow_to_session_map.session_key().uuid,
605 : GetCurrentTime(), NULL);
606 : }
607 119 : }
608 :
609 226 : TraceSession("Add", session_endpoint_key, session_agg_key, session_key,
610 : false);
611 226 : UpdateSessionStatsInfo(fe_fwd, setup_time, &session);
612 :
613 226 : session_endpoint_map_iter = session_endpoint_map_.find(
614 : session_endpoint_key);
615 226 : if (session_endpoint_map_iter == session_endpoint_map_.end()) {
616 35 : session_agg_info.session_map_.insert(make_pair(session_key, session));
617 35 : session_endpoint_info.session_agg_map_.insert(
618 70 : make_pair(session_agg_key, session_agg_info));
619 35 : session_endpoint_map_.insert(make_pair(session_endpoint_key,
620 : session_endpoint_info));
621 35 : AddFlowToSessionMap(fe_fwd, session_key, session_agg_key,
622 : session_endpoint_key);
623 : } else {
624 191 : session_agg_map_iter = session_endpoint_map_iter->
625 191 : second.session_agg_map_.find(
626 : session_agg_key);
627 191 : if (session_agg_map_iter ==
628 382 : session_endpoint_map_iter->second.session_agg_map_.end()) {
629 56 : session_agg_info.session_map_.insert(make_pair(session_key, session));
630 112 : session_endpoint_map_iter->second.session_agg_map_.insert(
631 112 : make_pair(session_agg_key, session_agg_info));
632 56 : AddFlowToSessionMap(fe_fwd, session_key, session_agg_key,
633 : session_endpoint_key);
634 : } else {
635 : session_map_iter =
636 135 : session_agg_map_iter->second.session_map_.find(session_key);
637 135 : if (session_map_iter ==
638 270 : session_agg_map_iter->second.session_map_.end()) {
639 20 : session_agg_map_iter->second.session_map_.insert(
640 20 : make_pair(session_key, session));
641 10 : AddFlowToSessionMap(fe_fwd, session_key, session_agg_key,
642 : session_endpoint_key);
643 : } else {
644 : /*
645 : * existing flow should match with the incoming add flow
646 : */
647 125 : assert(session.fwd_flow.uuid == fe_fwd->uuid());
648 : }
649 : }
650 : }
651 242 : }
652 :
653 168 : void SessionStatsCollector::DeleteSession(FlowEntry* fe,
654 : const boost::uuids::uuid &del_uuid,
655 : uint64_t teardown_time,
656 : const RevFlowDepParams *params) {
657 168 : SessionAggKey session_agg_key;
658 168 : SessionEndpointInfo::SessionAggMap::iterator session_agg_map_iter;
659 168 : SessionPreAggInfo session_agg_info;
660 168 : SessionKey session_key;
661 168 : SessionPreAggInfo::SessionMap::iterator session_map_iter;
662 168 : SessionEndpointInfo session_endpoint_info;
663 168 : SessionEndpointKey session_endpoint_key;
664 168 : SessionEndpointMap::iterator session_endpoint_map_iter;
665 168 : bool read_flow = true;
666 :
667 168 : if (del_uuid != fe->uuid()) {
668 3 : read_flow = false;
669 : }
670 :
671 : /*
672 : * If the given flow uuid is different from the existing one
673 : * then ignore the read from flow
674 : */
675 168 : FlowSessionMap::iterator flow_session_map_iter;
676 :
677 168 : flow_session_map_iter = flow_session_map_.find(fe);
678 168 : if (flow_session_map_iter != flow_session_map_.end()) {
679 101 : if (del_uuid != flow_session_map_iter->second.session_key().uuid) {
680 : /* We had never seen ADD for del_uuid, ignore delete request. This
681 : * can happen when the following events occur
682 : * 1. Add with UUID x
683 : * 2. Delete with UUID x
684 : * 3. Add with UUID y
685 : * 4. Delete with UUID y
686 : * When events 2 and 3 are suppressed and we receive only 1 and 4
687 : * In this case initiated delete for entry with UUID x */
688 2 : read_flow = false;
689 : /* Reset params because they correspond to entry with UUID y */
690 2 : params = NULL;
691 : }
692 101 : session_endpoint_key = flow_session_map_iter->second.session_endpoint_key();
693 101 : session_agg_key = flow_session_map_iter->second.session_agg_key();
694 101 : session_key = flow_session_map_iter->second.session_key();
695 : } else {
696 67 : return;
697 : }
698 101 : if (params && params->action_info_.action == 0) {
699 19 : params = NULL;
700 : }
701 :
702 101 : bool params_valid = true;
703 101 : if (params == NULL) {
704 45 : params_valid = false;
705 : }
706 :
707 101 : TraceSession("Del", session_endpoint_key, session_agg_key, session_key,
708 : params_valid);
709 101 : session_endpoint_map_iter = session_endpoint_map_.find(
710 : session_endpoint_key);
711 101 : if (session_endpoint_map_iter != session_endpoint_map_.end()) {
712 101 : session_agg_map_iter = session_endpoint_map_iter->
713 101 : second.session_agg_map_.find(
714 : session_agg_key);
715 101 : if (session_agg_map_iter !=
716 202 : session_endpoint_map_iter->second.session_agg_map_.end()) {
717 : session_map_iter =
718 101 : session_agg_map_iter->second.session_map_.find(session_key);
719 101 : if (session_map_iter !=
720 202 : session_agg_map_iter->second.session_map_.end()) {
721 : /*
722 : * Process the stats collector
723 : */
724 101 : session_map_iter->second.teardown_time = teardown_time;
725 101 : session_map_iter->second.deleted = true;
726 : /* Don't read stats for evicted flow, during delete */
727 101 : if (!session_map_iter->second.evicted) {
728 101 : SessionStatsChangedUnlocked(session_map_iter,
729 101 : &session_map_iter->second.del_stats);
730 : }
731 101 : if (read_flow) {
732 96 : CopyFlowInfo(session_map_iter->second, params);
733 : }
734 :
735 101 : assert(session_map_iter->second.fwd_flow.flow.get() == fe);
736 101 : DeleteFlowToSessionMap(fe);
737 101 : session_map_iter->second.fwd_flow.flow = NULL;
738 101 : session_map_iter->second.rev_flow.flow = NULL;
739 : }
740 : }
741 : }
742 302 : }
743 :
744 0 : void SessionStatsCollector::EvictedSessionStatsUpdate(const FlowEntryPtr &flow,
745 : uint32_t bytes,
746 : uint32_t packets,
747 : uint32_t oflow_bytes,
748 : const boost::uuids::uuid &u) {
749 0 : FlowSessionMap::iterator flow_session_map_iter;
750 0 : SessionInfo session_info;
751 0 : SessionIpPort session_key;
752 0 : SessionAggInfo session_agg_info;
753 0 : SessionIpPortProtocol session_agg_key;
754 0 : SessionEndpointMap::iterator session_ep_map_iter;
755 0 : SessionEndpointInfo::SessionAggMap::iterator session_agg_map_iter;
756 0 : SessionPreAggInfo::SessionMap::iterator session_map_iter;
757 :
758 : /* TODO: Evicted msg coming for reverse flow. We currently don't have
759 : * mapping from reverse_flow to flow_session_map */
760 0 : flow_session_map_iter = flow_session_map_.find(flow.get());
761 0 : if (flow_session_map_iter == flow_session_map_.end()) {
762 0 : return;
763 : }
764 :
765 0 : if (flow_session_map_iter->second.session_key().uuid != u) {
766 0 : return;
767 : }
768 :
769 0 : SessionEndpointKey session_db_ep_key = flow_session_map_iter->second.session_endpoint_key();
770 0 : SessionAggKey session_db_agg_key = flow_session_map_iter->second.session_agg_key();
771 0 : SessionKey session_db_key = flow_session_map_iter->second.session_key();
772 :
773 0 : session_ep_map_iter = session_endpoint_map_.find(session_db_ep_key);
774 0 : if (session_ep_map_iter != session_endpoint_map_.end()) {
775 0 : session_agg_map_iter = session_ep_map_iter->
776 0 : second.session_agg_map_.find(
777 : session_db_agg_key);
778 0 : if (session_agg_map_iter !=
779 0 : session_ep_map_iter->second.session_agg_map_.end()) {
780 : session_map_iter =
781 0 : session_agg_map_iter->second.session_map_.find(session_db_key);
782 0 : if (session_map_iter !=
783 0 : session_agg_map_iter->second.session_map_.end()) {
784 : /*
785 : * update the latest statistics
786 : */
787 0 : SessionFlowStatsInfo &session_flow = session_map_iter->second.fwd_flow;
788 0 : uint64_t k_bytes, total_bytes, diff_bytes = 0;
789 0 : uint64_t k_packets, total_packets, diff_packets = 0;
790 0 : k_bytes = FlowStatsCollector::GetFlowStats((oflow_bytes & 0xFFFF),
791 : bytes);
792 0 : k_packets = FlowStatsCollector::GetFlowStats((oflow_bytes & 0xFFFF0000),
793 : packets);
794 0 : total_bytes = GetUpdatedSessionFlowBytes(session_flow.total_bytes,
795 : k_bytes);
796 0 : total_packets = GetUpdatedSessionFlowPackets(session_flow.total_packets,
797 : k_packets);
798 0 : diff_bytes = total_bytes - session_flow.total_bytes;
799 0 : diff_packets = total_packets - session_flow.total_packets;
800 0 : session_flow.total_bytes = total_bytes;
801 0 : session_flow.total_packets = total_packets;
802 :
803 0 : SessionStatsParams &estats = session_map_iter->second.
804 0 : evict_stats;
805 0 : session_map_iter->second.evicted = true;
806 0 : estats.fwd_flow.valid = true;
807 0 : estats.fwd_flow.diff_bytes = diff_bytes;
808 0 : estats.fwd_flow.diff_packets = diff_packets;
809 : }
810 : }
811 : }
812 0 : }
813 :
814 101 : void SessionStatsCollector::AddFlowToSessionMap(FlowEntry *fe,
815 : SessionKey session_key,
816 : SessionAggKey session_agg_key,
817 : SessionEndpointKey session_endpoint_key) {
818 : FlowToSessionMap flow_to_session_map(session_key, session_agg_key,
819 101 : session_endpoint_key);
820 : std::pair<FlowSessionMap::iterator, bool> ret =
821 101 : flow_session_map_.insert(make_pair(fe, flow_to_session_map));
822 101 : if (ret.second == false) {
823 0 : FlowToSessionMap &prev = ret.first->second;
824 0 : assert(prev.session_key().uuid == fe->uuid());
825 : }
826 101 : }
827 :
828 101 : void SessionStatsCollector::DeleteFlowToSessionMap(FlowEntry *fe) {
829 101 : FlowSessionMap::iterator flow_session_map_iter;
830 101 : flow_session_map_iter = flow_session_map_.find(fe);
831 101 : if (flow_session_map_iter != flow_session_map_.end()) {
832 101 : flow_session_map_.erase(flow_session_map_iter);
833 : }
834 101 : }
835 :
836 0 : int SessionStatsCollector::ComputeSloRate(int rate, SecurityLoggingObject *slo)
837 : const {
838 : /* If rate is not configured, it will have -1 as value
839 : * -1 as value. In this case, pick the rate from SLO */
840 0 : if (rate == -1) {
841 0 : rate = slo->rate();
842 : }
843 0 : return rate;
844 : }
845 :
846 0 : void SessionStatsCollector::UpdateSloStateRules(SecurityLoggingObject *slo,
847 : SessionSloState *state) {
848 0 : vector<autogen::SecurityLoggingObjectRuleEntryType>::const_iterator it;
849 0 : it = slo->rules().begin();
850 0 : while (it != slo->rules().end()) {
851 0 : state->UpdateSessionSloStateRuleEntry(it->rule_uuid, it->rate);
852 0 : it++;
853 : }
854 :
855 0 : SloRuleList::const_iterator acl_it = slo->firewall_policy_list().begin();
856 0 : while (acl_it != slo->firewall_policy_list().end()) {
857 0 : AclKey key(acl_it->uuid_);
858 0 : AclDBEntry *acl = static_cast<AclDBEntry *>(agent_uve_->agent()->
859 0 : acl_table()->FindActiveEntry(&key));
860 0 : if (acl) {
861 0 : int index = 0;
862 0 : int rate = ComputeSloRate(acl_it->rate_, slo);
863 0 : const AclEntry *ae = acl->GetAclEntryAtIndex(index);
864 0 : while (ae != NULL) {
865 0 : state->UpdateSessionSloStateRuleEntry(ae->uuid(), rate);
866 0 : index++;
867 0 : ae = acl->GetAclEntryAtIndex(index);
868 : }
869 : }
870 0 : acl_it++;
871 0 : }
872 :
873 0 : SloRuleList::const_iterator fw_rule_it;
874 0 : fw_rule_it = slo->firewall_rule_list().begin();
875 0 : while (fw_rule_it != slo->firewall_rule_list().end()) {
876 0 : const SloRuleInfo &item = *fw_rule_it;
877 0 : int rate = ComputeSloRate(item.rate_, slo);
878 0 : state->UpdateSessionSloStateRuleEntry(to_string(item.uuid_), rate);
879 0 : fw_rule_it++;
880 : }
881 :
882 0 : }
883 :
884 0 : void SessionStatsCollector::SloNotify(DBTablePartBase *partition,
885 : DBEntryBase *e) {
886 0 : SecurityLoggingObject *slo = static_cast<SecurityLoggingObject *>(e);
887 : SessionSloState *state =
888 0 : static_cast<SessionSloState *>(slo->GetState(partition->parent(),
889 : slo_listener_id_));
890 0 : if (slo->IsDeleted()) {
891 0 : if (!state)
892 0 : return;
893 0 : slo->ClearState(partition->parent(), slo_listener_id_);
894 0 : delete state;
895 0 : return;
896 : }
897 :
898 0 : if (!state) {
899 0 : state = new SessionSloState();
900 0 : slo->SetState(slo->get_table(), slo_listener_id_, state);
901 : }
902 0 : UpdateSloStateRules(slo, state);
903 : }
904 :
905 0 : void SessionStatsCollector::AddSessionSloRuleEntry(const std::string &uuid,
906 : int rate,
907 : SecurityLoggingObject *slo,
908 : SessionSloRuleMap *slo_rule_map) {
909 0 : SessionSloRuleEntry slo_rule_entry(rate, slo->uuid());
910 0 : std::pair<SessionSloRuleMap::iterator, bool> ret;
911 0 : ret = slo_rule_map->insert(make_pair(uuid,
912 0 : slo_rule_entry));
913 0 : }
914 :
915 0 : void SessionStatsCollector::AddSloRules(
916 : const std::vector<autogen::SecurityLoggingObjectRuleEntryType> &list,
917 : SecurityLoggingObject *slo,
918 : SessionSloRuleMap *slo_rule_map) {
919 0 : vector<autogen::SecurityLoggingObjectRuleEntryType>::const_iterator it;
920 0 : it = list.begin();
921 0 : while (it != list.end()) {
922 0 : AddSessionSloRuleEntry(it->rule_uuid, it->rate, slo, slo_rule_map);
923 0 : it++;
924 : }
925 0 : }
926 :
927 0 : void SessionStatsCollector::AddSloFirewallPolicies(SecurityLoggingObject *slo,
928 : SessionSloRuleMap *r_map) {
929 0 : const SloRuleList &list = slo->firewall_policy_list();
930 0 : SloRuleList::const_iterator acl_it = list.begin();
931 0 : while (acl_it != list.end()) {
932 0 : AclKey key(acl_it->uuid_);
933 0 : AclDBEntry *acl = static_cast<AclDBEntry *>(agent_uve_->agent()->
934 0 : acl_table()->FindActiveEntry(&key));
935 0 : if (acl) {
936 0 : int index = 0;
937 0 : const AclEntry *ae = acl->GetAclEntryAtIndex(index);
938 0 : int rate = ComputeSloRate(acl_it->rate_, slo);
939 0 : while (ae != NULL) {
940 0 : AddSessionSloRuleEntry(ae->uuid(), rate, slo, r_map);
941 0 : index++;
942 0 : ae = acl->GetAclEntryAtIndex(index);
943 : }
944 : }
945 0 : acl_it++;
946 0 : }
947 0 : }
948 :
949 0 : void SessionStatsCollector::AddSloFirewallRules(SecurityLoggingObject *slo,
950 : SessionSloRuleMap *rule_map) {
951 0 : const SloRuleList &list = slo->firewall_rule_list();
952 0 : SloRuleList::const_iterator it = list.begin();
953 0 : while (it != list.end()) {
954 0 : int rate = ComputeSloRate(it->rate_, slo);
955 0 : AddSessionSloRuleEntry(to_string(it->uuid_), rate, slo, rule_map);
956 0 : it++;
957 : }
958 0 : }
959 :
960 0 : void SessionStatsCollector::AddSloEntryRules(SecurityLoggingObject *slo,
961 : SessionSloRuleMap *slo_rule_map) {
962 0 : AddSloRules(slo->rules(), slo, slo_rule_map);
963 0 : AddSloFirewallPolicies(slo, slo_rule_map);
964 0 : AddSloFirewallRules(slo, slo_rule_map);
965 0 : }
966 :
967 0 : void SessionStatsCollector::AddSloEntry(const boost::uuids::uuid &uuid,
968 : SessionSloRuleMap *slo_rule_map) {
969 0 : SecurityLoggingObjectKey slo_key(uuid);
970 0 : SecurityLoggingObject *slo = static_cast<SecurityLoggingObject *>
971 0 : (agent_uve_->agent()->slo_table()->FindActiveEntry(&slo_key));
972 0 : if (slo) {
973 0 : if (slo->status()) {
974 0 : AddSloEntryRules(slo, slo_rule_map);
975 : }
976 : }
977 0 : }
978 :
979 0 : void SessionStatsCollector::AddSloList(const UuidList &slo_list,
980 : SessionSloRuleMap *slo_rule_map) {
981 0 : UuidList::const_iterator sit = slo_list.begin();
982 0 : while (sit != slo_list.end()) {
983 0 : AddSloEntry(*sit, slo_rule_map);
984 0 : sit++;
985 : }
986 0 : }
987 :
988 0 : void SessionStatsCollector::MakeSloList(const FlowEntry *fe,
989 : SessionSloRuleMap *vmi_session_slo_rule_map,
990 : SessionSloRuleMap *vn_session_slo_rule_map) {
991 0 : if (fe == NULL) {
992 0 : return;
993 : }
994 0 : const Interface *itf = fe->intf_entry();
995 0 : if (!itf) {
996 0 : return;
997 : }
998 0 : if (itf->type() != Interface::VM_INTERFACE) {
999 0 : return;
1000 : }
1001 0 : const VmInterface *vmi = static_cast<const VmInterface *>(itf);
1002 0 : AddSloList(vmi->slo_list(), vmi_session_slo_rule_map);
1003 0 : if (vmi->vn()) {
1004 0 : AddSloList(vmi->vn()->slo_list(), vn_session_slo_rule_map);
1005 : }
1006 0 : return;
1007 : }
1008 :
1009 0 : void SessionStatsCollector::BuildSloList(
1010 : const SessionStatsInfo &stats_info,
1011 : const FlowEntry *fe,
1012 : SessionSloRuleMap *global_session_slo_rule_map,
1013 : SessionSloRuleMap *vmi_session_slo_rule_map,
1014 : SessionSloRuleMap *vn_session_slo_rule_map) {
1015 :
1016 0 : vmi_session_slo_rule_map->clear();
1017 0 : vn_session_slo_rule_map->clear();
1018 0 : global_session_slo_rule_map->clear();
1019 :
1020 0 : if (agent_uve_->agent()->oper_db()->global_vrouter()->slo_uuid() !=
1021 0 : boost::uuids::nil_uuid()) {
1022 0 : AddSloEntry(
1023 0 : agent_uve_->agent()->oper_db()->global_vrouter()->slo_uuid(),
1024 : global_session_slo_rule_map);
1025 : }
1026 :
1027 0 : if (stats_info.deleted) {
1028 0 : AddSloList(stats_info.export_info.vmi_slo_list, vmi_session_slo_rule_map);
1029 0 : AddSloList(stats_info.export_info.vn_slo_list, vn_session_slo_rule_map);
1030 : } else {
1031 0 : MakeSloList(fe, vmi_session_slo_rule_map, vn_session_slo_rule_map);
1032 : }
1033 0 : }
1034 :
1035 0 : bool SessionStatsCollector::UpdateSloMatchRuleEntry(
1036 : const boost::uuids::uuid &slo_uuid,
1037 : const std::string &match_uuid,
1038 : bool *match) {
1039 0 : SecurityLoggingObjectKey slo_key(slo_uuid);
1040 0 : SecurityLoggingObject *slo = static_cast<SecurityLoggingObject *>
1041 0 : (agent_uve_->agent()->slo_table()->FindActiveEntry(&slo_key));
1042 0 : if (slo) {
1043 : SessionSloState *state =
1044 0 : static_cast<SessionSloState *>(slo->GetState(agent_uve_->agent()->slo_table(),
1045 : slo_listener_id_));
1046 0 : if (state) {
1047 0 : return state->UpdateSessionSloStateRuleRefCount(match_uuid, match);
1048 : }
1049 : }
1050 0 : *match = false;
1051 0 : return false;
1052 0 : }
1053 :
1054 0 : bool SessionStatsCollector::CheckPolicyMatch(const SessionSloRuleMap &map,
1055 : const std::string &policy_uuid,
1056 : const bool &deleted_flag,
1057 : bool *match,
1058 : const bool &exported_once) {
1059 0 : SessionSloRuleMap::const_iterator it;
1060 0 : if (!policy_uuid.empty()) {
1061 0 : it = map.find(policy_uuid);
1062 0 : if (it != map.end()) {
1063 : /* Always logging tear down session, which is exported atleast once
1064 : * earlier will be logged other tear down sessions will be reported
1065 : * with SLO rate checking
1066 : */
1067 0 : if (deleted_flag && exported_once) {
1068 0 : *match = true;
1069 0 : return true;
1070 : }
1071 0 : return UpdateSloMatchRuleEntry(it->second.slo_uuid, policy_uuid, match);
1072 : }
1073 : }
1074 0 : *match = false;
1075 0 : return false;
1076 : }
1077 :
1078 0 : bool SessionStatsCollector::FindSloMatchRule(const SessionSloRuleMap &map,
1079 : const std::string &fw_policy_uuid,
1080 : const std::string &nw_policy_uuid,
1081 : const std::string &sg_policy_uuid,
1082 : const bool &deleted_flag,
1083 : bool *match,
1084 : const bool &exported_once) {
1085 0 : SessionSloRuleMap::const_iterator it;
1086 0 : bool fw_logged = false, nw_logged = false, sg_logged = false;
1087 0 : bool fw_match = false, nw_match = false, sg_match = false;
1088 :
1089 0 : fw_logged = CheckPolicyMatch(map, fw_policy_uuid, deleted_flag,
1090 : &fw_match, exported_once);
1091 0 : nw_logged = CheckPolicyMatch(map, nw_policy_uuid, deleted_flag,
1092 : &nw_match, exported_once);
1093 0 : sg_logged = CheckPolicyMatch(map, sg_policy_uuid, deleted_flag,
1094 : &sg_match, exported_once);
1095 :
1096 0 : if (fw_match || nw_match || sg_match) {
1097 0 : *match = true;
1098 : } else {
1099 0 : *match = false;
1100 : }
1101 :
1102 0 : if (fw_logged || nw_logged || sg_logged) {
1103 0 : return true;
1104 : }
1105 0 : return false;
1106 : }
1107 :
1108 0 : bool SessionStatsCollector::MatchSloForFlow(
1109 : const SessionStatsInfo &stats_info,
1110 : const FlowEntry *fe,
1111 : const std::string &fw_policy_uuid,
1112 : const std::string &nw_policy_uuid,
1113 : const std::string &sg_policy_uuid,
1114 : const bool &deleted_flag,
1115 : bool *logged,
1116 : const bool &exported_once) {
1117 :
1118 : bool is_vmi_slo_logged, is_vn_slo_logged, is_global_slo_logged;
1119 : bool vmi_slo_match, vn_slo_match, global_slo_match;
1120 0 : SessionSloRuleMap vmi_session_slo_rule_map;
1121 0 : SessionSloRuleMap vn_session_slo_rule_map;
1122 0 : SessionSloRuleMap global_session_slo_rule_map;
1123 :
1124 : /*
1125 : * Get the list of slos need to be matched for the given flow
1126 : */
1127 0 : BuildSloList(stats_info, fe,
1128 : &global_session_slo_rule_map,
1129 : &vmi_session_slo_rule_map,
1130 : &vn_session_slo_rule_map);
1131 : /*
1132 : * Match each type of policy for the given flow against the slo list
1133 : */
1134 :
1135 0 : is_vmi_slo_logged = FindSloMatchRule(vmi_session_slo_rule_map,
1136 : fw_policy_uuid,
1137 : nw_policy_uuid,
1138 : sg_policy_uuid,
1139 : deleted_flag,
1140 : &vmi_slo_match,
1141 : exported_once);
1142 :
1143 0 : is_vn_slo_logged = FindSloMatchRule(vn_session_slo_rule_map,
1144 : fw_policy_uuid,
1145 : nw_policy_uuid,
1146 : sg_policy_uuid,
1147 : deleted_flag,
1148 : &vn_slo_match,
1149 : exported_once);
1150 :
1151 0 : is_global_slo_logged = FindSloMatchRule(global_session_slo_rule_map,
1152 : fw_policy_uuid,
1153 : nw_policy_uuid,
1154 : sg_policy_uuid,
1155 : deleted_flag,
1156 : &global_slo_match,
1157 : exported_once);
1158 0 : if ((is_vmi_slo_logged) ||
1159 0 : (is_vn_slo_logged) ||
1160 : (is_global_slo_logged)) {
1161 0 : *logged = true;
1162 : }
1163 0 : if (vmi_slo_match || vn_slo_match || global_slo_match) {
1164 0 : return true;
1165 : } else {
1166 0 : return false;
1167 : }
1168 0 : }
1169 :
1170 0 : void SessionStatsCollector::GetPolicyIdFromDeletedFlow(
1171 : const SessionFlowExportInfo &flow_info,
1172 : std::string &fw_policy_uuid,
1173 : std::string &nw_policy_uuid,
1174 : std::string &sg_policy_uuid) {
1175 0 : fw_policy_uuid = flow_info.aps_rule_uuid;
1176 0 : sg_policy_uuid = flow_info.sg_rule_uuid;
1177 0 : nw_policy_uuid = flow_info.nw_ace_uuid;
1178 0 : return;
1179 : }
1180 :
1181 0 : void SessionStatsCollector::GetPolicyIdFromFlow(
1182 : const FlowEntry *fe,
1183 : std::string &fw_policy_uuid,
1184 : std::string &nw_policy_uuid,
1185 : std::string &sg_policy_uuid) {
1186 0 : fw_policy_uuid = fe->fw_policy_uuid();
1187 0 : sg_policy_uuid = fe->sg_rule_uuid();
1188 0 : nw_policy_uuid = fe->nw_ace_uuid();
1189 0 : return;
1190 : }
1191 :
1192 0 : bool SessionStatsCollector::FlowLogging(
1193 : const SessionStatsInfo &stats_info,
1194 : const FlowEntry *fe,
1195 : bool *logged,
1196 : const bool &exported_once) {
1197 :
1198 0 : bool matched = false, deleted_flag=false;
1199 0 : std::string fw_policy_uuid = "", nw_policy_uuid = "", sg_policy_uuid = "";
1200 :
1201 0 : GetPolicyIdFromFlow(fe,
1202 : fw_policy_uuid,
1203 : nw_policy_uuid,
1204 : sg_policy_uuid);
1205 :
1206 0 : matched = MatchSloForFlow(stats_info,
1207 : fe,
1208 : fw_policy_uuid,
1209 : nw_policy_uuid,
1210 : sg_policy_uuid,
1211 : deleted_flag,
1212 : logged,
1213 : exported_once);
1214 :
1215 0 : return matched;
1216 0 : }
1217 :
1218 0 : bool SessionStatsCollector::DeletedFlowLogging(
1219 : const SessionStatsInfo &stats_info,
1220 : const SessionFlowExportInfo &flow_info,
1221 : bool *logged,
1222 : const bool &exported_once) {
1223 :
1224 0 : bool matched = false, deleted_flag = true;
1225 0 : std::string fw_policy_uuid = "", nw_policy_uuid = "", sg_policy_uuid = "";
1226 :
1227 0 : GetPolicyIdFromDeletedFlow(flow_info,
1228 : fw_policy_uuid,
1229 : nw_policy_uuid,
1230 : sg_policy_uuid);
1231 :
1232 0 : matched = MatchSloForFlow(stats_info,
1233 : NULL,
1234 : fw_policy_uuid,
1235 : nw_policy_uuid,
1236 : sg_policy_uuid,
1237 : deleted_flag,
1238 : logged,
1239 : exported_once);
1240 :
1241 0 : return matched;
1242 0 : }
1243 :
1244 0 : bool SessionStatsCollector::HandleDeletedFlowLogging(
1245 : const SessionStatsInfo &stats_info) {
1246 :
1247 0 : bool logged = false;
1248 0 : const SessionExportInfo &info = stats_info.export_info;
1249 :
1250 : /*
1251 : * Deleted flow need to to be just checked whether SLO rules matched
1252 : * If SLO is macthed, it should be logged irrespective of the rate
1253 : */
1254 0 : if (DeletedFlowLogging(stats_info,
1255 0 : info.fwd_flow,
1256 : &logged,
1257 0 : stats_info.exported_atleast_once)) {
1258 0 : CheckFlowLogging(logged);
1259 0 : } else if (DeletedFlowLogging(stats_info,
1260 0 : info.rev_flow,
1261 : &logged,
1262 0 : stats_info.exported_atleast_once)) {
1263 0 : CheckFlowLogging(logged);
1264 : }
1265 0 : return false;
1266 : }
1267 :
1268 0 : bool SessionStatsCollector::HandleFlowLogging(
1269 : const SessionStatsInfo &stats_info) {
1270 0 : bool logged = false;
1271 :
1272 : /*
1273 : * FWD and REV flow of the Session need to be checked for SLO
1274 : * separately. If FWD flow matches or logged then rev flow
1275 : * is not required to check for SLO match.
1276 : * REV flow will be checked for SLO only when FWD flow
1277 : * is not matched for the SLO, since SLO is per session
1278 : */
1279 :
1280 0 : if (FlowLogging(stats_info,
1281 0 : stats_info.fwd_flow.flow.get(),
1282 : &logged,
1283 0 : stats_info.exported_atleast_once)) {
1284 0 : CheckFlowLogging(logged);
1285 0 : } else if (FlowLogging(stats_info,
1286 0 : stats_info.rev_flow.flow.get(),
1287 : &logged,
1288 0 : stats_info.exported_atleast_once)) {
1289 0 : CheckFlowLogging(logged);
1290 : }
1291 0 : return false;
1292 : }
1293 :
1294 57 : bool SessionStatsCollector::CheckSessionLogging(
1295 : const SessionStatsInfo &stats_info) {
1296 :
1297 57 : if (!agent_uve_->agent()->global_slo_status()) {
1298 : /* SLO is not enabled */
1299 57 : flow_stats_manager_->session_global_slo_logging_drops_++;
1300 57 : return false;
1301 : }
1302 :
1303 : /*
1304 : * Deleted flow will be logged if SLO is configured.
1305 : * Normal case will be logged only when there is a change in the
1306 : * stats. If there is no change in the session stats, it will
1307 : * not be considered to SLO match and rate. This will avoid logging
1308 : * of each session at least once. Also, idle session will not be
1309 : * considered for the rate count
1310 : */
1311 :
1312 0 : if (stats_info.deleted) {
1313 0 : if(HandleDeletedFlowLogging(stats_info)) {
1314 0 : return true;
1315 : }
1316 0 : } else if(HandleFlowLogging(stats_info)) {
1317 0 : return true;
1318 : }
1319 :
1320 0 : flow_stats_manager_->session_slo_logging_drops_++;
1321 0 : return false;
1322 : }
1323 :
1324 103 : uint64_t SessionStatsCollector::GetUpdatedSessionFlowBytes(uint64_t info_bytes,
1325 : uint64_t k_flow_bytes) const {
1326 103 : uint64_t oflow_bytes = 0xffff000000000000ULL & info_bytes;
1327 103 : uint64_t old_bytes = 0x0000ffffffffffffULL & info_bytes;
1328 103 : if (old_bytes > k_flow_bytes) {
1329 0 : oflow_bytes += 0x0001000000000000ULL;
1330 : }
1331 103 : return (oflow_bytes |= k_flow_bytes);
1332 : }
1333 :
1334 103 : uint64_t SessionStatsCollector::GetUpdatedSessionFlowPackets(
1335 : uint64_t info_packets,
1336 : uint64_t k_flow_pkts) const {
1337 103 : uint64_t oflow_pkts = 0xffffff0000000000ULL & info_packets;
1338 103 : uint64_t old_pkts = 0x000000ffffffffffULL & info_packets;
1339 103 : if (old_pkts > k_flow_pkts) {
1340 0 : oflow_pkts += 0x0000010000000000ULL;
1341 : }
1342 103 : return (oflow_pkts |= k_flow_pkts);
1343 : }
1344 :
1345 136 : void SessionStatsCollector::CopyFlowInfoInternal(SessionFlowExportInfo *info,
1346 : const boost::uuids::uuid &u,
1347 : FlowEntry *fe) const {
1348 136 : if (fe->uuid() != u) {
1349 0 : return;
1350 : }
1351 136 : FlowTable::GetFlowSandeshActionParams(fe->data().match_p.action_info,
1352 136 : info->action);
1353 136 : info->sg_rule_uuid = fe->sg_rule_uuid();
1354 136 : info->nw_ace_uuid = fe->nw_ace_uuid();
1355 136 : info->aps_rule_uuid = fe->fw_policy_uuid();
1356 136 : if (FlowEntry::ShouldDrop(fe->data().match_p.action_info.action)) {
1357 14 : info->drop_reason = FlowEntry::DropReasonStr(fe->data().drop_reason);
1358 : }
1359 : }
1360 :
1361 96 : void SessionStatsCollector::CopyFlowInfo(SessionStatsInfo &session,
1362 : const RevFlowDepParams *params) {
1363 96 : SessionExportInfo &info = session.export_info;
1364 96 : FlowEntry *fe = session.fwd_flow.flow.get();
1365 96 : FlowEntry *rfe = session.rev_flow.flow.get();
1366 96 : info.valid = true;
1367 :
1368 96 : const Interface *itf = fe->intf_entry();
1369 96 : if ((itf != NULL) && (itf->type() == Interface::VM_INTERFACE)) {
1370 96 : const VmInterface *vmi = static_cast<const VmInterface *>(itf);
1371 96 : if (vmi != NULL) {
1372 96 : info.vmi_slo_list = vmi->slo_list();
1373 96 : if (vmi->vn()) {
1374 70 : info.vn_slo_list = vmi->vn()->slo_list();
1375 : }
1376 : }
1377 : }
1378 :
1379 96 : if (fe->IsIngressFlow()) {
1380 35 : info.vm_cfg_name = fe->data().vm_cfg_name;
1381 61 : } else if (rfe) {
1382 : /* TODO: vm_cfg_name should be passed in RevFlowDepParams because rfe
1383 : * may now point to different UUID altogether */
1384 61 : info.vm_cfg_name = rfe->data().vm_cfg_name;
1385 : }
1386 96 : string rid = agent_uve_->agent()->router_id().to_string();
1387 96 : if (fe->is_flags_set(FlowEntry::LocalFlow)) {
1388 20 : info.other_vrouter = rid;
1389 : } else {
1390 76 : info.other_vrouter = fe->peer_vrouter();
1391 : }
1392 96 : info.underlay_proto = fe->tunnel_type().GetType();
1393 96 : CopyFlowInfoInternal(&info.fwd_flow, session.fwd_flow.uuid, fe);
1394 96 : if (params) {
1395 56 : FlowTable::GetFlowSandeshActionParams(params->action_info_,
1396 56 : info.rev_flow.action);
1397 56 : info.rev_flow.sg_rule_uuid = params->sg_uuid_;
1398 56 : info.rev_flow.nw_ace_uuid = params->nw_ace_uuid_;
1399 56 : if (FlowEntry::ShouldDrop(params->action_info_.action)) {
1400 0 : info.rev_flow.drop_reason = FlowEntry::DropReasonStr(params->
1401 0 : drop_reason_);
1402 : }
1403 40 : } else if (rfe) {
1404 40 : CopyFlowInfoInternal(&info.rev_flow, session.rev_flow.uuid, rfe);
1405 : }
1406 96 : }
1407 :
1408 0 : void SessionStatsCollector::FillSessionFlowStats
1409 : (const SessionFlowStatsParams &stats, SessionFlowInfo *flow_info,
1410 : bool is_sampling, bool is_logging) const {
1411 0 : if (!stats.valid) {
1412 0 : return;
1413 : }
1414 0 : flow_info->set_tcp_flags(stats.tcp_flags);
1415 0 : flow_info->set_underlay_source_port(stats.underlay_src_port);
1416 0 : if (is_sampling) {
1417 0 : flow_info->set_sampled_pkts(stats.diff_packets);
1418 0 : flow_info->set_sampled_bytes(stats.diff_bytes);
1419 : }
1420 0 : if (is_logging) {
1421 0 : flow_info->set_logged_pkts(stats.diff_packets);
1422 0 : flow_info->set_logged_bytes(stats.diff_bytes);
1423 : }
1424 : }
1425 :
1426 0 : void SessionStatsCollector::FillSessionFlowInfo
1427 : (const SessionFlowStatsInfo &session_flow, const SessionStatsInfo &sinfo,
1428 : const SessionFlowExportInfo &einfo, SessionFlowInfo *flow_info) const {
1429 0 : FlowEntry *fe = session_flow.flow.get();
1430 0 : std::string action_str, drop_reason = "";
1431 :
1432 0 : flow_info->set_flow_uuid(session_flow.uuid);
1433 0 : flow_info->set_setup_time(sinfo.setup_time);
1434 0 : if (sinfo.teardown_time) {
1435 0 : flow_info->set_teardown_time(sinfo.teardown_time);
1436 0 : flow_info->set_teardown_bytes(session_flow.total_bytes);
1437 0 : flow_info->set_teardown_pkts(session_flow.total_packets);
1438 : }
1439 0 : if (sinfo.deleted) {
1440 0 : if (!sinfo.export_info.valid) {
1441 0 : return;
1442 : }
1443 0 : if (!einfo.action.empty()) {
1444 0 : flow_info->set_action(einfo.action);
1445 : }
1446 0 : if (!einfo.sg_rule_uuid.empty()) {
1447 0 : flow_info->set_sg_rule_uuid(StringToUuid(einfo.sg_rule_uuid));
1448 : }
1449 0 : if (!einfo.nw_ace_uuid.empty()) {
1450 0 : flow_info->set_nw_ace_uuid(StringToUuid(einfo.nw_ace_uuid));
1451 : }
1452 0 : if (!einfo.drop_reason.empty()) {
1453 0 : flow_info->set_drop_reason(einfo.drop_reason);
1454 : }
1455 : } else {
1456 0 : FlowTable::GetFlowSandeshActionParams(fe->data().match_p.action_info,
1457 : action_str);
1458 0 : flow_info->set_action(action_str);
1459 0 : flow_info->set_sg_rule_uuid(StringToUuid(fe->sg_rule_uuid()));
1460 0 : flow_info->set_nw_ace_uuid(StringToUuid(fe->nw_ace_uuid()));
1461 0 : if (FlowEntry::ShouldDrop(fe->data().match_p.action_info.action)) {
1462 0 : drop_reason = FlowEntry::DropReasonStr(fe->data().drop_reason);
1463 0 : flow_info->set_drop_reason(drop_reason);
1464 : }
1465 : }
1466 0 : }
1467 :
1468 0 : bool SessionStatsCollector::CheckAndDeleteSessionStatsFlow(
1469 : SessionPreAggInfo::SessionMap::iterator session_map_iter) {
1470 0 : FlowEntry *fe = session_map_iter->second.fwd_flow.flow.get();
1471 0 : FlowEntry *rfe = session_map_iter->second.rev_flow.flow.get();
1472 0 : FLOW_LOCK(fe, rfe, FlowEvent::FLOW_MESSAGE);
1473 0 : if (fe->deleted()) {
1474 0 : DeleteSession(fe, session_map_iter->first.uuid,
1475 : GetCurrentTime(), NULL);
1476 0 : return true;
1477 : }
1478 0 : return false;
1479 0 : }
1480 :
1481 0 : bool SessionStatsCollector::SessionStatsChangedLocked
1482 : (SessionPreAggInfo::SessionMap::iterator session_map_iter,
1483 : SessionStatsParams *params) const {
1484 0 : FlowEntry *fe = session_map_iter->second.fwd_flow.flow.get();
1485 0 : FlowEntry *rfe = session_map_iter->second.rev_flow.flow.get();
1486 0 : FLOW_LOCK(fe, rfe, FlowEvent::FLOW_MESSAGE);
1487 0 : return SessionStatsChangedUnlocked(session_map_iter, params);
1488 0 : }
1489 :
1490 101 : bool SessionStatsCollector::SessionStatsChangedUnlocked
1491 : (SessionPreAggInfo::SessionMap::iterator session_map_iter,
1492 : SessionStatsParams *params) const {
1493 :
1494 101 : bool fwd_updated = FetchFlowStats(&session_map_iter->second.fwd_flow,
1495 : ¶ms->fwd_flow);
1496 101 : bool rev_updated = FetchFlowStats(&session_map_iter->second.rev_flow,
1497 : ¶ms->rev_flow);
1498 101 : return (fwd_updated || rev_updated);
1499 : }
1500 :
1501 202 : bool SessionStatsCollector::FetchFlowStats
1502 : (SessionFlowStatsInfo *info, SessionFlowStatsParams *params) const {
1503 : vr_flow_stats k_stats;
1504 : KFlowData kinfo;
1505 : uint64_t k_bytes, bytes, k_packets;
1506 202 : const vr_flow_entry *k_flow = NULL;
1507 202 : KSyncFlowMemory *ksync_obj = agent_uve_->agent()->ksync()->
1508 202 : ksync_flow_memory();
1509 : /* Update gen-id and flow-handle before reading stats using them. For
1510 : * reverse-flow, it is possible that flow-handle is not set yet
1511 : */
1512 202 : FlowEntry *fe = info->flow.get();
1513 202 : if (fe && (info->uuid == fe->uuid())) {
1514 192 : info->flow_handle = fe->flow_handle();
1515 192 : info->gen_id = fe->gen_id();
1516 : }
1517 :
1518 202 : k_flow = ksync_obj->GetKFlowStatsAndInfo(info->flow->key(),
1519 : info->flow_handle,
1520 202 : info->gen_id, &k_stats, &kinfo);
1521 202 : if (!k_flow) {
1522 99 : SandeshFlowKey skey;
1523 99 : skey.set_nh(info->flow->key().nh);
1524 99 : skey.set_sip(info->flow->key().src_addr.to_string());
1525 99 : skey.set_dip(info->flow->key().dst_addr.to_string());
1526 99 : skey.set_src_port(info->flow->key().src_port);
1527 99 : skey.set_dst_port(info->flow->key().dst_port);
1528 99 : skey.set_protocol(info->flow->key().protocol);
1529 99 : SESSION_STATS_TRACE(Err, "Fetching stats failed", info->flow_handle,
1530 : info->gen_id, skey);
1531 99 : return false;
1532 99 : }
1533 :
1534 0 : k_bytes = FlowStatsCollector::GetFlowStats(k_stats.flow_bytes_oflow,
1535 103 : k_stats.flow_bytes);
1536 0 : k_packets = FlowStatsCollector::GetFlowStats(k_stats.flow_packets_oflow,
1537 103 : k_stats.flow_packets);
1538 :
1539 103 : bytes = 0x0000ffffffffffffULL & info->total_bytes;
1540 :
1541 103 : if (bytes != k_bytes) {
1542 103 : uint64_t total_bytes = GetUpdatedSessionFlowBytes(info->total_bytes,
1543 : k_bytes);
1544 : uint64_t total_packets = GetUpdatedSessionFlowPackets
1545 103 : (info->total_packets, k_packets);
1546 103 : params->diff_bytes = total_bytes - info->total_bytes;
1547 103 : params->diff_packets = total_packets - info->total_packets;
1548 103 : info->total_bytes = total_bytes;
1549 103 : info->total_packets = total_packets;
1550 103 : params->tcp_flags = kinfo.tcp_flags;
1551 103 : params->underlay_src_port = kinfo.underlay_src_port;
1552 103 : params->valid = true;
1553 103 : return true;
1554 : }
1555 0 : return false;
1556 : }
1557 :
1558 0 : void SessionStatsCollector::FillSessionInfoLocked
1559 : (SessionPreAggInfo::SessionMap::iterator session_map_iter,
1560 : const SessionStatsParams &stats, SessionInfo *session_info,
1561 : SessionIpPort *session_key, bool is_sampling, bool is_logging) const {
1562 0 : FlowEntry *fe = session_map_iter->second.fwd_flow.flow.get();
1563 0 : FlowEntry *rfe = session_map_iter->second.rev_flow.flow.get();
1564 0 : FLOW_LOCK(fe, rfe, FlowEvent::FLOW_MESSAGE);
1565 0 : FillSessionInfoUnlocked(session_map_iter, stats, session_info, session_key, NULL,
1566 : true, is_sampling, is_logging);
1567 0 : }
1568 :
1569 0 : void SessionStatsCollector::FillSessionEvictStats
1570 : (SessionPreAggInfo::SessionMap::iterator session_map_iter,
1571 : SessionInfo *session_info, bool is_sampling, bool is_logging) const {
1572 0 : const SessionStatsParams &estats = session_map_iter->second.evict_stats;
1573 0 : if (!estats.fwd_flow.valid) {
1574 0 : return;
1575 : }
1576 0 : if (is_logging) {
1577 0 : session_info->forward_flow_info.set_logged_pkts(estats.fwd_flow.
1578 0 : diff_packets);
1579 0 : session_info->forward_flow_info.set_logged_bytes(estats.fwd_flow.
1580 0 : diff_bytes);
1581 : /* TODO: Evict stats for reverse flow is not supported yet */
1582 0 : session_info->reverse_flow_info.set_logged_pkts(0);
1583 0 : session_info->reverse_flow_info.set_logged_bytes(0);
1584 : }
1585 0 : if (is_sampling) {
1586 0 : session_info->forward_flow_info.set_sampled_pkts(estats.fwd_flow.
1587 0 : diff_packets);
1588 0 : session_info->forward_flow_info.set_sampled_bytes(estats.fwd_flow.
1589 0 : diff_bytes);
1590 : /* TODO: Evict stats for reverse flow is not supported yet */
1591 0 : session_info->reverse_flow_info.set_sampled_pkts(0);
1592 0 : session_info->reverse_flow_info.set_sampled_bytes(0);
1593 : }
1594 : }
1595 :
1596 0 : void SessionStatsCollector::FillSessionInfoUnlocked
1597 : (SessionPreAggInfo::SessionMap::iterator session_map_iter,
1598 : const SessionStatsParams &stats,
1599 : SessionInfo *session_info,
1600 : SessionIpPort *session_key,
1601 : const RevFlowDepParams *params,
1602 : bool read_flow, bool is_sampling, bool is_logging) const {
1603 0 : string rid = agent_uve_->agent()->router_id().to_string();
1604 0 : FlowEntry *fe = session_map_iter->second.fwd_flow.flow.get();
1605 0 : FlowEntry *rfe = session_map_iter->second.rev_flow.flow.get();
1606 0 : boost::system::error_code ec;
1607 : /*
1608 : * Fill the session Key
1609 : */
1610 0 : session_key->set_ip(session_map_iter->first.remote_ip);
1611 0 : session_key->set_port(session_map_iter->first.client_port);
1612 0 : FillSessionFlowInfo(session_map_iter->second.fwd_flow,
1613 0 : session_map_iter->second,
1614 0 : session_map_iter->second.export_info.fwd_flow,
1615 : &session_info->forward_flow_info);
1616 0 : FillSessionFlowInfo(session_map_iter->second.rev_flow,
1617 0 : session_map_iter->second,
1618 0 : session_map_iter->second.export_info.rev_flow,
1619 : &session_info->reverse_flow_info);
1620 0 : bool first_time_export = false;
1621 0 : if (!session_map_iter->second.exported_atleast_once) {
1622 0 : first_time_export = true;
1623 : /* Mark the flow as exported */
1624 0 : session_map_iter->second.exported_atleast_once = true;
1625 : }
1626 :
1627 0 : const bool &evicted = session_map_iter->second.evicted;
1628 0 : const bool &deleted = session_map_iter->second.deleted;
1629 0 : if (evicted) {
1630 0 : const SessionStatsParams &estats = session_map_iter->second.evict_stats;
1631 0 : FillSessionEvictStats(session_map_iter, session_info, is_sampling,
1632 : is_logging);
1633 0 : flow_stats_manager_->UpdateSessionExportStats(1, first_time_export,
1634 0 : estats.sampled);
1635 : } else {
1636 0 : const SessionStatsParams *real_stats = &stats;
1637 0 : if (deleted) {
1638 0 : real_stats = &session_map_iter->second.del_stats;
1639 : }
1640 0 : FillSessionFlowStats(real_stats->fwd_flow,
1641 : &session_info->forward_flow_info, is_sampling,
1642 : is_logging);
1643 0 : FillSessionFlowStats(real_stats->rev_flow,
1644 : &session_info->reverse_flow_info, is_sampling,
1645 : is_logging);
1646 0 : flow_stats_manager_->UpdateSessionExportStats(1, first_time_export,
1647 0 : real_stats->sampled);
1648 : }
1649 0 : if (deleted) {
1650 0 : SessionExportInfo &info = session_map_iter->second.export_info;
1651 0 : if (info.valid) {
1652 0 : if (!info.vm_cfg_name.empty()) {
1653 0 : session_info->set_vm(info.vm_cfg_name);
1654 : }
1655 0 : session_info->set_other_vrouter_ip(
1656 0 : AddressFromString(info.other_vrouter, &ec));
1657 0 : session_info->set_underlay_proto(info.underlay_proto);
1658 : }
1659 : } else {
1660 0 : session_info->set_vm(fe->data().vm_cfg_name);
1661 0 : if (fe->is_flags_set(FlowEntry::LocalFlow)) {
1662 0 : session_info->set_other_vrouter_ip(AddressFromString(rid, &ec));
1663 : } else {
1664 : /* For Egress flows, pick VM name from reverse flow */
1665 0 : if (!fe->IsIngressFlow() && rfe) {
1666 0 : session_info->set_vm(rfe->data().vm_cfg_name);
1667 : }
1668 0 : session_info->set_other_vrouter_ip(
1669 0 : AddressFromString(fe->peer_vrouter(), &ec));
1670 : }
1671 0 : session_info->set_underlay_proto(fe->tunnel_type().GetType());
1672 : }
1673 0 : }
1674 :
1675 0 : void SessionStatsCollector::UpdateAggregateStats(const SessionInfo &sinfo,
1676 : SessionAggInfo *agg_info,
1677 : bool is_sampling,
1678 : bool is_logging) const {
1679 0 : if (is_sampling) {
1680 0 : agg_info->set_sampled_forward_bytes(agg_info->get_sampled_forward_bytes() +
1681 0 : sinfo.get_forward_flow_info().get_sampled_bytes());
1682 0 : agg_info->set_sampled_forward_pkts(agg_info->get_sampled_forward_pkts() +
1683 0 : sinfo.get_forward_flow_info().get_sampled_pkts());
1684 0 : agg_info->set_sampled_reverse_bytes(agg_info->get_sampled_reverse_bytes() +
1685 0 : sinfo.get_reverse_flow_info().get_sampled_bytes());
1686 0 : agg_info->set_sampled_reverse_pkts(agg_info->get_sampled_reverse_pkts() +
1687 0 : sinfo.get_reverse_flow_info().get_sampled_pkts());
1688 : }
1689 0 : if (is_logging) {
1690 0 : agg_info->set_logged_forward_bytes(agg_info->get_logged_forward_bytes() +
1691 0 : sinfo.get_forward_flow_info().get_logged_bytes());
1692 0 : agg_info->set_logged_forward_pkts(agg_info->get_logged_forward_pkts() +
1693 0 : sinfo.get_forward_flow_info().get_logged_pkts());
1694 0 : agg_info->set_logged_reverse_bytes(agg_info->get_logged_reverse_bytes() +
1695 0 : sinfo.get_reverse_flow_info().get_logged_bytes());
1696 0 : agg_info->set_logged_reverse_pkts(agg_info->get_logged_reverse_pkts() +
1697 0 : sinfo.get_reverse_flow_info().get_logged_pkts());
1698 : }
1699 0 : }
1700 :
1701 0 : void SessionStatsCollector::FillSessionAggInfo
1702 : (SessionEndpointInfo::SessionAggMap::iterator it, SessionIpPortProtocol *key)
1703 : const {
1704 : /*
1705 : * Fill the session agg key
1706 : */
1707 0 : key->set_local_ip(it->first.local_ip);
1708 0 : key->set_service_port(it->first.server_port);
1709 0 : key->set_protocol(it->first.proto);
1710 0 : }
1711 :
1712 0 : void SessionStatsCollector::FillSessionTags(const TagList &list,
1713 : SessionEndpoint *ep) const {
1714 0 : UveTagData tinfo(UveTagData::SET);
1715 0 : agent_uve_->BuildTagNamesFromList(list, &tinfo);
1716 0 : if (!tinfo.application.empty()) {
1717 0 : ep->set_application(tinfo.application);
1718 : }
1719 0 : if (!tinfo.tier.empty()) {
1720 0 : ep->set_tier(tinfo.tier);
1721 : }
1722 0 : if (!tinfo.site.empty()) {
1723 0 : ep->set_site(tinfo.site);
1724 : }
1725 0 : if (!tinfo.deployment.empty()) {
1726 0 : ep->set_deployment(tinfo.deployment);
1727 : }
1728 0 : if (tinfo.label_set.size() != 0) {
1729 0 : ep->set_labels(tinfo.label_set);
1730 : }
1731 0 : if (tinfo.custom_tag_set.size() != 0) {
1732 0 : ep->set_custom_tags(tinfo.custom_tag_set);
1733 : }
1734 0 : }
1735 :
1736 0 : void SessionStatsCollector::FillSessionRemoteTags(const TagList &list,
1737 : SessionEndpoint *ep) const {
1738 0 : UveTagData tinfo(UveTagData::SET);
1739 0 : agent_uve_->BuildTagIdsFromList(list, &tinfo);
1740 0 : if (!tinfo.application.empty()){
1741 0 : ep->set_remote_application(tinfo.application);
1742 : }
1743 0 : if (!tinfo.tier.empty()){
1744 0 : ep->set_remote_tier(tinfo.tier);
1745 : }
1746 0 : if (!tinfo.site.empty()){
1747 0 : ep->set_remote_site(tinfo.site);
1748 : }
1749 0 : if (!tinfo.deployment.empty()){
1750 0 : ep->set_remote_deployment(tinfo.deployment);
1751 : }
1752 0 : if (tinfo.label_set.size() != 0) {
1753 0 : ep->set_remote_labels(tinfo.label_set);
1754 : }
1755 0 : if (tinfo.custom_tag_set.size() != 0) {
1756 0 : ep->set_remote_custom_tags(tinfo.custom_tag_set);
1757 : }
1758 0 : }
1759 :
1760 0 : void SessionStatsCollector::FillSessionEndpoint(SessionEndpointMap::iterator it,
1761 : SessionEndpoint *session_ep)
1762 : const {
1763 0 : string rid = agent_uve_->agent()->router_id().to_string();
1764 0 : boost::system::error_code ec;
1765 :
1766 0 : session_ep->set_vmi(it->first.vmi_cfg_name);
1767 0 : session_ep->set_vn(it->first.local_vn);
1768 0 : session_ep->set_remote_vn(it->first.remote_vn);
1769 0 : session_ep->set_is_client_session(it->first.is_client_session);
1770 0 : session_ep->set_is_si(it->first.is_si);
1771 0 : if (!it->first.remote_prefix.empty()) {
1772 0 : session_ep->set_remote_prefix(it->first.remote_prefix);
1773 : }
1774 0 : session_ep->set_security_policy_rule(it->first.match_policy);
1775 0 : if (it->first.local_tagset.size() > 0) {
1776 0 : FillSessionTags(it->first.local_tagset, session_ep);
1777 : }
1778 0 : if (it->first.remote_tagset.size() > 0) {
1779 0 : FillSessionRemoteTags(it->first.remote_tagset, session_ep);
1780 : }
1781 0 : session_ep->set_vrouter_ip(AddressFromString(rid, &ec));
1782 0 : }
1783 :
1784 26 : bool SessionStatsCollector::ProcessSessionEndpoint
1785 : (const SessionEndpointMap::iterator &it) {
1786 26 : SessionEndpointInfo::SessionAggMap::iterator session_agg_map_iter;
1787 26 : SessionEndpointInfo::SessionAggMap::iterator prev_agg_iter;
1788 26 : SessionPreAggInfo::SessionMap::iterator session_map_iter, prev;
1789 :
1790 26 : SessionInfo session_info;
1791 26 : SessionIpPort session_key;
1792 26 : uint32_t session_count = 0, session_agg_count = 0;
1793 26 : bool exit = false, ep_completed = true;
1794 :
1795 26 : SessionEndpoint &session_ep = session_msg_list_[GetSessionMsgIdx()];
1796 :
1797 26 : session_agg_map_iter = it->second.session_agg_map_.
1798 26 : lower_bound(session_agg_iteration_key_);
1799 75 : while (session_agg_map_iter != it->second.session_agg_map_.end()) {
1800 52 : SessionAggInfo session_agg_info;
1801 52 : SessionIpPortProtocol session_agg_key;
1802 52 : session_count = 0;
1803 52 : session_map_iter = session_agg_map_iter->second.session_map_.
1804 52 : lower_bound(session_iteration_key_);
1805 109 : while (session_map_iter != session_agg_map_iter->second.session_map_.end()) {
1806 57 : prev = session_map_iter;
1807 57 : SessionStatsParams params;
1808 57 : if (!session_map_iter->second.deleted &&
1809 0 : !session_map_iter->second.evicted) {
1810 0 : bool delete_marked = CheckAndDeleteSessionStatsFlow(session_map_iter);
1811 0 : if (!delete_marked) {
1812 0 : bool changed = SessionStatsChangedLocked(session_map_iter,
1813 : ¶ms);
1814 0 : if (!changed && session_map_iter->second.exported_atleast_once) {
1815 0 : ++session_map_iter;
1816 57 : continue;
1817 : }
1818 : }
1819 : }
1820 :
1821 57 : bool is_sampling = true;
1822 57 : if (IsSamplingEnabled()) {
1823 57 : is_sampling = SampleSession(session_map_iter, ¶ms);
1824 : }
1825 57 : bool is_logging = CheckSessionLogging(session_map_iter->second);
1826 :
1827 : /* Ignore session export if sampling & logging drop the session */
1828 57 : if (!is_sampling && !is_logging) {
1829 57 : ++session_map_iter;
1830 57 : if (prev->second.deleted) {
1831 57 : session_agg_map_iter->second.session_map_.erase(prev);
1832 : }
1833 57 : continue;
1834 : }
1835 0 : if (session_map_iter->second.deleted) {
1836 0 : FillSessionInfoUnlocked(session_map_iter, params, &session_info,
1837 : &session_key, NULL, true, is_sampling,
1838 : is_logging);
1839 : } else {
1840 0 : FillSessionInfoLocked(session_map_iter, params, &session_info,
1841 : &session_key, is_sampling, is_logging);
1842 : }
1843 0 : session_agg_info.sessionMap.insert(make_pair(session_key,
1844 : session_info));
1845 0 : UpdateAggregateStats(session_info, &session_agg_info, is_sampling,
1846 : is_logging);
1847 0 : ++session_map_iter;
1848 0 : ++session_count;
1849 0 : if (prev->second.deleted) {
1850 0 : session_agg_map_iter->second.session_map_.erase(prev);
1851 : }
1852 0 : if (session_count ==
1853 0 : agent_uve_->agent()->params()->max_sessions_per_aggregate()) {
1854 0 : exit = true;
1855 0 : break;
1856 : }
1857 : }
1858 52 : if (session_count) {
1859 0 : FillSessionAggInfo(session_agg_map_iter, &session_agg_key);
1860 0 : session_ep.sess_agg_info.insert(make_pair(session_agg_key,
1861 : session_agg_info));
1862 : }
1863 52 : if (exit) {
1864 0 : break;
1865 : }
1866 52 : session_iteration_key_.Reset();
1867 52 : prev_agg_iter = session_agg_map_iter;
1868 52 : session_agg_map_iter++;
1869 52 : if (prev_agg_iter->second.session_map_.size() == 0) {
1870 52 : it->second.session_agg_map_.erase(prev_agg_iter);
1871 : }
1872 52 : ++session_agg_count;
1873 52 : if (session_agg_count ==
1874 52 : agent_uve_->agent()->params()->max_aggregates_per_session_endpoint()) {
1875 3 : break;
1876 : }
1877 55 : }
1878 : /* Don't export SessionEndpoint if there are 0 aggregates */
1879 26 : if (session_ep.sess_agg_info.size()) {
1880 0 : FillSessionEndpoint(it, &session_ep);
1881 0 : EnqueueSessionMsg();
1882 : }
1883 :
1884 26 : if (session_count ==
1885 26 : agent_uve_->agent()->params()->max_sessions_per_aggregate()) {
1886 0 : ep_completed = false;
1887 0 : session_ep_iteration_key_ = it->first;
1888 0 : session_agg_iteration_key_ = session_agg_map_iter->first;
1889 0 : if (session_map_iter == session_agg_map_iter->second.session_map_.end()) {
1890 0 : prev_agg_iter = session_agg_map_iter;
1891 0 : ++session_agg_map_iter;
1892 0 : if (prev_agg_iter->second.session_map_.size() == 0) {
1893 0 : it->second.session_agg_map_.erase(prev_agg_iter);
1894 : }
1895 0 : if (session_agg_map_iter == it->second.session_agg_map_.end()) {
1896 : /* session_iteration_key_ and session_agg_iteration_key_ are
1897 : * both reset when ep_completed is returned as true in the
1898 : * calling function */
1899 0 : ep_completed = true;
1900 : } else {
1901 0 : session_iteration_key_.Reset();
1902 0 : session_agg_iteration_key_ = session_agg_map_iter->first;
1903 : }
1904 : } else {
1905 0 : session_iteration_key_ = session_map_iter->first;
1906 : }
1907 26 : } else if (session_agg_count ==
1908 26 : agent_uve_->agent()->params()->
1909 26 : max_aggregates_per_session_endpoint()) {
1910 3 : ep_completed = false;
1911 3 : session_ep_iteration_key_ = it->first;
1912 3 : if (session_agg_map_iter == it->second.session_agg_map_.end()) {
1913 : /* session_iteration_key_ and session_agg_iteration_key_ are both
1914 : * reset when ep_completed is returned as true in the calling
1915 : * function */
1916 0 : ep_completed = true;
1917 : } else {
1918 3 : session_agg_iteration_key_ = session_agg_map_iter->first;
1919 3 : session_iteration_key_.Reset();
1920 : }
1921 : }
1922 26 : return ep_completed;
1923 26 : }
1924 :
1925 6 : uint32_t SessionStatsCollector::RunSessionEndpointStats(uint32_t max_count) {
1926 : SessionEndpointMap::iterator it = session_endpoint_map_.
1927 6 : lower_bound(session_ep_iteration_key_);
1928 6 : if (it == session_endpoint_map_.end()) {
1929 0 : it = session_endpoint_map_.begin();
1930 : }
1931 6 : if (it == session_endpoint_map_.end()) {
1932 0 : return 0;
1933 : }
1934 :
1935 6 : uint32_t count = 0;
1936 32 : while (count < max_count) {
1937 32 : if (it == session_endpoint_map_.end()) {
1938 6 : break;
1939 : }
1940 :
1941 : /* ProcessSessionEndpoint will build 1 SessionEndpoint message. This
1942 : * may or may not include all aggregates and all sessions within each
1943 : * aggregate. It returns true if the built message includes all
1944 : * aggregates and all sessions of each aggregate */
1945 26 : bool ep_completed = ProcessSessionEndpoint(it);
1946 26 : ++count;
1947 26 : if (ep_completed) {
1948 23 : SessionEndpointMap::iterator prev = it;
1949 23 : ++it;
1950 23 : ++session_ep_visited_;
1951 23 : session_agg_iteration_key_.Reset();
1952 23 : session_iteration_key_.Reset();
1953 23 : if (prev->second.session_agg_map_.size() == 0) {
1954 23 : session_endpoint_map_.erase(prev);
1955 : }
1956 : }
1957 : }
1958 :
1959 : //Send any pending session export messages
1960 6 : DispatchPendingSessionMsg();
1961 :
1962 : // Update iterator for next pass
1963 6 : if (it == session_endpoint_map_.end()) {
1964 6 : session_ep_iteration_key_.Reset();
1965 : } else {
1966 0 : session_ep_iteration_key_ = it->first;
1967 : }
1968 :
1969 6 : session_task_ = NULL;
1970 6 : return count;
1971 : }
1972 : /////////////////////////////////////////////////////////////////////////////
1973 : // Introspect routines
1974 : /////////////////////////////////////////////////////////////////////////////
1975 : //TBD
1976 :
1977 : /////////////////////////////////////////////////////////////////////////////
1978 : // Session Stats task
1979 : /////////////////////////////////////////////////////////////////////////////
1980 6 : SessionStatsCollector::SessionTask::SessionTask(SessionStatsCollector *ssc) :
1981 6 : Task(ssc->task_id(), ssc->instance_id()), ssc_(ssc) {
1982 6 : }
1983 :
1984 12 : SessionStatsCollector::SessionTask::~SessionTask() {
1985 12 : }
1986 :
1987 0 : std::string SessionStatsCollector::SessionTask::Description() const {
1988 0 : return "Session Stats Collector Task";
1989 : }
1990 :
1991 6 : bool SessionStatsCollector::SessionTask::Run() {
1992 6 : ssc_->RunSessionEndpointStats(kSessionsPerTask);
1993 6 : return true;
1994 : }
1995 :
1996 57 : bool SessionStatsCollector::IsSamplingEnabled() const {
1997 57 : int32_t cfg_rate = agent_uve_->agent()->oper_db()->global_vrouter()->
1998 57 : flow_export_rate();
1999 57 : if (cfg_rate == GlobalVrouter::kDisableSampling) {
2000 0 : return false;
2001 : }
2002 57 : return true;
2003 : }
2004 :
2005 57 : bool SessionStatsCollector::SampleSession
2006 : (SessionPreAggInfo::SessionMap::iterator session_map_iter,
2007 : SessionStatsParams *params) const {
2008 57 : int32_t cfg_rate = agent_uve_->agent()->oper_db()->global_vrouter()->
2009 57 : flow_export_rate();
2010 : /* If session export is disabled, update stats and return */
2011 57 : if (!cfg_rate) {
2012 57 : flow_stats_manager_->session_export_disable_drops_++;
2013 57 : return false;
2014 : }
2015 0 : const bool &deleted = session_map_iter->second.deleted;
2016 0 : const bool &evicted = session_map_iter->second.evicted;
2017 0 : SessionStatsParams *stats = params;
2018 0 : if (evicted) {
2019 0 : stats = &session_map_iter->second.evict_stats;
2020 0 : } else if (deleted) {
2021 0 : stats = &session_map_iter->second.del_stats;
2022 : }
2023 0 : stats->sampled = false;
2024 : /* For session-sampling diff_bytes should consider the diff bytes for both
2025 : * forward and reverse flow */
2026 0 : uint64_t diff_bytes = stats->fwd_flow.diff_bytes +
2027 0 : stats->rev_flow.diff_bytes;
2028 0 : const SessionStatsInfo &info = session_map_iter->second;
2029 : /* Subject a flow to sampling algorithm only when all of below is met:-
2030 : * a. actual session-export-rate is >= 80% of configured flow-export-rate.
2031 : * This is done only for first time.
2032 : * b. diff_bytes is lesser than the threshold
2033 : * c. Flow-sample does not have teardown time or the sample for the flow is
2034 : * not exported earlier.
2035 : */
2036 0 : bool subject_flows_to_algorithm = false;
2037 0 : if ((diff_bytes < threshold()) &&
2038 0 : (!info.teardown_time || !info.exported_atleast_once) &&
2039 0 : ((!flow_stats_manager_->sessions_sampled_atleast_once_ &&
2040 0 : flow_stats_manager_->session_export_rate() >= ((double)cfg_rate) * 0.8)
2041 0 : || flow_stats_manager_->sessions_sampled_atleast_once_)) {
2042 0 : subject_flows_to_algorithm = true;
2043 0 : flow_stats_manager_->set_sessions_sampled_atleast_once();
2044 : }
2045 :
2046 0 : if (subject_flows_to_algorithm) {
2047 0 : double probability = diff_bytes/threshold();
2048 0 : uint32_t num = rand() % threshold();
2049 0 : if (num > diff_bytes) {
2050 : /* Do not export the flow, if the random number generated is more
2051 : * than the diff_bytes */
2052 0 : flow_stats_manager_->session_export_sampling_drops_++;
2053 : /* The second part of the if condition below is not required but
2054 : * added for better readability. It is not required because
2055 : * exported_atleast_once() will always be false if teardown time is
2056 : * set. If both teardown_time and exported_atleast_once are true we
2057 : * will never be here */
2058 0 : if (info.teardown_time && !info.exported_atleast_once) {
2059 : /* This counter indicates the number of sessions that were
2060 : * never exported */
2061 0 : flow_stats_manager_->session_export_drops_++;
2062 : }
2063 0 : return false;
2064 : }
2065 0 : stats->sampled = true;
2066 : /* Normalize the diff_bytes and diff_packets reported using the
2067 : * probability value */
2068 0 : if (probability == 0) {
2069 0 : stats->fwd_flow.diff_bytes = 0;
2070 0 : stats->fwd_flow.diff_packets = 0;
2071 0 : stats->rev_flow.diff_bytes = 0;
2072 0 : stats->rev_flow.diff_packets = 0;
2073 : } else {
2074 0 : stats->fwd_flow.diff_bytes = stats->fwd_flow.diff_bytes/
2075 : probability;
2076 0 : stats->fwd_flow.diff_packets = stats->fwd_flow.diff_packets/
2077 : probability;
2078 0 : stats->rev_flow.diff_bytes = stats->rev_flow.diff_bytes/
2079 : probability;
2080 0 : stats->rev_flow.diff_packets = stats->rev_flow.diff_packets/
2081 : probability;
2082 : }
2083 : }
2084 :
2085 0 : return true;
2086 : }
2087 :
2088 0 : uint64_t SessionStatsCollector::threshold() const {
2089 0 : return flow_stats_manager_->threshold();
2090 : }
2091 :
2092 : /////////////////////////////////////////////////////////////////////////////
2093 : // SessionStatsCollectorObject methods
2094 : /////////////////////////////////////////////////////////////////////////////
2095 2 : SessionStatsCollectorObject::SessionStatsCollectorObject(Agent *agent,
2096 0 : FlowStatsManager *mgr) {
2097 4 : for (int i = 0; i < kMaxSessionCollectors; i++) {
2098 2 : uint32_t instance_id = mgr->AllocateIndex();
2099 : boost::asio::io_context& io_ref =
2100 : const_cast<boost::asio::io_context&>
2101 2 : (*agent->event_manager()->io_service());
2102 2 : collectors[i].reset(
2103 : AgentStaticObjectFactory::CreateRef<SessionStatsCollector>(
2104 : io_ref,
2105 : agent->uve(), instance_id, mgr, this));
2106 : }
2107 2 : }
2108 :
2109 0 : SessionStatsCollector* SessionStatsCollectorObject::GetCollector(uint8_t idx) const {
2110 0 : if (idx >= 0 && idx < kMaxSessionCollectors) {
2111 0 : return collectors[idx].get();
2112 : }
2113 0 : return NULL;
2114 : }
2115 :
2116 0 : void SessionStatsCollectorObject::SetExpiryTime(int time) {
2117 0 : for (int i = 0; i < kMaxSessionCollectors; i++) {
2118 0 : collectors[i]->set_expiry_time(time);
2119 : }
2120 0 : }
2121 :
2122 0 : int SessionStatsCollectorObject::GetExpiryTime() const {
2123 : /* Same expiry time would be configured for all the collectors. Pick value
2124 : * from any one of them */
2125 0 : return collectors[0]->expiry_time();
2126 : }
2127 :
2128 374 : SessionStatsCollector* SessionStatsCollectorObject::FlowToCollector
2129 : (const FlowEntry *flow) {
2130 374 : uint8_t idx = 0;
2131 374 : FlowTable *table = flow->flow_table();
2132 374 : if (table) {
2133 374 : idx = table->table_index() % kMaxSessionCollectors;
2134 : }
2135 374 : return collectors[idx].get();
2136 : }
2137 :
2138 2 : void SessionStatsCollectorObject::Shutdown() {
2139 4 : for (int i = 0; i < kMaxSessionCollectors; i++) {
2140 2 : collectors[i]->Shutdown();
2141 2 : collectors[i].reset();
2142 : }
2143 2 : }
2144 :
2145 0 : size_t SessionStatsCollectorObject::Size() const {
2146 0 : size_t size = 0;
2147 0 : for (int i = 0; i < kMaxSessionCollectors; i++) {
2148 0 : size += collectors[i]->Size();
2149 : }
2150 0 : return size;
2151 : }
2152 :
2153 2 : void SessionStatsCollectorObject::RegisterDBClients() {
2154 4 : for (int i = 0; i < kMaxSessionCollectors; i++) {
2155 2 : if (collectors[i].get()) {
2156 2 : collectors[i].get()->RegisterDBClients();
2157 : }
2158 : }
2159 2 : }
2160 :
2161 374 : FlowEntry* SessionStatsReq::reverse_flow() const {
2162 374 : FlowEntry *rflow = NULL;
2163 374 : if (flow_.get()) {
2164 374 : rflow = flow_->reverse_flow_entry();
2165 : }
2166 374 : return rflow;
2167 : }
2168 :
2169 119 : bool FlowToSessionMap::IsEqual(FlowToSessionMap &rhs) {
2170 119 : if (!(session_key_.IsEqual(rhs.session_key()))) {
2171 3 : return false;
2172 : }
2173 116 : if (!(session_agg_key_.IsEqual(rhs.session_agg_key()))) {
2174 0 : return false;
2175 : }
2176 116 : if (!(session_endpoint_key_.IsEqual(rhs.session_endpoint_key()))) {
2177 21 : return false;
2178 : }
2179 95 : return true;
2180 : }
2181 :
2182 0 : void SessionSloState::DeleteSessionSloStateRuleEntry(std::string uuid) {
2183 0 : SessionSloRuleStateMap::iterator it;
2184 0 : it = session_rule_state_map_.find(uuid);
2185 0 : if (it != session_rule_state_map_.end()) {
2186 0 : session_rule_state_map_.erase(it);
2187 : }
2188 0 : }
2189 :
2190 0 : void SessionSloState::UpdateSessionSloStateRuleEntry(std::string uuid,
2191 : int rate) {
2192 0 : SessionSloRuleStateMap::iterator it;
2193 0 : it = session_rule_state_map_.find(uuid);
2194 0 : if (it == session_rule_state_map_.end()) {
2195 0 : SessionSloRuleState slo_state_rule_entry = {};
2196 0 : slo_state_rule_entry.rate = rate;
2197 0 : session_rule_state_map_.insert(make_pair(uuid, slo_state_rule_entry));
2198 : } else {
2199 0 : SessionSloRuleState &prev = it->second;
2200 0 : if (prev.rate != rate) {
2201 0 : prev.rate = rate;
2202 0 : prev.ref_count = 0;
2203 : }
2204 : }
2205 :
2206 0 : }
2207 :
2208 0 : bool SessionSloState::UpdateSessionSloStateRuleRefCount(
2209 : const std::string &uuid,
2210 : bool *match) {
2211 0 : SessionSloRuleStateMap::iterator it;
2212 0 : bool is_logged = false;
2213 0 : it = session_rule_state_map_.find(uuid);
2214 0 : if (it != session_rule_state_map_.end()) {
2215 0 : *match = true;
2216 0 : if (it->second.ref_count == 0) {
2217 0 : is_logged = true;
2218 : }
2219 0 : it->second.ref_count++;
2220 0 : if (it->second.ref_count == it->second.rate) {
2221 0 : it->second.ref_count = 0;
2222 : }
2223 : }
2224 0 : return is_logged;
2225 : }
|
