Line data Source code
1 : /*
2 : * Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
3 : */
4 :
5 : #include <utility>
6 :
7 : #include <boost/uuid/uuid.hpp>
8 : #include <boost/uuid/uuid_io.hpp>
9 :
10 : #include <db/db.h>
11 : #include <base/util.h>
12 : #include <base/string_util.h>
13 :
14 : #include <cmn/agent_cmn.h>
15 : #include <init/agent_param.h>
16 : #include <boost/functional/factory.hpp>
17 : #include <cmn/agent_factory.h>
18 : #include <oper/interface_common.h>
19 : #include <oper/mirror_table.h>
20 : #include <oper/global_vrouter.h>
21 :
22 : #include <ksync/ksync_index.h>
23 : #include <ksync/ksync_entry.h>
24 : #include <ksync/ksync_object.h>
25 : #include <ksync/ksync_netlink.h>
26 : #include <ksync/ksync_sock.h>
27 : #include <uve/agent_uve.h>
28 : #include <vrouter/flow_stats/flow_stats_collector.h>
29 : #include <uve/vn_uve_table.h>
30 : #include <uve/vm_uve_table.h>
31 : #include <uve/interface_uve_stats_table.h>
32 : #include <uve/vrouter_uve_entry.h>
33 : #include <algorithm>
34 : #include <pkt/flow_proto.h>
35 : #include <pkt/flow_mgmt.h>
36 : #include <vrouter/ksync/ksync_init.h>
37 : #include <vrouter/flow_stats/flow_stats_types.h>
38 :
39 : bool flow_ageing_debug_ = false;
40 6 : FlowStatsCollector::FlowStatsCollector(boost::asio::io_context &io, int intvl,
41 : uint32_t flow_cache_timeout,
42 : AgentUveBase *uve,
43 : uint32_t instance_id,
44 : FlowAgingTableKey *key,
45 : FlowStatsManager *aging_module,
46 6 : FlowStatsCollectorObject *obj) :
47 : StatsCollector(TaskScheduler::GetInstance()->GetTaskId
48 : (kTaskFlowStatsCollector), instance_id,
49 : io, kFlowStatsTimerInterval, "Flow stats collector"),
50 6 : agent_uve_(uve),
51 6 : task_id_(uve->agent()->task_scheduler()->GetTaskId
52 6 : (kTaskFlowStatsCollector)),
53 : rand_gen_(boost::uuids::random_generator()),
54 6 : flow_iteration_key_(NULL),
55 6 : entries_to_visit_(0),
56 6 : flow_tcp_syn_age_time_(FlowTcpSynAgeTime),
57 6 : retry_delete_(true),
58 6 : request_queue_(agent_uve_->agent()->task_scheduler()->
59 : GetTaskId(kTaskFlowStatsCollector),
60 : instance_id,
61 : boost::bind(&FlowStatsCollector::RequestHandler,
62 : this, _1)),
63 6 : flow_aging_key_(*key), instance_id_(instance_id),
64 6 : flow_stats_manager_(aging_module), parent_(obj), ageing_task_(NULL),
65 18 : current_time_(GetCurrentTime()), ageing_task_starts_(0) {
66 6 : if (flow_cache_timeout) {
67 : // Convert to usec
68 2 : flow_age_time_intvl_ = 1000000L * (uint64_t)flow_cache_timeout;
69 : } else {
70 4 : flow_age_time_intvl_ = FlowAgeTime;
71 : }
72 6 : deleted_ = false;
73 6 : request_queue_.set_name("Flow stats collector");
74 : request_queue_.set_measure_busy_time
75 6 : (agent_uve_->agent()->MeasureQueueDelay());
76 : request_queue_.SetEntryCallback
77 6 : (boost::bind(&FlowStatsCollector::RequestHandlerEntry, this));
78 : request_queue_.SetExitCallback
79 6 : (boost::bind(&FlowStatsCollector::RequestHandlerExit, this, _1));
80 : // Aging timer fires every kFlowStatsTimerInterval msec. Compute
81 : // number of timer fires needed to scan complete table
82 6 : timers_per_scan_ = TimersPerScan();
83 6 : InitDone();
84 6 : }
85 :
86 12 : FlowStatsCollector::~FlowStatsCollector() {
87 6 : flow_stats_manager_->FreeIndex(instance_id_);
88 12 : }
89 :
90 0 : boost::uuids::uuid FlowStatsCollector::rand_gen() {
91 0 : return rand_gen_();
92 : }
93 :
94 212 : uint64_t FlowStatsCollector::GetCurrentTime() {
95 212 : return UTCTimestampUsec();
96 : }
97 :
98 6 : void FlowStatsCollector::Shutdown() {
99 6 : assert(ageing_task_ == NULL);
100 6 : StatsCollector::Shutdown();
101 6 : request_queue_.Shutdown();
102 6 : }
103 :
104 : // We want to scan the flow table every 25% of configured ageing time.
105 : // Compute number of timer fires needed to scan the flow-table once.
106 6 : uint32_t FlowStatsCollector::TimersPerScan() {
107 : uint64_t scan_time_millisec;
108 : /* Use Age Time itself as scan-time for flows */
109 :
110 : // Convert aging-time configured in micro-sec to millisecond
111 6 : scan_time_millisec = flow_age_time_intvl_ / 1000;
112 :
113 : // Compute time in which we must scan the complete table to honor the
114 : // kFlowScanTime
115 6 : scan_time_millisec = (scan_time_millisec * kFlowScanTime) / 100;
116 :
117 : // Enforce min value on scan-time
118 6 : if (scan_time_millisec < kFlowStatsTimerInterval) {
119 0 : scan_time_millisec = kFlowStatsTimerInterval;
120 : }
121 :
122 : // Number of timer fires needed to scan table once
123 6 : return scan_time_millisec / kFlowStatsTimerInterval;
124 : }
125 :
126 : // Update entries_to_visit_ based on total flows
127 : // Timer fires every kFlowScanTime. Its possible that we may not have visited
128 : // all entries by the time next timer fires. So, keep accumulating the number
129 : // of entries to visit into entries_to_visit_
130 : //
131 : // A lower-bound and an upper-bound are enforced on entries_to_visit_
132 0 : void FlowStatsCollector::UpdateEntriesToVisit() {
133 : // Compute number of flows to visit per scan-time
134 0 : uint32_t count = flow_export_info_list_.size();
135 0 : uint32_t entries = count / timers_per_scan_;
136 :
137 : // Update number of entries to visit in flow.
138 : // The scan for previous timer may still be in progress. So, accmulate
139 : // number of entries to visit
140 0 : entries_to_visit_ += entries;
141 :
142 : // Cap number of entries to visit to 25% of table
143 0 : if (entries_to_visit_ > ((count * kFlowScanTime)/100))
144 0 : entries_to_visit_ = (count * kFlowScanTime)/100;
145 :
146 : // Apply lower-limit
147 0 : if (entries_to_visit_ < kMinFlowsPerTimer)
148 0 : entries_to_visit_ = kMinFlowsPerTimer;
149 :
150 0 : return;
151 : }
152 :
153 0 : bool FlowStatsCollector::ShouldBeAged(FlowExportInfo *info,
154 : const vr_flow_entry *k_flow,
155 : const vr_flow_stats &k_stats,
156 : uint64_t curr_time) {
157 0 : FlowEntry *flow = info->flow();
158 : //If both forward and reverse flow are marked
159 : //as TCP closed then immediately remote the flow
160 0 : if (k_flow != NULL) {
161 : uint64_t k_flow_bytes, bytes;
162 0 : k_flow_bytes = GetFlowStats(k_stats.flow_bytes_oflow,
163 0 : k_stats.flow_bytes);
164 0 : bytes = 0x0000ffffffffffffULL & info->bytes();
165 : /* Don't account for agent overflow bits while comparing change in
166 : * stats */
167 0 : if (bytes < k_flow_bytes) {
168 0 : return false;
169 : }
170 : }
171 :
172 0 : uint64_t diff_time = curr_time - info->last_modified_time();
173 0 : if (diff_time < flow_age_time_intvl()) {
174 0 : return false;
175 : }
176 :
177 0 : if (flow->is_flags_set(FlowEntry::BgpRouterService)) {
178 0 : return false;
179 : }
180 :
181 0 : return true;
182 : }
183 :
184 144 : uint64_t FlowStatsCollector::GetFlowStats(const uint16_t &oflow_data,
185 : const uint32_t &data) {
186 144 : uint64_t flow_stats = (uint64_t) oflow_data << (sizeof(uint32_t) * 8);
187 144 : flow_stats |= data;
188 144 : return flow_stats;
189 : }
190 :
191 19 : uint64_t FlowStatsCollector::GetUpdatedFlowBytes(const FlowExportInfo *stats,
192 : uint64_t k_flow_bytes) {
193 19 : uint64_t oflow_bytes = 0xffff000000000000ULL & stats->bytes();
194 19 : uint64_t old_bytes = 0x0000ffffffffffffULL & stats->bytes();
195 19 : if (old_bytes > k_flow_bytes) {
196 0 : oflow_bytes += 0x0001000000000000ULL;
197 : }
198 19 : return (oflow_bytes |= k_flow_bytes);
199 : }
200 :
201 19 : uint64_t FlowStatsCollector::GetUpdatedFlowPackets(const FlowExportInfo *stats,
202 : uint64_t k_flow_pkts) {
203 19 : uint64_t oflow_pkts = 0xffffff0000000000ULL & stats->packets();
204 19 : uint64_t old_pkts = 0x000000ffffffffffULL & stats->packets();
205 19 : if (old_pkts > k_flow_pkts) {
206 0 : oflow_pkts += 0x0000010000000000ULL;
207 : }
208 19 : return (oflow_pkts |= k_flow_pkts);
209 : }
210 :
211 19 : void FlowStatsCollector::UpdateFloatingIpStats(const FlowExportInfo *flow,
212 : uint64_t bytes, uint64_t pkts) {
213 19 : InterfaceUveTable::FipInfo fip_info;
214 19 : FlowEntry *fe = flow->flow();
215 19 : if (!fe) {
216 0 : return;
217 : }
218 :
219 : /* Ignore Non-Floating-IP flow */
220 19 : if (!fe->fip() || fe->fip_vmi().uuid_ == boost::uuids::nil_uuid()) {
221 19 : return;
222 : }
223 :
224 : InterfaceUveStatsTable *table = static_cast<InterfaceUveStatsTable *>
225 0 : (agent_uve_->interface_uve_table());
226 :
227 0 : fip_info.bytes_ = bytes;
228 0 : fip_info.packets_ = pkts;
229 0 : fip_info.fip_ = fe->fip();
230 0 : fip_info.fip_vmi_ = fe->fip_vmi();
231 0 : fip_info.is_local_flow_ = fe->is_flags_set(FlowEntry::LocalFlow);
232 0 : fip_info.is_ingress_flow_ = fe->is_flags_set(FlowEntry::IngressDir);
233 0 : fip_info.is_reverse_flow_ = fe->is_flags_set(FlowEntry::ReverseFlow);
234 0 : fip_info.vn_ = fe->data().source_vn_match;
235 :
236 0 : fip_info.rev_fip_ = NULL;
237 0 : if (fe->fip() != ReverseFlowFip(flow)) {
238 : /* This is the case where Source and Destination VMs (part of
239 : * same compute node) ping to each other to their respective
240 : * Floating IPs. In this case for each flow we need to increment
241 : * stats for both the VMs */
242 0 : fip_info.rev_fip_ = ReverseFlowFipEntry(flow);
243 : }
244 :
245 0 : table->UpdateFloatingIpStats(fip_info);
246 19 : }
247 :
248 0 : InterfaceUveTable::FloatingIp *FlowStatsCollector::ReverseFlowFipEntry
249 : (const FlowExportInfo *flow) {
250 0 : uint32_t fip = ReverseFlowFip(flow);
251 0 : VmInterfaceKey vmi = ReverseFlowFipVmi(flow);
252 0 : Interface *intf = dynamic_cast<Interface *>
253 0 : (agent_uve_->agent()->interface_table()->FindActiveEntry(&vmi));
254 :
255 0 : if (intf) {
256 : InterfaceUveStatsTable *table = static_cast<InterfaceUveStatsTable *>
257 0 : (agent_uve_->interface_uve_table());
258 0 : const string &vn = flow->flow()->data().source_vn_match;
259 0 : return table->FipEntry(fip, vn, intf);
260 : }
261 0 : return NULL;
262 0 : }
263 :
264 0 : uint32_t FlowStatsCollector::ReverseFlowFip(const FlowExportInfo *info) {
265 0 : FlowEntry *rflow = info->reverse_flow();
266 0 : if (rflow) {
267 0 : return rflow->fip();
268 : }
269 0 : return 0;
270 : }
271 :
272 0 : VmInterfaceKey FlowStatsCollector::ReverseFlowFipVmi
273 : (const FlowExportInfo *info)
274 : {
275 0 : FlowEntry *rflow = info->reverse_flow();
276 0 : if (rflow) {
277 0 : return rflow->fip_vmi();
278 : }
279 : return VmInterfaceKey(
280 0 : AgentKey::ADD_DEL_CHANGE, boost::uuids::nil_uuid(), "");
281 : }
282 :
283 19 : void FlowStatsCollector::UpdateVmiTagBasedStats(FlowExportInfo *info,
284 : uint64_t bytes, uint64_t pkts) {
285 19 : FlowEntry *flow = info->flow();
286 :
287 19 : const Interface *itf = flow->intf_entry();
288 19 : if (!itf) {
289 0 : return;
290 : }
291 19 : if (itf->type() != Interface::VM_INTERFACE) {
292 0 : return;
293 : }
294 19 : const VmInterface *vmi = static_cast<const VmInterface *>(itf);
295 19 : const string &src_vn = flow->data().source_vn_match;
296 19 : const string &dst_vn = flow->data().dest_vn_match;
297 :
298 : /* Ignore flows for which source VN or destination VN are not known */
299 19 : if (!src_vn.length() || !dst_vn.length()) {
300 0 : return;
301 : }
302 :
303 : InterfaceUveStatsTable *itf_table = static_cast<InterfaceUveStatsTable *>
304 19 : (agent_uve_->interface_uve_table());
305 19 : EndpointStatsInfo ep;
306 19 : ep.vmi = vmi;
307 19 : ep.local_tagset = flow->local_tagset();
308 19 : ep.remote_tagset = flow->remote_tagset();
309 19 : ep.remote_prefix = flow->RemotePrefix();
310 19 : ep.policy = flow->fw_policy_name_uuid();
311 19 : ep.diff_bytes = bytes;
312 19 : ep.diff_pkts = pkts;
313 19 : FlowTable::GetFlowSandeshActionParams(flow->data().match_p.action_info,
314 : ep.action);
315 19 : if (flow->is_flags_set(FlowEntry::LocalFlow)) {
316 : /* When VM A talks to VM B which is in different compute nodes, the
317 : * following flows are created
318 : * (1) A-B, Ingress, Forward, pol1
319 : * (2) B-A, Egress, Reverse, pol1
320 : * (3) A-B, Egress, Forward, pol2
321 : * (4) B-A, Inress, Reverse, pol2
322 : * When both A and B are in single compute, we have only the following
323 : * flows (Flows marked as LocalFlow)
324 : * (1) A-B, Ingress, Forward, pol1
325 : * (2) B-A, Inress, Reverse, pol2
326 : * To simulate session stats similar to case where VMs are in different
327 : * computes, for local flows, we do the following.
328 : * (a) when "A-B, Ingress, Forward, pol1" flow is seen, we also
329 : * update stats for "A-B, Egress, Forward, pol2". This is because
330 : * diff stats for "A-B, Ingress, Forward, pol1" and
331 : * "A-B, Egress, Forward, pol2" are same. Policy for implicit flow
332 : * is picked from reverse flow
333 : * (b) when "B-A, Ingress, Reverse, pol2" flow is seen, we also
334 : * update stats for "B-A, Egress, Reverse, pol1". This is because diff
335 : * stats for "B-A, Ingress, Reverse, pol2" and
336 : * "B-A, Egress, Reverse, pol1" is same. Policy for implicit flow is
337 : * picked from reverse flow
338 : */
339 5 : ep.local_vn = src_vn;
340 5 : ep.remote_vn = dst_vn;
341 5 : ep.in_stats = true;
342 : bool egress_flow_is_client;
343 5 : if (flow->is_flags_set(FlowEntry::ReverseFlow)) {
344 2 : ep.client = false;
345 2 : egress_flow_is_client = true;
346 : } else {
347 3 : ep.client = true;
348 3 : egress_flow_is_client = false;
349 : }
350 5 : itf_table->UpdateVmiTagBasedStats(ep);
351 :
352 : /* Local flows will not have egress flows in the system. So we need to
353 : * explicitly build stats for egress flow using the data available from
354 : * ingress flow. Egress flow stats has to be updated on destination
355 : * VMI. We skip updation if we are unable to pick destination VMI from
356 : * reverse flow. */
357 :
358 5 : FlowEntry* rflow = info->reverse_flow();
359 5 : if (rflow) {
360 0 : const Interface *ritf = rflow->intf_entry();
361 0 : if (ritf && (ritf->type() == Interface::VM_INTERFACE)) {
362 0 : ep.local_tagset = flow->remote_tagset();
363 0 : ep.remote_tagset = flow->local_tagset();
364 0 : ep.local_vn = dst_vn;
365 0 : ep.remote_vn = src_vn;
366 0 : ep.policy = rflow->fw_policy_name_uuid();
367 0 : ep.client = egress_flow_is_client;
368 0 : ep.vmi = static_cast<const VmInterface *>(ritf);
369 0 : ep.in_stats = false;
370 0 : itf_table->UpdateVmiTagBasedStats(ep);
371 : }
372 : }
373 : } else {
374 14 : if (flow->is_flags_set(FlowEntry::IngressDir)) {
375 6 : ep.local_vn = src_vn;
376 6 : ep.remote_vn = dst_vn;
377 6 : ep.in_stats = true;
378 6 : if (flow->is_flags_set(FlowEntry::ReverseFlow)) {
379 5 : ep.client = false;
380 : } else {
381 1 : ep.client = true;
382 : }
383 : } else {
384 8 : ep.local_vn = dst_vn;
385 8 : ep.remote_vn = src_vn;
386 8 : ep.in_stats = false;
387 8 : if (flow->is_flags_set(FlowEntry::ReverseFlow)) {
388 1 : ep.client = true;
389 : } else {
390 7 : ep.client = false;
391 : }
392 : }
393 14 : itf_table->UpdateVmiTagBasedStats(ep);
394 : }
395 19 : }
396 :
397 19 : void FlowStatsCollector::UpdateInterVnStats(FlowExportInfo *info,
398 : uint64_t bytes, uint64_t pkts) {
399 19 : FlowEntry *flow = info->flow();
400 19 : string src_vn = flow->data().source_vn_match;
401 19 : string dst_vn = flow->data().dest_vn_match;
402 : VnUveTable *vn_table = static_cast<VnUveTable *>
403 19 : (agent_uve_->vn_uve_table());
404 :
405 19 : if (!src_vn.length())
406 0 : src_vn = FlowHandler::UnknownVn();
407 19 : if (!dst_vn.length())
408 0 : dst_vn = FlowHandler::UnknownVn();
409 :
410 : /* When packet is going from src_vn to dst_vn it should be interpreted
411 : * as ingress to vrouter and hence in-stats for src_vn w.r.t. dst_vn
412 : * should be incremented. Similarly when the packet is egressing vrouter
413 : * it should be considered as out-stats for dst_vn w.r.t. src_vn.
414 : * Here the direction "in" and "out" should be interpreted w.r.t vrouter
415 : */
416 19 : if (flow->is_flags_set(FlowEntry::LocalFlow)) {
417 5 : vn_table->UpdateInterVnStats(src_vn, dst_vn, bytes, pkts, false);
418 5 : vn_table->UpdateInterVnStats(dst_vn, src_vn, bytes, pkts, true);
419 : } else {
420 14 : if (flow->is_flags_set(FlowEntry::IngressDir)) {
421 6 : vn_table->UpdateInterVnStats(src_vn, dst_vn, bytes, pkts, false);
422 : } else {
423 8 : vn_table->UpdateInterVnStats(dst_vn, src_vn, bytes, pkts, true);
424 : }
425 : }
426 19 : }
427 :
428 44 : void FlowStatsCollector::UpdateFlowStats(FlowExportInfo *info,
429 : uint64_t teardown_time) {
430 44 : if (!info) {
431 19 : return;
432 : }
433 44 : FlowEntry *fe = info->flow();
434 44 : KSyncFlowMemory *ksync_obj = agent_uve_->agent()->ksync()->
435 44 : ksync_flow_memory();
436 : /* Fetch vrouter Flow entry using gen_id and flow_handle from FlowExportInfo
437 : * to account for the case where FlowEntry's flow_handle/gen_id has changed
438 : * during Delete processing by FlowStatsCollector */
439 : vr_flow_stats k_stats;
440 44 : const vr_flow_entry *k_flow = ksync_obj->GetKFlowStats(fe->key(),
441 : info->flow_handle(),
442 44 : info->gen_id(),
443 : &k_stats);
444 44 : if (k_flow) {
445 19 : UpdateFlowStatsInternal(info, k_stats.flow_bytes,
446 19 : k_stats.flow_bytes_oflow,
447 : k_stats.flow_packets,
448 19 : k_stats.flow_packets_oflow,
449 : teardown_time, true);
450 19 : return;
451 : }
452 : }
453 :
454 0 : void FlowStatsCollector::FlowDeleteEnqueue(FlowExportInfo *info, uint64_t t) {
455 0 : flows_aged_++;
456 0 : FlowEntry *fe = info->flow();
457 0 : agent_uve_->agent()->pkt()->get_flow_proto()->DeleteFlowRequest(fe);
458 0 : info->set_delete_enqueue_time(t);
459 0 : FlowEntry *rflow = info->reverse_flow();
460 0 : if (rflow) {
461 0 : FlowExportInfo *rev_info = FindFlowExportInfo(rflow);
462 0 : if (rev_info) {
463 0 : rev_info->set_delete_enqueue_time(t);
464 : }
465 : }
466 0 : }
467 :
468 0 : void FlowStatsCollector::FlowEvictEnqueue(FlowExportInfo *info, uint64_t t,
469 : uint32_t flow_handle,
470 : uint16_t gen_id) {
471 0 : flows_evicted_++;
472 0 : FlowEntry *fe = info->flow();
473 0 : agent_uve_->agent()->pkt()->get_flow_proto()->EvictFlowRequest
474 0 : (fe, flow_handle, gen_id, (gen_id + 1));
475 0 : info->set_evict_enqueue_time(t);
476 0 : }
477 :
478 0 : void FlowStatsCollector::UpdateFlowStatsInternalLocked(FlowExportInfo *info,
479 : uint32_t bytes,
480 : uint16_t oflow_bytes,
481 : uint32_t pkts,
482 : uint16_t oflow_pkts,
483 : uint64_t time,
484 : bool teardown_time) {
485 0 : FlowEntry *flow = info->flow();
486 0 : FlowEntry *rflow = info->reverse_flow();
487 0 : FLOW_LOCK(flow, rflow, FlowEvent::FLOW_MESSAGE);
488 0 : UpdateFlowStatsInternal(info, bytes, oflow_bytes, pkts, oflow_pkts, time,
489 : teardown_time);
490 0 : }
491 :
492 19 : void FlowStatsCollector::UpdateFlowStatsInternal(FlowExportInfo *info,
493 : uint32_t bytes,
494 : uint16_t oflow_bytes,
495 : uint32_t pkts,
496 : uint16_t oflow_pkts,
497 : uint64_t time,
498 : bool teardown_time) {
499 : uint64_t k_bytes, k_packets, total_bytes, total_packets;
500 19 : k_bytes = GetFlowStats(oflow_bytes, bytes);
501 19 : k_packets = GetFlowStats(oflow_pkts, pkts);
502 :
503 19 : total_bytes = GetUpdatedFlowBytes(info, k_bytes);
504 19 : total_packets = GetUpdatedFlowPackets(info, k_packets);
505 19 : uint64_t diff_bytes = total_bytes - info->bytes();
506 19 : uint64_t diff_pkts = total_packets - info->packets();
507 19 : info->set_bytes(total_bytes);
508 19 : info->set_packets(total_packets);
509 19 : if (teardown_time) {
510 19 : info->set_teardown_time(time);
511 : } else {
512 0 : info->set_last_modified_time(time);
513 : }
514 :
515 : /* In TSN mode, we don't export flows or statistics based on flows */
516 19 : if (agent_uve_->agent()->tsn_enabled()) {
517 0 : return;
518 : }
519 : //Update Inter-VN stats
520 19 : UpdateInterVnStats(info, diff_bytes, diff_pkts);
521 : //Update Endpoint stats
522 19 : UpdateVmiTagBasedStats(info, diff_bytes, diff_pkts);
523 : //Update Floating-IP stats
524 19 : UpdateFloatingIpStats(info, diff_bytes, diff_pkts);
525 : }
526 :
527 : // Check if flow needs to be evicted
528 0 : bool FlowStatsCollector::EvictFlow(KSyncFlowMemory *ksync_obj,
529 : const vr_flow_entry *k_flow,
530 : uint16_t k_flow_flags,
531 : uint32_t flow_handle, uint16_t gen_id,
532 : FlowExportInfo *info, uint64_t curr_time) {
533 0 : FlowEntry *fe = info->flow();
534 :
535 0 : if ((fe->key().protocol != IPPROTO_TCP))
536 0 : return false;
537 :
538 0 : if (ksync_obj->IsEvictionMarked(k_flow, k_flow_flags) == false)
539 0 : return false;
540 :
541 : // Flow evict already enqueued? Re-Enqueue request after retry-time
542 0 : uint64_t evict_time = info->evict_enqueue_time();
543 0 : if (evict_time) {
544 0 : if ((curr_time - evict_time) > kFlowDeleteRetryTime) {
545 0 : FlowEvictEnqueue(info, curr_time, flow_handle, gen_id);
546 : }
547 : } else {
548 0 : FlowEvictEnqueue(info, curr_time, flow_handle, gen_id);
549 : }
550 :
551 0 : return true;
552 : }
553 :
554 0 : bool FlowStatsCollector::AgeFlow(KSyncFlowMemory *ksync_obj,
555 : const vr_flow_entry *k_flow,
556 : const vr_flow_stats &k_stats,
557 : const KFlowData& kinfo,
558 : FlowExportInfo *info, uint64_t curr_time) {
559 0 : FlowEntry *fe = info->flow();
560 0 : FlowEntry *rfe = info->reverse_flow();
561 :
562 : // if we come across deleted entry, retry flow deletion after some time
563 : // duplicate delete will be suppressed in flow_table
564 0 : uint64_t delete_time = info->delete_enqueue_time();
565 0 : if (delete_time) {
566 0 : if ((curr_time - delete_time) > kFlowDeleteRetryTime) {
567 0 : FlowDeleteEnqueue(info, curr_time);
568 : }
569 0 : return true;
570 : }
571 :
572 : // Delete short flows
573 0 : if ((flow_stats_manager_->delete_short_flow() == true) &&
574 0 : fe->is_flags_set(FlowEntry::ShortFlow)) {
575 0 : FlowDeleteEnqueue(info, curr_time);
576 0 : return true;
577 : }
578 :
579 0 : bool deleted = false;
580 0 : FlowExportInfo *rev_info = NULL;
581 : // Can the flow be aged?
582 0 : if (ShouldBeAged(info, k_flow, k_stats, curr_time)) {
583 0 : rev_info = FindFlowExportInfo(rfe);
584 : // ShouldBeAged looks at one flow only. So, check for both forward and
585 : // reverse flows
586 0 : if (rev_info) {
587 0 : const vr_flow_entry *k_flow_rev = NULL;
588 : vr_flow_stats k_rflow_stats;
589 0 : k_flow_rev = ksync_obj->GetKFlowStats(rfe->key(),
590 : rev_info->flow_handle(),
591 0 : rev_info->gen_id(),
592 : &k_rflow_stats);
593 0 : if (ShouldBeAged(rev_info, k_flow_rev, k_rflow_stats, curr_time)) {
594 0 : deleted = true;
595 : }
596 : } else {
597 0 : deleted = true;
598 : }
599 : }
600 :
601 0 : if (deleted == true) {
602 0 : FlowDeleteEnqueue(info, curr_time);
603 : }
604 :
605 : // Update stats for flows not being deleted
606 : // Stats for deleted flow are updated when we get DELETE message
607 0 : if (deleted == false && k_flow) {
608 : uint64_t k_bytes, bytes;
609 :
610 0 : k_bytes = GetFlowStats(k_stats.flow_bytes_oflow,
611 0 : k_stats.flow_bytes);
612 0 : bytes = 0x0000ffffffffffffULL & info->bytes();
613 : /* Don't account for agent overflow bits while comparing change in
614 : * stats */
615 0 : if (bytes != k_bytes) {
616 0 : UpdateFlowStatsInternalLocked(info,
617 0 : k_stats.flow_bytes,
618 0 : k_stats.flow_bytes_oflow,
619 0 : k_stats.flow_packets,
620 0 : k_stats.flow_packets_oflow,
621 : curr_time, false);
622 : }
623 : }
624 0 : return deleted;
625 : }
626 :
627 : // Check if a flow is to be aged or evicted. Returns number of flows visited
628 0 : uint32_t FlowStatsCollector::ProcessFlow(FlowExportInfoList::iterator &it,
629 : KSyncFlowMemory *ksync_obj,
630 : FlowExportInfo *info,
631 : uint64_t curr_time) {
632 0 : uint32_t count = 1;
633 0 : FlowEntry *fe = info->flow();
634 : /* Use flow-handle and gen-id from FlowExportInfo instead of FlowEntry.
635 : * The stats that FlowExportInfo holds corresponds to a given
636 : * (FlowKey, gen-id and FlowHandle). Since gen-id/flow-handle for a flow
637 : * can change dynamically, we need to pick gen-id and flow-handle from
638 : * FlowExportInfo. Otherwise stats will go wrong. Whenever gen-id/
639 : * flow-handle changes, the stats will be reset as part of AddFlow API
640 : */
641 0 : uint32_t flow_handle = info->flow_handle();
642 0 : uint16_t gen_id = info->gen_id();
643 :
644 : /* If Flow handle is still not populated in FlowStatsCollector, pick the
645 : * value from FlowEntry
646 : */
647 0 : if (flow_handle == FlowEntry::kInvalidFlowHandle) {
648 : {
649 0 : FlowEntry *rflow = NULL;
650 0 : FLOW_LOCK(fe, rflow, FlowEvent::FLOW_MESSAGE);
651 : // since flow processing and stats collector can run in parallel
652 : // flow handle and gen id not being the key for flow entry can
653 : // change while processing, so flow handle and gen id should be
654 : // fetched by holding an lock.
655 0 : flow_handle = fe->flow_handle();
656 0 : gen_id = fe->gen_id();
657 0 : info->CopyFlowInfo(fe);
658 0 : }
659 : }
660 0 : const vr_flow_entry *k_flow = NULL;
661 : vr_flow_stats k_stats;
662 : KFlowData kinfo;
663 :
664 : /* Teardown time is set when Evicted flow stats update message is received.
665 : * For flows whose teardown time is set, we need not read stats from
666 : * vrouter
667 : */
668 0 : if (!info->teardown_time()) {
669 0 : k_flow = ksync_obj->GetKFlowStatsAndInfo(fe->key(), flow_handle,
670 : gen_id, &k_stats, &kinfo);
671 :
672 : // Flow evicted?
673 0 : if (EvictFlow(ksync_obj, k_flow, kinfo.flags, flow_handle, gen_id,
674 0 : info, curr_time) == true) {
675 : // If retry_delete_ enabled, dont change flow_export_info_list_
676 0 : if (retry_delete_ == true)
677 0 : return count;
678 :
679 : // We dont want to retry delete-events, remove flow from ageing list
680 0 : assert(info->is_linked());
681 : FlowExportInfoList::iterator flow_it =
682 0 : flow_export_info_list_.iterator_to(*info);
683 0 : flow_export_info_list_.erase(flow_it);
684 :
685 0 : return count;
686 : }
687 : }
688 :
689 :
690 : // Flow aged?
691 0 : if (AgeFlow(ksync_obj, k_flow, k_stats, kinfo, info, curr_time) == false)
692 0 : return count;
693 :
694 : // If retry_delete_ enabled, dont change flow_export_info_list_
695 0 : if (retry_delete_ == false)
696 0 : return count;
697 :
698 : // Flow aged, remove both forward and reverse flow
699 0 : assert(info->is_linked());
700 : FlowExportInfoList::iterator flow_it =
701 0 : flow_export_info_list_.iterator_to(*info);
702 0 : flow_export_info_list_.erase(flow_it);
703 :
704 0 : FlowEntry *rfe = info->reverse_flow();
705 0 : FlowExportInfo *rev_info = FindFlowExportInfo(rfe);
706 0 : if (rev_info) {
707 0 : if (rev_info->is_linked()) {
708 : FlowExportInfoList::iterator rev_flow_it =
709 0 : flow_export_info_list_.iterator_to(*rev_info);
710 0 : if (rev_flow_it == it) {
711 0 : it++;
712 : }
713 0 : flow_export_info_list_.erase(rev_flow_it);
714 : }
715 0 : count++;
716 : }
717 0 : return count;
718 : }
719 :
720 0 : uint32_t FlowStatsCollector::RunAgeing(uint32_t max_count) {
721 : FlowExportInfoList::iterator it;
722 0 : if (flow_iteration_key_ == NULL) {
723 0 : it = flow_export_info_list_.begin();
724 : } else {
725 0 : FlowEntryTree::iterator tree_it = flow_tree_.find(flow_iteration_key_);
726 : // Flow to iterate next is not found. Force stop this iteration.
727 : // We will continue from begining on next timer
728 0 : if (tree_it == flow_tree_.end()) {
729 0 : flow_iteration_key_ = NULL;
730 0 : return entries_to_visit_;
731 : }
732 0 : it = flow_export_info_list_.iterator_to(tree_it->second);
733 : }
734 :
735 0 : KSyncFlowMemory *ksync_obj = agent_uve_->agent()->ksync()->
736 0 : ksync_flow_memory();
737 0 : uint64_t curr_time = GetCurrentTime();
738 0 : uint32_t count = 0;
739 0 : while (count < max_count) {
740 0 : if (it == flow_export_info_list_.end()) {
741 0 : break;
742 : }
743 :
744 0 : FlowExportInfo *info = &(*it);
745 0 : it++;
746 0 : flows_visited_++;
747 0 : count += ProcessFlow(it, ksync_obj, info, curr_time);
748 : }
749 :
750 : // Update iterator for next pass
751 0 : if (it == flow_export_info_list_.end()) {
752 0 : flow_iteration_key_ = NULL;
753 : } else {
754 0 : flow_iteration_key_ = it->flow();
755 : }
756 :
757 0 : return count;
758 : }
759 :
760 : // Timer fired for ageing. Update the number of entries to visit and start the
761 : // task if its already not ruuning
762 4 : bool FlowStatsCollector::Run() {
763 4 : if (flow_tree_.size() == 0) {
764 4 : return true;
765 : }
766 :
767 : // Update number of entries to visit in flow.
768 0 : UpdateEntriesToVisit();
769 :
770 : // Start task to scan the entries
771 0 : if (ageing_task_ == NULL) {
772 0 : ageing_task_starts_++;
773 :
774 0 : if (flow_ageing_debug_) {
775 0 : LOG(DEBUG,
776 : UTCUsecToString(ClockMonotonicUsec())
777 : << " AgeingTasks Num " << ageing_task_starts_
778 : << " Request count " << request_queue_.Length()
779 : << " Tree size " << flow_tree_.size()
780 : << " List size " << flow_export_info_list_.size()
781 : << " flows visited " << flows_visited_
782 : << " flows aged " << flows_aged_
783 : << " flows evicted " << flows_evicted_);
784 : }
785 0 : flows_visited_ = 0;
786 0 : flows_aged_ = 0;
787 0 : flows_evicted_ = 0;
788 0 : ageing_task_ = new AgeingTask(this);
789 0 : agent_uve_->agent()->task_scheduler()->Enqueue(ageing_task_);
790 : }
791 0 : return true;
792 : }
793 :
794 0 : bool FlowStatsCollector::RunAgeingTask() {
795 : // Run ageing per task
796 0 : uint32_t count = RunAgeing(kFlowsPerTask);
797 : // Update number of entries visited
798 0 : if (count < entries_to_visit_)
799 0 : entries_to_visit_ -= count;
800 : else
801 0 : entries_to_visit_ = 0;
802 : // Done with task if we reach end of tree or count is exceeded
803 0 : if (flow_iteration_key_ == NULL || entries_to_visit_ == 0) {
804 0 : entries_to_visit_ = 0;
805 0 : ageing_task_ = NULL;
806 0 : return true;
807 : }
808 :
809 : // More entries to visit. Continue the task
810 0 : return false;
811 : }
812 :
813 : /////////////////////////////////////////////////////////////////////////////
814 : // Utility methods to enqueue events into work-queue
815 : /////////////////////////////////////////////////////////////////////////////
816 100 : void FlowStatsCollector::AddEvent(const FlowEntryPtr &flow) {
817 100 : FlowExportInfo info(flow, GetCurrentTime());
818 : boost::shared_ptr<FlowExportReq>
819 100 : req(new FlowExportReq(FlowExportReq::ADD_FLOW, info));
820 100 : request_queue_.Enqueue(req);
821 100 : }
822 :
823 44 : void FlowStatsCollector::DeleteEvent(const FlowEntryPtr &flow,
824 : const RevFlowDepParams ¶ms) {
825 44 : FlowExportInfo info(flow);
826 : boost::shared_ptr<FlowExportReq>
827 : req(new FlowExportReq(FlowExportReq::DELETE_FLOW, info,
828 44 : GetCurrentTime(), params));
829 44 : request_queue_.Enqueue(req);
830 44 : }
831 :
832 0 : void FlowStatsCollector::UpdateStatsEvent(const FlowEntryPtr &flow,
833 : uint32_t bytes,
834 : uint32_t packets,
835 : uint32_t oflow_bytes,
836 : const boost::uuids::uuid &u) {
837 0 : FlowExportInfo info(flow);
838 : boost::shared_ptr<FlowExportReq>
839 : req(new FlowExportReq(FlowExportReq::UPDATE_FLOW_STATS, info, bytes,
840 0 : packets, oflow_bytes, u));
841 0 : request_queue_.Enqueue(req);
842 0 : }
843 :
844 62 : bool FlowStatsCollector::RequestHandlerEntry() {
845 62 : current_time_ = GetCurrentTime();
846 62 : return true;
847 : }
848 :
849 62 : void FlowStatsCollector::RequestHandlerExit(bool done) {
850 62 : }
851 :
852 144 : bool FlowStatsCollector::RequestHandler(boost::shared_ptr<FlowExportReq> req) {
853 144 : const FlowExportInfo &info = req->info();
854 144 : FlowEntry *flow = info.flow();
855 144 : FlowEntry *rflow = info.reverse_flow();
856 188 : FLOW_LOCK(flow, rflow, FlowEvent::FLOW_MESSAGE);
857 :
858 144 : switch (req->event()) {
859 100 : case FlowExportReq::ADD_FLOW: {
860 100 : AddFlow(req->info());
861 100 : break;
862 : }
863 :
864 44 : case FlowExportReq::DELETE_FLOW: {
865 44 : FlowEntryTree::iterator it;
866 : // Get the FlowExportInfo for flow
867 44 : if (FindFlowExportInfo(flow, it) == false)
868 0 : break;
869 :
870 : /* We don't export flows in TSN mode */
871 44 : if (agent_uve_->agent()->tsn_enabled() == false) {
872 44 : FlowExportInfo *info = &it->second;
873 : /* While updating stats for evicted flows, we set the teardown_time
874 : * and export the flow. So delete handling for evicted flows need
875 : * not update stats and export flow */
876 44 : if (!info->teardown_time()) {
877 44 : UpdateFlowStats(info, req->time());
878 : }
879 : }
880 : /* Remove the entry from our tree */
881 44 : DeleteFlow(it);
882 44 : break;
883 : }
884 :
885 0 : case FlowExportReq::UPDATE_FLOW_STATS: {
886 : /* We don't export flows in TSN mode */
887 0 : if (agent_uve_->agent()->tsn_enabled() == false) {
888 0 : EvictedFlowStatsUpdate(flow, req->bytes(), req->packets(),
889 0 : req->oflow_bytes(), req->uuid());
890 : }
891 0 : break;
892 : }
893 :
894 0 : default:
895 0 : assert(0);
896 : }
897 :
898 144 : if (deleted_ && parent_->CanDelete()) {
899 0 : flow_stats_manager_->Free(flow_aging_key_);
900 : }
901 :
902 144 : return true;
903 144 : }
904 :
905 : FlowExportInfo *
906 0 : FlowStatsCollector::FindFlowExportInfo(const FlowEntry *fe) {
907 0 : FlowEntryTree::iterator it = flow_tree_.find(fe);
908 0 : if (it == flow_tree_.end()) {
909 0 : return NULL;
910 : }
911 :
912 0 : return &it->second;
913 : }
914 :
915 : const FlowExportInfo *
916 0 : FlowStatsCollector::FindFlowExportInfo(const FlowEntry *fe) const {
917 0 : FlowEntryTree::const_iterator it = flow_tree_.find(fe);
918 0 : if (it == flow_tree_.end()) {
919 0 : return NULL;
920 : }
921 :
922 0 : return &it->second;
923 : }
924 :
925 44 : bool FlowStatsCollector::FindFlowExportInfo(const FlowEntry *fe,
926 : FlowEntryTree::iterator &it) {
927 44 : it = flow_tree_.find(fe);
928 44 : if (it == flow_tree_.end()) {
929 0 : return false;
930 : }
931 44 : return true;
932 : }
933 :
934 44 : void FlowStatsCollector::NewFlow(FlowEntry *flow) {
935 : /* In TSN mode, we don't export flows or statistics based on flows */
936 44 : if (agent_uve_->agent()->tsn_enabled()) {
937 0 : return;
938 : }
939 44 : const FlowKey &key = flow->key();
940 44 : uint8_t proto = key.protocol;
941 44 : uint16_t sport = key.src_port;
942 44 : uint16_t dport = key.dst_port;
943 :
944 : // Update vrouter port bitmap
945 : VrouterUveEntry *vre = static_cast<VrouterUveEntry *>(
946 44 : agent_uve_->vrouter_uve_entry());
947 44 : vre->UpdateBitmap(proto, sport, dport);
948 :
949 : // Update source-vn port bitmap
950 44 : VnUveTable *vnte = static_cast<VnUveTable *>(agent_uve_->vn_uve_table());
951 44 : vnte->UpdateBitmap(flow->data().source_vn_match, proto, sport, dport);
952 : // Update dest-vn port bitmap
953 44 : vnte->UpdateBitmap(flow->data().dest_vn_match, proto, sport, dport);
954 :
955 44 : const VmInterface *port = dynamic_cast<const VmInterface *>
956 44 : (flow->intf_entry());
957 44 : if (port == NULL) {
958 0 : return;
959 : }
960 44 : const VmEntry *vm = port->vm();
961 44 : if (vm == NULL) {
962 0 : return;
963 : }
964 :
965 : // update vm and interface (all interfaces of vm) bitmap
966 44 : VmUveTable *vmt = static_cast<VmUveTable *>(agent_uve_->vm_uve_table());
967 44 : vmt->UpdateBitmap(vm, proto, sport, dport);
968 : }
969 :
970 100 : void FlowStatsCollector::AddFlow(FlowExportInfo info) {
971 : /* Before inserting update the gen_id and flow_handle in FlowExportInfo.
972 : * Locks for accessing fields of flow are taken in calling function.
973 : */
974 100 : FlowEntry* fe = info.flow();
975 100 : info.CopyFlowInfo(fe);
976 : std::pair<FlowEntryTree::iterator, bool> ret =
977 100 : flow_tree_.insert(make_pair(fe, info));
978 100 : if (ret.second == false) {
979 56 : FlowExportInfo &prev = ret.first->second;
980 56 : if (prev.uuid() != fe->uuid()) {
981 : /* Received ADD request for already added entry with a different
982 : * UUID. Because of state-compression of messages to
983 : * FlowStatsCollector in FlowMgmt, we have not received DELETE for
984 : * previous UUID. Send FlowExport to indicate delete for the flow.
985 : * This export need not be sent if teardown time is already set.
986 : * Teardown time would be set if EvictedFlowStats update request
987 : * comes before this duplicate add.
988 : */
989 0 : if (!prev.teardown_time()) {
990 0 : UpdateFlowStats(&prev, info.last_modified_time());
991 : }
992 : /* After sending Delete to collector (if required), reset the stats
993 : */
994 0 : prev.ResetStats();
995 : }
996 56 : prev.CopyFlowInfo(fe);
997 56 : prev.set_delete_enqueue_time(0);
998 56 : prev.set_evict_enqueue_time(0);
999 56 : prev.set_teardown_time(0);
1000 : } else {
1001 44 : NewFlow(info.flow());
1002 : }
1003 200 : if (ret.first->second.is_linked() == false) {
1004 44 : flow_export_info_list_.push_back(ret.first->second);
1005 : }
1006 100 : }
1007 :
1008 : // The flow being deleted may be the first flow to visit in next ageing
1009 : // iteration. Update the flow to visit next in such case
1010 0 : void FlowStatsCollector::UpdateFlowIterationKey
1011 : (const FlowEntry *del_flow, FlowEntryTree::iterator &tree_it) {
1012 : // Flow not found in tree is not a valid scenario. Lets be safe and
1013 : // restart walk here
1014 0 : if (tree_it == flow_tree_.end()) {
1015 0 : flow_iteration_key_ = NULL;
1016 : }
1017 :
1018 0 : if (flow_iteration_key_ == NULL) {
1019 0 : return;
1020 : }
1021 :
1022 : // The flow to visit next for ageing is being deleted. Update next flow to
1023 : // visit
1024 : FlowExportInfoList::iterator it =
1025 0 : flow_export_info_list_.iterator_to(tree_it->second);
1026 : ++it;
1027 :
1028 : // If this is end of list, start from begining again
1029 0 : if (it == flow_export_info_list_.end())
1030 0 : it = flow_export_info_list_.begin();
1031 :
1032 0 : if (it == flow_export_info_list_.end()) {
1033 0 : flow_iteration_key_ = NULL;
1034 : } else {
1035 0 : flow_iteration_key_ = it->flow();
1036 : }
1037 : }
1038 :
1039 44 : void FlowStatsCollector::DeleteFlow(FlowEntryTree::iterator &it) {
1040 : // Update flow_iteration_key_ if flow being deleted is flow to visit in
1041 : // next ageing cycle
1042 : // Nothing to do if flow being deleted is not the next-iteration key
1043 44 : if (it->first == flow_iteration_key_) {
1044 0 : UpdateFlowIterationKey(it->first, it);
1045 : }
1046 :
1047 44 : if (it == flow_tree_.end())
1048 0 : return;
1049 :
1050 88 : if (it->second.is_linked()) {
1051 : FlowExportInfoList::iterator it1 =
1052 44 : flow_export_info_list_.iterator_to(it->second);
1053 88 : flow_export_info_list_.erase(it1);
1054 : }
1055 :
1056 44 : flow_tree_.erase(it);
1057 : }
1058 :
1059 0 : void FlowStatsCollector::EvictedFlowStatsUpdate(const FlowEntryPtr &flow,
1060 : uint32_t bytes,
1061 : uint32_t packets,
1062 : uint32_t oflow_bytes,
1063 : const boost::uuids::uuid &u) {
1064 0 : FlowExportInfo *info = FindFlowExportInfo(flow.get());
1065 0 : if (info) {
1066 : /* Ignore stats update request for Evicted flow, if we don't have
1067 : * FlowEntry corresponding to the Evicted Flow. The match is done using
1068 : * UUID
1069 : */
1070 0 : if (info->uuid() != u) {
1071 0 : return;
1072 : }
1073 : /* We are updating stats of evicted flow. Set teardown_time here.
1074 : * When delete event is being handled we don't export flow if
1075 : * teardown time is set */
1076 0 : UpdateFlowStatsInternal(info, bytes, oflow_bytes & 0xFFFF,
1077 : packets, oflow_bytes & 0xFFFF0000,
1078 : GetCurrentTime(), true);
1079 : }
1080 : }
1081 :
1082 : /////////////////////////////////////////////////////////////////////////////
1083 : // Introspect routines
1084 : /////////////////////////////////////////////////////////////////////////////
1085 0 : static void KeyToSandeshFlowKey(const FlowKey &key,
1086 : SandeshFlowKey &skey) {
1087 0 : skey.set_nh(key.nh);
1088 0 : skey.set_sip(key.src_addr.to_string());
1089 0 : skey.set_dip(key.dst_addr.to_string());
1090 0 : skey.set_src_port(key.src_port);
1091 0 : skey.set_dst_port(key.dst_port);
1092 0 : skey.set_protocol(key.protocol);
1093 0 : }
1094 :
1095 0 : static void FlowExportInfoToSandesh(const FlowExportInfo &value,
1096 : SandeshFlowExportInfo &info) {
1097 0 : SandeshFlowKey skey;
1098 0 : FlowEntry *flow = value.flow();
1099 0 : FlowEntry *rflow = value.reverse_flow();
1100 0 : KeyToSandeshFlowKey(flow->key(), skey);
1101 0 : info.set_key(skey);
1102 0 : info.set_uuid(to_string(flow->uuid()));
1103 0 : if (rflow) {
1104 0 : info.set_rev_flow_uuid(to_string(rflow->uuid()));
1105 : }
1106 0 : if (!flow->data().origin_vn_src.empty()) {
1107 0 : info.set_source_vn(flow->data().origin_vn_src);
1108 : } else {
1109 0 : info.set_source_vn(flow->data().source_vn_match);
1110 : }
1111 0 : if (!flow->data().origin_vn_dst.empty()) {
1112 0 : info.set_dest_vn(flow->data().origin_vn_dst);
1113 : } else {
1114 0 : info.set_dest_vn(flow->data().dest_vn_match);
1115 : }
1116 0 : info.set_sg_rule_uuid(flow->sg_rule_uuid());
1117 0 : info.set_nw_ace_uuid(flow->nw_ace_uuid());
1118 0 : info.set_teardown_time(value.teardown_time());
1119 0 : info.set_last_modified_time(value.last_modified_time());
1120 0 : info.set_bytes(value.bytes());
1121 0 : info.set_packets(value.packets());
1122 0 : info.set_flow_handle(flow->flow_handle());
1123 0 : std::vector<ActionStr> action_str_l;
1124 0 : SetActionStr(flow->data().match_p.action_info, action_str_l);
1125 0 : info.set_action(action_str_l);
1126 0 : info.set_vm_cfg_name(flow->data().vm_cfg_name);
1127 0 : info.set_peer_vrouter(flow->peer_vrouter());
1128 0 : info.set_tunnel_type(flow->tunnel_type().ToString());
1129 0 : const VmInterfaceKey &vmi = flow->fip_vmi();
1130 0 : string vmi_str = to_string(vmi.uuid_) + vmi.name_;
1131 0 : info.set_fip_vmi(vmi_str);
1132 0 : Ip4Address ip(flow->fip());
1133 0 : info.set_fip(ip.to_string());
1134 0 : info.set_delete_enqueued(value.delete_enqueue_time() ? true : false);
1135 0 : }
1136 :
1137 0 : void FlowStatsRecordsReq::HandleRequest() const {
1138 0 : auto FillFlowStatsCollectorObject = [](
1139 : vector<FlowStatsRecord>& list,
1140 : const FlowStatsCollectorObject& collector_obj) {
1141 0 : for (size_t i_col = 0;
1142 0 : i_col < FlowStatsCollectorObject::kMaxCollectors;
1143 : i_col++) {
1144 : FlowStatsCollector *collector =
1145 0 : collector_obj.GetCollector(i_col);
1146 :
1147 : FlowExportInfoList::iterator it =
1148 0 : collector->flow_export_info_list_.begin();
1149 0 : while (it != collector->flow_export_info_list_.end()) {
1150 0 : const FlowExportInfo &value = *it;
1151 : ++it;
1152 :
1153 0 : SandeshFlowKey skey;
1154 0 : KeyToSandeshFlowKey(value.flow()->key(), skey);
1155 :
1156 0 : SandeshFlowExportInfo info;
1157 0 : FlowExportInfoToSandesh(value, info);
1158 :
1159 0 : FlowStatsRecord rec;
1160 0 : rec.set_info(info);
1161 0 : list.push_back(rec);
1162 0 : }
1163 : }
1164 0 : };
1165 :
1166 0 : vector<FlowStatsRecord> list;
1167 0 : FlowStatsRecordsResp *resp = new FlowStatsRecordsResp();
1168 :
1169 0 : FillFlowStatsCollectorObject(
1170 : list,
1171 : *(Agent::GetInstance()->flow_stats_manager()->
1172 0 : default_flow_stats_collector_obj()));
1173 :
1174 : FlowStatsManager::FlowAgingTableMap &flow_aging_map =
1175 0 : Agent::GetInstance()->flow_stats_manager()->flow_aging_table_map_;
1176 0 : for (auto &collector_obj : flow_aging_map) {
1177 0 : FillFlowStatsCollectorObject(list, *collector_obj.second.get());
1178 : }
1179 :
1180 0 : resp->set_records_list(list);
1181 :
1182 0 : resp->set_context(context());
1183 0 : resp->Response();
1184 0 : return;
1185 0 : }
1186 :
1187 : /////////////////////////////////////////////////////////////////////////////
1188 : // Flow Stats Ageing task
1189 : /////////////////////////////////////////////////////////////////////////////
1190 0 : FlowStatsCollector::AgeingTask::AgeingTask(FlowStatsCollector *fsc) :
1191 0 : Task(fsc->task_id(), fsc->instance_id()), fsc_(fsc) {
1192 0 : }
1193 :
1194 0 : FlowStatsCollector::AgeingTask::~AgeingTask() {
1195 0 : }
1196 :
1197 0 : std::string FlowStatsCollector::AgeingTask::Description() const {
1198 0 : return "Flow Stats Collector Ageing Task";
1199 : }
1200 :
1201 0 : bool FlowStatsCollector::AgeingTask::Run() {
1202 0 : return fsc_->RunAgeingTask();
1203 : }
1204 :
1205 : /////////////////////////////////////////////////////////////////////////////
1206 : // FlowStatsCollectorObject methods
1207 : /////////////////////////////////////////////////////////////////////////////
1208 3 : FlowStatsCollectorObject::FlowStatsCollectorObject(Agent *agent,
1209 : FlowStatsCollectorReq *req,
1210 0 : FlowStatsManager *mgr) {
1211 3 : FlowAgingTableKey *key = &(req->key);
1212 9 : for (int i = 0; i < kMaxCollectors; i++) {
1213 6 : uint32_t instance_id = mgr->AllocateIndex();
1214 : boost::asio::io_context& io_ref =
1215 : const_cast<boost::asio::io_context&>
1216 6 : (*agent->event_manager()->io_service());
1217 12 : collectors[i].reset(
1218 : AgentStaticObjectFactory::CreateRef<FlowStatsCollector>(
1219 : io_ref,
1220 6 : req->flow_stats_interval, req->flow_cache_timeout,
1221 : agent->uve(), instance_id, key, mgr, this));
1222 : }
1223 3 : }
1224 :
1225 36 : FlowStatsCollector* FlowStatsCollectorObject::GetCollector(uint8_t idx) const {
1226 36 : if (idx >= 0 && idx < kMaxCollectors) {
1227 36 : return collectors[idx].get();
1228 : }
1229 0 : return NULL;
1230 : }
1231 :
1232 1 : void FlowStatsCollectorObject::SetExpiryTime(int time) {
1233 3 : for (int i = 0; i < kMaxCollectors; i++) {
1234 2 : collectors[i]->set_expiry_time(time);
1235 : }
1236 1 : }
1237 :
1238 0 : int FlowStatsCollectorObject::GetExpiryTime() const {
1239 : /* Same expiry time would be configured for all the collectors. Pick value
1240 : * from any one of them */
1241 0 : return collectors[0]->expiry_time();
1242 : }
1243 :
1244 0 : void FlowStatsCollectorObject::MarkDelete() {
1245 0 : for (int i = 0; i < kMaxCollectors; i++) {
1246 0 : collectors[i]->set_deleted(true);
1247 : }
1248 0 : }
1249 :
1250 3 : void FlowStatsCollectorObject::ClearDelete() {
1251 9 : for (int i = 0; i < kMaxCollectors; i++) {
1252 6 : collectors[i]->set_deleted(false);
1253 : }
1254 3 : }
1255 :
1256 0 : bool FlowStatsCollectorObject::IsDeleted() const {
1257 0 : for (int i = 0; i < kMaxCollectors; i++) {
1258 0 : if (!collectors[i]->deleted()) {
1259 0 : return false;
1260 : }
1261 : }
1262 0 : return true;
1263 : }
1264 :
1265 3 : void FlowStatsCollectorObject::SetFlowAgeTime(uint64_t value) {
1266 9 : for (int i = 0; i < kMaxCollectors; i++) {
1267 6 : collectors[i]->set_flow_age_time_intvl(value);
1268 : }
1269 3 : }
1270 :
1271 0 : uint64_t FlowStatsCollectorObject::GetFlowAgeTime() const {
1272 : /* Same age time would be configured for all the collectors. Pick value
1273 : * from any one of them */
1274 0 : return collectors[0]->flow_age_time_intvl();
1275 : }
1276 :
1277 0 : bool FlowStatsCollectorObject::CanDelete() const {
1278 0 : for (int i = 0; i < kMaxCollectors; i++) {
1279 0 : if (collectors[i]->flow_tree_.size() != 0 ||
1280 0 : collectors[i]->request_queue_.IsQueueEmpty() == false) {
1281 0 : return false;
1282 : }
1283 : }
1284 0 : return true;
1285 : }
1286 :
1287 3 : void FlowStatsCollectorObject::Shutdown() {
1288 9 : for (int i = 0; i < kMaxCollectors; i++) {
1289 6 : collectors[i]->Shutdown();
1290 6 : collectors[i].reset();
1291 : }
1292 3 : }
1293 :
1294 22 : FlowStatsCollector* FlowStatsCollectorObject::FlowToCollector
1295 : (const FlowEntry *flow) {
1296 22 : uint8_t idx = 0;
1297 22 : FlowTable *table = flow->flow_table();
1298 22 : if (table) {
1299 22 : idx = table->table_index() % kMaxCollectors;
1300 : }
1301 22 : return collectors[idx].get();
1302 : }
1303 :
1304 0 : void FlowStatsCollectorObject::UpdateAgeTimeInSeconds(uint32_t age_time) {
1305 0 : for (int i = 0; i < kMaxCollectors; i++) {
1306 0 : collectors[i]->UpdateFlowAgeTimeInSecs(age_time);
1307 : }
1308 0 : }
1309 :
1310 0 : uint32_t FlowStatsCollectorObject::GetAgeTimeInSeconds() const {
1311 : /* Same age time would be configured for all the collectors. Pick value
1312 : * from any one of them */
1313 0 : return collectors[0]->flow_age_time_intvl_in_secs();
1314 : }
1315 :
1316 0 : size_t FlowStatsCollectorObject::Size() const {
1317 0 : size_t size = 0;
1318 0 : for (int i = 0; i < kMaxCollectors; i++) {
1319 0 : size += collectors[i]->Size();
1320 : }
1321 0 : return size;
1322 : }
|
