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