Line data Source code
1 : /*
2 : * Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
3 : */
4 :
5 : #include <boost/uuid/uuid_io.hpp>
6 :
7 : #include <db/db.h>
8 : #include <base/util.h>
9 :
10 : #include <cmn/agent_cmn.h>
11 : #include <boost/functional/factory.hpp>
12 : #include <cmn/agent_factory.h>
13 : #include <oper/interface_common.h>
14 : #include <oper/mirror_table.h>
15 :
16 : #include <ksync/ksync_index.h>
17 : #include <ksync/ksync_entry.h>
18 : #include <ksync/ksync_object.h>
19 : #include <ksync/ksync_netlink.h>
20 : #include <ksync/ksync_sock.h>
21 : #include <uve/agent_uve.h>
22 : #include <vrouter/flow_stats/flow_stats_collector.h>
23 : #include <vrouter/flow_stats/session_stats_collector.h>
24 : #include <uve/vn_uve_table.h>
25 : #include <uve/vm_uve_table.h>
26 : #include <uve/interface_uve_stats_table.h>
27 : #include <algorithm>
28 : #include <pkt/flow_proto.h>
29 : #include <vrouter/ksync/ksync_init.h>
30 : #include <vrouter/flow_stats/flow_stats_types.h>
31 : #include <oper/global_vrouter.h>
32 : #include <init/agent_param.h>
33 :
34 : SandeshTraceBufferPtr FlowExportStatsTraceBuf(SandeshTraceBufferCreate(
35 : "FlowExportStats", 3000));
36 : const uint8_t FlowStatsManager::kCatchAllProto;
37 :
38 0 : void FlowStatsManager::UpdateThreshold(uint64_t new_value, bool check_oflow) {
39 0 : if (check_oflow && new_value < threshold_) {
40 : /* Retain the same value for threshold if it results in overflow */
41 0 : return;
42 : }
43 0 : if (new_value < kMinFlowSamplingThreshold) {
44 0 : threshold_ = kMinFlowSamplingThreshold;
45 : } else {
46 0 : threshold_ = new_value;
47 : }
48 : }
49 :
50 0 : void SetFlowStatsInterval_InSeconds::HandleRequest() const {
51 : SandeshResponse *resp;
52 0 : if (get_interval() > 0) {
53 0 : FlowStatsManager *fam = Agent::GetInstance()->flow_stats_manager();
54 0 : FlowStatsCollectorObject *obj = fam->default_flow_stats_collector_obj();
55 0 : if (obj) {
56 0 : obj->SetExpiryTime(get_interval() * 1000);
57 : }
58 0 : resp = new FlowStatsCfgResp();
59 : } else {
60 0 : resp = new FlowStatsCfgErrResp();
61 : }
62 :
63 0 : resp->set_context(context());
64 0 : resp->Response();
65 0 : return;
66 : }
67 :
68 0 : void GetFlowStatsInterval::HandleRequest() const {
69 : FlowStatsIntervalResp_InSeconds *resp =
70 0 : new FlowStatsIntervalResp_InSeconds();
71 0 : resp->set_flow_stats_interval((Agent::GetInstance()->flow_stats_manager()->
72 0 : default_flow_stats_collector_obj()->GetExpiryTime())/1000);
73 :
74 0 : resp->set_context(context());
75 0 : resp->Response();
76 0 : return;
77 : }
78 :
79 3 : FlowStatsManager::FlowStatsManager(Agent *agent) : agent_(agent),
80 3 : request_queue_(agent_->task_scheduler()->GetTaskId("Agent::FlowStatsManager"),
81 : StatsCollector::FlowStatsCollector,
82 : boost::bind(&FlowStatsManager::RequestHandler, this, _1)),
83 3 : prev_flow_export_rate_compute_time_(0),
84 3 : threshold_(kDefaultFlowSamplingThreshold),
85 3 : prev_cfg_flow_export_rate_(0), session_export_rate_(0),
86 3 : session_export_count_(), session_sample_exports_(), session_msg_exports_(),
87 3 : session_exports_(), session_export_disable_drops_(),
88 3 : session_export_sampling_drops_(), session_export_without_sampling_(),
89 3 : session_export_drops_(), session_global_slo_logging_drops_(),
90 3 : session_slo_logging_drops_(),
91 3 : timer_(TimerManager::CreateTimer(*(agent_->event_manager())->io_service(),
92 : "FlowThresholdTimer",
93 : TaskScheduler::GetInstance()->GetTaskId("Agent::FlowStatsManager"), 0)),
94 6 : delete_short_flow_(true) {
95 3 : session_export_count_ = 0;
96 3 : session_sample_exports_ = 0;
97 3 : session_msg_exports_ = 0;
98 3 : session_exports_ = 0;
99 3 : session_export_disable_drops_ = 0;
100 3 : session_export_sampling_drops_ = 0;
101 3 : session_export_without_sampling_ = 0;
102 3 : session_export_drops_ = 0;
103 3 : sessions_sampled_atleast_once_ = false;
104 3 : session_global_slo_logging_drops_ = 0;
105 3 : session_slo_logging_drops_ = 0;
106 3 : request_queue_.set_measure_busy_time(agent->MeasureQueueDelay());
107 771 : for (uint16_t i = 0; i < sizeof(protocol_list_)/sizeof(protocol_list_[0]);
108 : i++) {
109 768 : protocol_list_[i] = NULL;
110 : }
111 : SessionStatsCollectorPtr session_obj(new SessionStatsCollectorObject(agent,
112 3 : this));
113 3 : session_stats_collector_obj_ = session_obj;
114 3 : }
115 :
116 3 : FlowStatsManager::~FlowStatsManager() {
117 3 : assert(flow_aging_table_map_.size() == 0);
118 3 : }
119 :
120 6 : bool FlowStatsManager::RequestHandler(boost::shared_ptr<FlowStatsCollectorReq>
121 : req) {
122 6 : switch (req->event) {
123 6 : case FlowStatsCollectorReq::ADD_FLOW_STATS_COLLECTOR: {
124 6 : AddReqHandler(req);
125 6 : break;
126 : }
127 :
128 0 : case FlowStatsCollectorReq::DELETE_FLOW_STATS_COLLECTOR: {
129 0 : DeleteReqHandler(req);
130 0 : break;
131 : }
132 :
133 0 : case FlowStatsCollectorReq::FREE_FLOW_STATS_COLLECTOR: {
134 0 : FreeReqHandler(req);
135 0 : break;
136 : }
137 :
138 0 : default: {
139 0 : assert(0);
140 : break;
141 : }
142 : }
143 6 : return true;
144 : }
145 :
146 6 : void FlowStatsManager::AddReqHandler(boost::shared_ptr<FlowStatsCollectorReq>
147 : req) {
148 6 : FlowAgingTableMap::iterator it = flow_aging_table_map_.find(req->key);
149 6 : if (it != flow_aging_table_map_.end()) {
150 3 : it->second->SetFlowAgeTime(
151 3 : 1000000L * (uint64_t)req->flow_cache_timeout);
152 3 : it->second->ClearDelete();
153 3 : return;
154 : }
155 :
156 3 : FlowAgingTablePtr aging_table(new FlowStatsCollectorObject(agent(),
157 3 : req.get(),
158 3 : this));
159 3 : flow_aging_table_map_.insert(FlowAgingTableEntry(req->key, aging_table));
160 3 : if (req->key.proto == kCatchAllProto && req->key.port == 0) {
161 3 : default_flow_stats_collector_obj_ = aging_table;
162 : }
163 :
164 3 : if (req->key.port == 0) {
165 3 : protocol_list_[req->key.proto] = aging_table.get();
166 : }
167 3 : }
168 :
169 0 : void FlowStatsManager::DeleteReqHandler(boost::shared_ptr<FlowStatsCollectorReq>
170 : req) {
171 0 : FlowAgingTableMap::iterator it = flow_aging_table_map_.find(req->key);
172 0 : if (it == flow_aging_table_map_.end()) {
173 0 : return;
174 : }
175 :
176 0 : FlowAgingTablePtr flow_aging_table_ptr = it->second;
177 0 : flow_aging_table_ptr->MarkDelete();
178 :
179 0 : if (flow_aging_table_ptr->CanDelete()) {
180 0 : flow_aging_table_map_.erase(it);
181 0 : protocol_list_[req->key.proto] = NULL;
182 : }
183 0 : }
184 :
185 0 : void FlowStatsManager::FreeReqHandler(boost::shared_ptr<FlowStatsCollectorReq>
186 : req) {
187 0 : FlowAgingTableMap::iterator it = flow_aging_table_map_.find(req->key);
188 0 : if (it == flow_aging_table_map_.end()) {
189 0 : return;
190 : }
191 :
192 0 : FlowAgingTablePtr flow_aging_table_ptr = it->second;
193 0 : if (flow_aging_table_ptr->IsDeleted() == false) {
194 0 : return;
195 : }
196 0 : assert(flow_aging_table_ptr->CanDelete());
197 0 : flow_aging_table_ptr->Shutdown();
198 0 : flow_aging_table_map_.erase(it);
199 0 : protocol_list_[req->key.proto] = NULL;
200 0 : }
201 :
202 6 : void FlowStatsManager::Add(const FlowAgingTableKey &key,
203 : uint64_t flow_stats_interval,
204 : uint64_t flow_cache_timeout) {
205 : boost::shared_ptr<FlowStatsCollectorReq>
206 : req(new FlowStatsCollectorReq(
207 : FlowStatsCollectorReq::ADD_FLOW_STATS_COLLECTOR,
208 6 : key, flow_stats_interval, flow_cache_timeout));
209 6 : request_queue_.Enqueue(req);
210 6 : }
211 :
212 0 : void FlowStatsManager::Delete(const FlowAgingTableKey &key) {
213 0 : if (key.proto == kCatchAllProto) {
214 0 : return;
215 : }
216 : boost::shared_ptr<FlowStatsCollectorReq>
217 : req(new FlowStatsCollectorReq(
218 : FlowStatsCollectorReq::DELETE_FLOW_STATS_COLLECTOR,
219 0 : key));
220 0 : request_queue_.Enqueue(req);
221 0 : }
222 :
223 0 : void FlowStatsManager::Free(const FlowAgingTableKey &key) {
224 : boost::shared_ptr<FlowStatsCollectorReq>
225 : req(new FlowStatsCollectorReq(
226 : FlowStatsCollectorReq::FREE_FLOW_STATS_COLLECTOR,
227 0 : key));
228 0 : request_queue_.Enqueue(req);
229 0 : }
230 :
231 : const FlowStatsCollectorObject*
232 0 : FlowStatsManager::Find(uint32_t proto, uint32_t port) const {
233 :
234 0 : FlowAgingTableKey key1(proto, port);
235 0 : FlowAgingTableMap::const_iterator key1_it = flow_aging_table_map_.find(key1);
236 :
237 0 : if (key1_it == flow_aging_table_map_.end()){
238 0 : return NULL;
239 : }
240 :
241 0 : return key1_it->second.get();
242 : }
243 :
244 : FlowStatsCollectorObject*
245 22 : FlowStatsManager::GetFlowStatsCollectorObject(const FlowEntry *flow) const {
246 22 : FlowStatsCollectorObject* col = NULL;
247 :
248 22 : const FlowKey &key = flow->key();
249 22 : FlowAgingTableKey key1(key.protocol, key.src_port);
250 : FlowAgingTableMap::const_iterator key1_it =
251 22 : flow_aging_table_map_.find(key1);
252 :
253 22 : if (key1_it != flow_aging_table_map_.end()) {
254 0 : col = key1_it->second.get();
255 0 : if (!col->IsDeleted())
256 0 : return col;
257 : }
258 :
259 22 : FlowAgingTableKey key2(key.protocol, key.dst_port);
260 : FlowAgingTableMap::const_iterator key2_it =
261 22 : flow_aging_table_map_.find(key2);
262 22 : if (key2_it != flow_aging_table_map_.end()) {
263 0 : col = key2_it->second.get();
264 0 : if (!col->IsDeleted())
265 0 : return col;
266 : }
267 :
268 22 : if (protocol_list_[key.protocol] != NULL) {
269 0 : col = protocol_list_[key.protocol];
270 0 : if (!col->IsDeleted())
271 0 : return col;
272 : }
273 22 : return default_flow_stats_collector_obj_.get();
274 : }
275 :
276 : FlowStatsCollector*
277 44 : FlowStatsManager::GetFlowStatsCollector(const FlowEntry *flow) const {
278 : /* If the reverse flow already has FlowStatsCollector assigned, return
279 : * the same to ensure that forward and reverse flows go to same
280 : * FlowStatsCollector */
281 44 : const FlowEntry *rflow = flow->reverse_flow_entry();
282 44 : if (rflow && rflow->fsc()) {
283 22 : return rflow->fsc();
284 : }
285 22 : FlowStatsCollectorObject* obj = GetFlowStatsCollectorObject(flow);
286 :
287 22 : return obj->FlowToCollector(flow);
288 : }
289 :
290 100 : void FlowStatsManager::AddEvent(FlowEntryPtr &flow) {
291 100 : if (flow == NULL) {
292 0 : return;
293 : }
294 :
295 100 : FlowStatsCollector *fsc = NULL;
296 100 : if (flow->fsc() == NULL) {
297 44 : fsc = GetFlowStatsCollector(flow.get());
298 44 : flow->set_fsc(fsc);
299 : } else {
300 56 : fsc = flow->fsc();
301 : }
302 :
303 100 : fsc->AddEvent(flow);
304 :
305 100 : SessionStatsCollector *ssc = NULL;
306 100 : ssc = session_stats_collector_obj_->FlowToCollector(flow.get());
307 100 : if (ssc) {
308 100 : ssc->AddEvent(flow);
309 : }
310 : }
311 :
312 44 : void FlowStatsManager::DeleteEvent(const FlowEntryPtr &flow,
313 : const RevFlowDepParams ¶ms) {
314 44 : if (flow == NULL) {
315 0 : return;
316 : }
317 44 : FlowStatsCollector *fsc = flow->fsc();
318 : /* Ignore delete requests if FlowStatsCollector is NULL */
319 44 : if (fsc != NULL) {
320 44 : fsc->DeleteEvent(flow, params);
321 44 : flow->set_fsc(NULL);
322 : }
323 :
324 44 : SessionStatsCollector *ssc = NULL;
325 44 : ssc = session_stats_collector_obj_->FlowToCollector(flow.get());
326 44 : if (ssc) {
327 44 : ssc->DeleteEvent(flow, params);
328 : }
329 : }
330 :
331 0 : void FlowStatsManager::UpdateStatsEvent(const FlowEntryPtr &flow,
332 : uint32_t bytes, uint32_t packets,
333 : uint32_t oflow_bytes,
334 : const boost::uuids::uuid &u) {
335 0 : if (flow == NULL) {
336 0 : return;
337 : }
338 :
339 0 : FlowStatsCollector *fsc = flow->fsc();
340 0 : if (fsc == NULL) {
341 : /* Ignore stats update request, if the flow does not have any
342 : * FlowStatsCollector associated with it */
343 0 : return;
344 : }
345 :
346 0 : fsc->UpdateStatsEvent(flow, bytes, packets, oflow_bytes, u);
347 :
348 0 : SessionStatsCollector *ssc = NULL;
349 0 : ssc = session_stats_collector_obj_->FlowToCollector(flow.get());
350 0 : if (ssc) {
351 0 : ssc->UpdateSessionStatsEvent(flow, bytes, packets, oflow_bytes, u);
352 : }
353 : }
354 :
355 9 : uint32_t FlowStatsManager::AllocateIndex() {
356 9 : return instance_table_.Insert(NULL);
357 : }
358 :
359 9 : void FlowStatsManager::FreeIndex(uint32_t idx) {
360 9 : instance_table_.Remove(idx);
361 9 : }
362 :
363 0 : void FlowStatsManager::FlowStatsReqHandler(Agent *agent,
364 : uint32_t protocol, uint32_t port, uint64_t timeout) {
365 0 : if (timeout == 0) {
366 0 : agent->flow_stats_manager()->Delete(
367 0 : FlowAgingTableKey(protocol, port));
368 : } else {
369 0 : agent->flow_stats_manager()->Add(
370 0 : FlowAgingTableKey(protocol, port),
371 0 : agent->params()->flow_stats_interval(), timeout);
372 : }
373 0 : }
374 :
375 3 : void FlowStatsManager::RegisterDBClients() {
376 3 : session_stats_collector_obj_->RegisterDBClients();
377 3 : return;
378 : }
379 :
380 :
381 6 : void FlowStatsManager::Init(uint64_t flow_stats_interval,
382 : uint64_t flow_cache_timeout) {
383 6 : Add(FlowAgingTableKey(kCatchAllProto, 0),
384 : flow_stats_interval, flow_cache_timeout);
385 :
386 6 : if (agent_->tsn_enabled()) {
387 : /* In TSN mode, we don't support add/delete of FlowStatsCollector
388 : * (so we don't invoke set_flow_stats_req_handler)
389 : * Also, we don't export flows, so we don't start UpdateSessionThreshold
390 : * timer */
391 0 : return;
392 : }
393 6 : agent_->set_flow_stats_req_handler(&(FlowStatsManager::FlowStatsReqHandler));
394 :
395 6 : timer_->Start(FlowThresoldUpdateTime,
396 : boost::bind(&FlowStatsManager::UpdateSessionThreshold, this));
397 : }
398 :
399 0 : void FlowStatsManager::InitDone() {
400 0 : AgentProfile *profile = agent_->oper_db()->agent_profile();
401 0 : profile->RegisterFlowStatsCb(boost::bind(&FlowStatsManager::SetProfileData,
402 : this, _1));
403 0 : }
404 :
405 3 : void FlowStatsManager::Shutdown() {
406 3 : default_flow_stats_collector_obj_->Shutdown();
407 3 : default_flow_stats_collector_obj_.reset();
408 3 : session_stats_collector_obj_->Shutdown();
409 3 : session_stats_collector_obj_.reset();
410 3 : flow_aging_table_map_.clear();
411 3 : protocol_list_[0] = NULL;
412 3 : timer_->Cancel();
413 3 : TimerManager::DeleteTimer(timer_);
414 3 : request_queue_.Shutdown();
415 3 : }
416 :
417 0 : void ShowAgingConfig::HandleRequest() const {
418 : SandeshResponse *resp;
419 :
420 0 : FlowStatsManager *fam = Agent::GetInstance()->flow_stats_manager();
421 0 : resp = new AgingConfigResponse();
422 :
423 0 : FlowStatsManager::FlowAgingTableMap::const_iterator it = fam->begin();
424 0 : while (it != fam->end()) {
425 0 : AgingConfig cfg;
426 0 : cfg.set_protocol(it->first.proto);
427 0 : cfg.set_port(it->first.port);
428 0 : cfg.set_cache_timeout(it->second->GetAgeTimeInSeconds());
429 0 : cfg.set_stats_interval(0);
430 : std::vector<AgingConfig> &list =
431 : const_cast<std::vector<AgingConfig>&>(
432 0 : ((AgingConfigResponse *)resp)->get_aging_config_list());
433 0 : list.push_back(cfg);
434 0 : it++;
435 0 : }
436 :
437 0 : resp->set_context(context());
438 0 : resp->Response();
439 0 : return;
440 : }
441 :
442 0 : void AddAgingConfig::HandleRequest() const {
443 0 : FlowStatsManager *fam = Agent::GetInstance()->flow_stats_manager();
444 0 : fam->Add(FlowAgingTableKey(get_protocol(), get_port()),
445 0 : get_stats_interval(), get_cache_timeout());
446 0 : SandeshResponse *resp = new FlowStatsCfgResp();
447 0 : resp->set_context(context());
448 0 : resp->Response();
449 0 : return;
450 : }
451 :
452 0 : void DeleteAgingConfig::HandleRequest() const {
453 0 : FlowStatsManager *fam = Agent::GetInstance()->flow_stats_manager();
454 0 : fam->Delete(FlowAgingTableKey(get_protocol(), get_port()));
455 :
456 0 : SandeshResponse *resp = new FlowStatsCfgResp();
457 0 : resp->set_context(context());
458 0 : resp->Response();
459 0 : return;
460 : }
461 :
462 0 : static void SetQueueStats(Agent *agent, FlowStatsCollector *fsc,
463 : ProfileData::WorkQueueStats *stats) {
464 0 : stats->name_ = fsc->queue()->Description();
465 0 : stats->queue_count_ = fsc->queue()->Length();
466 0 : stats->enqueue_count_ = fsc->queue()->NumEnqueues();
467 0 : stats->dequeue_count_ = fsc->queue()->NumDequeues();
468 0 : stats->max_queue_count_ = fsc->queue()->max_queue_len();
469 0 : stats->start_count_ = fsc->queue()->task_starts();
470 0 : stats->busy_time_ = fsc->queue()->busy_time();
471 0 : fsc->queue()->set_measure_busy_time(agent->MeasureQueueDelay());
472 0 : if (agent->MeasureQueueDelay())
473 0 : fsc->queue()->ClearStats();
474 0 : }
475 :
476 0 : void FlowStatsManager::SetProfileData(ProfileData *data) {
477 0 : uint32_t qsize = flow_aging_table_map_.size() *
478 0 : FlowStatsCollectorObject::kMaxCollectors;
479 0 : data->flow_.flow_stats_queue_.resize(qsize);
480 0 : int i = 0;
481 0 : FlowAgingTableMap::iterator it = flow_aging_table_map_.begin();
482 0 : while (it != flow_aging_table_map_.end()) {
483 0 : FlowStatsCollectorObject *obj = it->second.get();
484 0 : for (int j = 0; j < FlowStatsCollectorObject::kMaxCollectors; j++) {
485 0 : SetQueueStats(agent(), obj->GetCollector(j),
486 0 : &data->flow_.flow_stats_queue_[i]);
487 0 : i++;
488 : }
489 0 : it++;
490 : }
491 0 : }
492 :
493 0 : void FlowStatsManager::UpdateSessionSampleExportStats(uint32_t count) {
494 0 : session_sample_exports_ += count;
495 0 : }
496 :
497 0 : void FlowStatsManager::UpdateSessionMsgExportStats(uint32_t count) {
498 0 : session_msg_exports_ += count;
499 0 : }
500 :
501 0 : void FlowStatsManager::UpdateSessionExportStats(uint32_t count,
502 : bool first_export,
503 : bool sampled) {
504 0 : session_export_count_ += count;
505 0 : if (first_export) {
506 0 : session_exports_ += count;
507 : }
508 0 : if (!sampled) {
509 0 : session_export_without_sampling_ += count;
510 : }
511 0 : }
|
