Merge "Fix fd leak in linklocal flows."

src/vnsw/agent/pkt/flow_table.cc

@@ -107,6 +107,25 @@ FlowEntry::FlowEntry(const FlowKey &k) :
     alloc_count_.fetch_and_increment();
 }
 
+FlowEntry::~FlowEntry() {
+    if (is_flags_set(FlowEntry::LinkLocalBindLocalSrcPort) &&
+        (linklocal_src_port_fd_ == PktFlowInfo::kLinkLocalInvalidFd ||
+         !linklocal_src_port_)) {
+        LOG(DEBUG, "Linklocal Flow Inconsistency fd = " <<
+            linklocal_src_port_fd_ << " port = " << linklocal_src_port_ <<
+            " flow index = " << flow_handle_ << " source = " <<
+            key_.src_addr.to_string() << " dest = " <<
+            key_.dst_addr.to_string() << " protocol = " << key_.protocol <<
+            " sport = " << key_.src_port << " dport = " << key_.dst_port);
+    }
+    if (linklocal_src_port_fd_ != PktFlowInfo::kLinkLocalInvalidFd) {
+        close(linklocal_src_port_fd_);
+        Agent::GetInstance()->pkt()->flow_table()->
+            DelLinkLocalFlowInfo(linklocal_src_port_fd_);
+    }
+    alloc_count_.fetch_and_decrement();
+}
+
 void FlowEntry::GetSourceRouteInfo(const AgentRoute *rt) {
     const AgentPath *path = NULL;
     if (rt) {
@@ -925,6 +944,22 @@ void FlowEntry::GetPolicyInfo() {
     GetPolicyInfo(data_.vn_entry.get());
 }
 
+void FlowTable::AddLinkLocalFlowInfo(int fd, uint32_t index, const FlowKey &key,
+                                     const uint64_t timestamp) {
+    LinkLocalFlowInfoMap::iterator it = linklocal_flow_info_map_.find(fd);
+    if (it == linklocal_flow_info_map_.end()) {
+        linklocal_flow_info_map_.insert(
+          LinkLocalFlowInfoPair(fd, LinkLocalFlowInfo(index, key, timestamp)));
+    } else {
+        it->second.flow_index = index;
+        it->second.flow_key = key;
+    }
+}
+
+void FlowTable::DelLinkLocalFlowInfo(int fd) {
+    linklocal_flow_info_map_.erase(fd);
+}
+
 void FlowTable::Add(FlowEntry *flow, FlowEntry *rflow) {
     flow->reset_flags(FlowEntry::ReverseFlow);
     /* reverse flow may not be aviable always, eg: Flow Audit */
@@ -1419,16 +1454,19 @@ void FlowEntry::InitFwdFlow(const PktFlowInfo *info, const PktInfo *pkt,
         flow_handle_ = pkt->GetAgentHdr().cmd_param;
     }
 
-    if (InitFlowCmn(info, ctrl, rev_ctrl) == false) {
-        return;
-    }
     if (info->linklocal_bind_local_port) {
         linklocal_src_port_ = info->nat_sport;
         linklocal_src_port_fd_ = info->linklocal_src_port_fd;
+        Agent::GetInstance()->pkt()->flow_table()->AddLinkLocalFlowInfo(
+            linklocal_src_port_fd_, flow_handle_, key_, stats_.setup_time);
         set_flags(FlowEntry::LinkLocalBindLocalSrcPort);
     } else {
         reset_flags(FlowEntry::LinkLocalBindLocalSrcPort);
     }
+
+    if (InitFlowCmn(info, ctrl, rev_ctrl) == false) {
+        return;
+    }
     stats_.intf_in = pkt->GetAgentHdr().ifindex;
 
     if (info->ingress) {

src/vnsw/agent/pkt/flow_table.h

@@ -331,12 +331,7 @@ class FlowEntry {
         UnknownUnicastFlood = 1 << 11
     };
     FlowEntry(const FlowKey &k);
-    virtual ~FlowEntry() {
-        if (linklocal_src_port_fd_ != PktFlowInfo::kLinkLocalInvalidFd) {
-            close(linklocal_src_port_fd_);
-        }
-        alloc_count_.fetch_and_decrement();
-    };
+    virtual ~FlowEntry();
 
     bool ActionRecompute();
     void UpdateKSync(FlowTable* table);
@@ -434,6 +429,7 @@ class FlowEntry {
     bool set_pending_recompute(bool value);
     const MacAddress &smac() const { return data_.smac; }
     const MacAddress &dmac() const { return data_.dmac; }
+
 private:
     friend class FlowTable;
     friend class FlowStatsCollector;
@@ -642,6 +638,17 @@ class FlowTable {
         SecurityGroupList sg_l_;
     };
 
+    struct LinkLocalFlowInfo {
+        uint32_t flow_index;
+        FlowKey flow_key;
+        uint64_t timestamp;
+
+        LinkLocalFlowInfo(uint32_t index, const FlowKey &key, uint64_t t) :
+            flow_index(index), flow_key(key), timestamp(t) {}
+    };
+    typedef std::map<int, LinkLocalFlowInfo> LinkLocalFlowInfoMap;
+    typedef std::pair<int, LinkLocalFlowInfo> LinkLocalFlowInfoPair;
+
     FlowTable(Agent *agent);
     virtual ~FlowTable();
 
@@ -683,6 +690,13 @@ class FlowTable {
         return flow_entry_map_.end(); 
     }
 
+    const LinkLocalFlowInfoMap &linklocal_flow_info_map() {
+        return linklocal_flow_info_map_;
+    }
+    void AddLinkLocalFlowInfo(int fd, uint32_t index, const FlowKey &key,
+                              const uint64_t timestamp);
+    void DelLinkLocalFlowInfo(int fd);
+
     DBTableBase::ListenerId nh_listener_id();
     AgentRoute *GetL2Route(const VrfEntry *entry, const MacAddress &mac);
     AgentRoute *GetUcRoute(const VrfEntry *entry, const IpAddress &addr);
@@ -739,6 +753,9 @@ class FlowTable {
     InetUnicastRouteEntry inet6_route_key_;
     FlowIndexTree flow_index_tree_;
 
+    // maintain the linklocal flow info against allocated fd, debug purpose only
+    LinkLocalFlowInfoMap linklocal_flow_info_map_;
+
     void AclNotify(DBTablePartBase *part, DBEntryBase *e);
     void IntfNotify(DBTablePartBase *part, DBEntryBase *e);
     void VnNotify(DBTablePartBase *part, DBEntryBase *e);

src/vnsw/agent/pkt/pkt.sandesh

@@ -74,6 +74,17 @@ struct SandeshFlowData {
     50: bool enable_rpf;
 }
 
+struct LinkLocalFlowInfo {
+    1: u32 fd;
+    2: u32 flow_index;
+    3: string source_addr;
+    4: string dest_addr;
+    5: u32 protocol;
+    6: u16 source_port;
+    7: u16 dest_port;
+    8: string timestamp;
+}
+
 request sandesh FetchFlowRecord {
     1: i32 nh;
     2: string sip;
@@ -101,11 +112,18 @@ request sandesh FetchAllFlowRecords {
 request sandesh DeleteAllFlowRecords {
 }
 
+request sandesh FetchLinkLocalFlowInfo {
+}
+
 response sandesh FlowRecordsResp {
     1: list<SandeshFlowData> flow_list;
     2: string flow_key (link="NextFlowRecordsSet");
 }
 
+response sandesh LinkLocalFlowInfoResp {
+    1: list<LinkLocalFlowInfo> linklocal_flow_list;
+}
+
 trace sandesh TapErr {
     1: string err;
 }

src/vnsw/agent/pkt/pkt_flow_info.cc

@@ -1380,13 +1380,20 @@ void PktFlowInfo::Add(const PktInfo *pkt, PktControlInfo *in,
         short_flow_reason = FlowEntry::SHORT_FLOW_LIMIT;
     }
 
-    if (!short_flow && linklocal_bind_local_port &&
-        flow->linklocal_src_port_fd() == PktFlowInfo::kLinkLocalInvalidFd) {
-        nat_sport = LinkLocalBindPort(in->vm_, pkt->ip_proto);
-        if (!nat_sport) {
-            flow_table->agent()->stats()->incr_flow_drop_due_to_max_limit();
-            short_flow = true;
-            short_flow_reason = FlowEntry::SHORT_LINKLOCAL_SRC_NAT;
+    // copy the linklocal fd from the flow, required for recompute cases
+    linklocal_src_port_fd = flow->linklocal_src_port_fd();
+    if (linklocal_bind_local_port) {
+        nat_sport = flow->linklocal_src_port();
+        if (!short_flow) {
+            if (flow->linklocal_src_port_fd() ==
+                PktFlowInfo::kLinkLocalInvalidFd) {
+                nat_sport = LinkLocalBindPort(in->vm_, pkt->ip_proto);
+            }
+            if (!nat_sport) {
+                flow_table->agent()->stats()->incr_flow_drop_due_to_max_limit();
+                short_flow = true;
+                short_flow_reason = FlowEntry::SHORT_LINKLOCAL_SRC_NAT;
+            }
         }
     }
 

src/vnsw/agent/pkt/pkt_sandesh_flow.cc

@@ -425,4 +425,32 @@ void FlowAgeTimeReq::HandleRequest() const {
     resp->Response();
 }
 
+void FetchLinkLocalFlowInfo::HandleRequest() const {
+    LinkLocalFlowInfoResp *resp = new LinkLocalFlowInfoResp();
+    std::vector<LinkLocalFlowInfo> &list =
+        const_cast<std::vector<LinkLocalFlowInfo>&>
+        (resp->get_linklocal_flow_list());
+
+    const FlowTable::LinkLocalFlowInfoMap &flow_map =
+        Agent::GetInstance()->pkt()->flow_table()->linklocal_flow_info_map();
+    FlowTable::LinkLocalFlowInfoMap::const_iterator it = flow_map.begin();
+    while (it != flow_map.end()) {
+        LinkLocalFlowInfo info;
+        info.fd = it->first;
+        info.flow_index = it->second.flow_index;
+        info.source_addr = it->second.flow_key.src_addr.to_string();
+        info.dest_addr = it->second.flow_key.dst_addr.to_string();
+        info.protocol = it->second.flow_key.protocol;
+        info.source_port = it->second.flow_key.src_port;
+        info.dest_port = it->second.flow_key.dst_port;
+        info.timestamp = integerToString(UTCUsecToPTime(it->second.timestamp));
+        list.push_back(info);
+        ++it;
+    }
+
+    resp->set_context(context());
+    resp->set_more(false);
+    resp->Response();
+}
+
 ////////////////////////////////////////////////////////////////////////////////