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