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