Merge "Set Vrouter to Drop New Flows on flow limit"

src/vnsw/agent/cmn/agent.h

@@ -53,6 +53,10 @@ typedef boost::intrusive_ptr<VmEntry> VmEntryRef;
 typedef boost::intrusive_ptr<const VmEntry> VmEntryConstRef;
 void intrusive_ptr_release(const VmEntry* p);
 void intrusive_ptr_add_ref(const VmEntry* p);
+typedef IntrusivePtrRef<VmEntry> VmEntryBackRef;
+typedef IntrusivePtrRef<const VmEntry> VmEntryConstBackRef;
+void intrusive_ptr_add_back_ref(const IntrusiveReferrer ref, const VmEntry* p);
+void intrusive_ptr_del_back_ref(const IntrusiveReferrer ref, const VmEntry* p);
 
 class VnEntry;
 typedef boost::intrusive_ptr<VnEntry> VnEntryRef;

src/vnsw/agent/cmn/agent_db.h

@@ -39,14 +39,14 @@ class AgentRefCount {
     friend void intrusive_ptr_add_back_ref(const IntrusiveReferrer ref,
                                            const Derived* p) {
         const AgentRefCount *entry = (const AgentRefCount *) (p);
-        tbb::mutex::scoped_lock lock(entry->mutex_);
+        tbb::mutex::scoped_lock lock(entry->back_ref_mutex_);
         entry->back_ref_set_.insert(ref);
     }
 
     friend void intrusive_ptr_del_back_ref(const IntrusiveReferrer ref,
                                            const Derived* p) {
         const AgentRefCount *entry = (const AgentRefCount *) (p);
-        tbb::mutex::scoped_lock lock(entry->mutex_);
+        tbb::mutex::scoped_lock lock(entry->back_ref_mutex_);
         entry->back_ref_set_.erase(ref);
     }
 
@@ -57,9 +57,11 @@ class AgentRefCount {
     AgentRefCount& operator=(const AgentRefCount&) { return *this; }
     virtual ~AgentRefCount() {assert(refcount_ == 0);};
     void swap(AgentRefCount&) {};
-private:
-    mutable tbb::mutex mutex_;
+
+    mutable tbb::mutex back_ref_mutex_;
     mutable std::set<IntrusiveReferrer> back_ref_set_;
+
+private:
     mutable tbb::atomic<uint32_t> refcount_;
 };
 

src/vnsw/agent/oper/agent.sandesh

@@ -162,6 +162,7 @@ struct ItfSandeshData {
    55: optional string service_health_check_ip;
    56: optional string qos_config (link="AgentQosConfigSandeshReq"); //QOS config uuid
    57: list<AliasIpSandeshList> alias_ip_list;// Alias IP List
+   58: bool drop_new_flows;
 }
 
 /**
@@ -232,6 +233,7 @@ struct VmSgUuid {
 struct VmSandeshData {
     1: string uuid;                   // VM uuid
     2: optional list<VmSgUuid> sg_uuid_list;
+    3: bool drop_new_flows;
 }
 
 /**

src/vnsw/agent/oper/interface.cc

@@ -1063,6 +1063,7 @@ void Interface::SetItfSandeshData(ItfSandeshData &data) const {
 
         data.set_service_health_check_ip(
                 vintf->service_health_check_ip().to_string());
+        data.set_drop_new_flows(vintf->drop_new_flows());
 
         break;
     }

src/vnsw/agent/oper/vm.cc

@@ -18,7 +18,8 @@ using namespace autogen;
 
 VmTable *VmTable::vm_table_;
 
-VmEntry::VmEntry(const uuid &id) : uuid_(id), name_("") {
+VmEntry::VmEntry(const uuid &id) : uuid_(id), name_(""),
+    drop_new_flows_(false) {
     flow_count_ = 0;
     linklocal_flow_count_ = 0;
 }
@@ -56,13 +57,57 @@ bool VmEntry::DBEntrySandesh(Sandesh *sresp, std::string &name) const {
         data.set_uuid(str_uuid);
         std::vector<VmSandeshData> &list =
                 const_cast<std::vector<VmSandeshData>&>(resp->get_vm_list());
+        data.set_drop_new_flows(drop_new_flows_);
         list.push_back(data);
         return true;
     }
 
     return false;
 }
 
+void VmEntry::update_flow_count(int val) const {
+    VmTable *vm_table = static_cast<VmTable *>(get_table());
+    int max_flows = vm_table->agent()->max_vm_flows();
+    int tmp = flow_count_.fetch_and_add(val);
+
+    if (max_flows == 0) {
+        // max_flows are not configured,
+        // disable drop new flows and return
+        SetInterfacesDropNewFlows(false);
+        return;
+    }
+
+    if (val < 0) {
+        assert(tmp >= val);
+        if ((tmp + val) < ((max_flows * kDropNewFlowsRecoveryThreshold)/100)) {
+            SetInterfacesDropNewFlows(false);
+        }
+    } else {
+        if ((tmp + val) >= max_flows) {
+            SetInterfacesDropNewFlows(true);
+        }
+    }
+}
+
+void VmEntry::SetInterfacesDropNewFlows(bool drop_new_flows) const {
+    if (drop_new_flows_ == drop_new_flows) {
+        return;
+    }
+    drop_new_flows_ = drop_new_flows;
+    VmTable *vm_table = static_cast<VmTable *>(get_table());
+    DBRequest req;
+    req.oper = DBRequest::DB_ENTRY_ADD_CHANGE;
+    tbb::mutex::scoped_lock lock(back_ref_mutex_);
+    std::set<IntrusiveReferrer>::const_iterator it = back_ref_set_.begin();
+    for (; it != back_ref_set_.end(); it++) {
+        VmInterface *vm_intf = static_cast<VmInterface *>((*it).first);
+        req.key.reset(new VmInterfaceKey(AgentKey::RESYNC,
+                                         vm_intf->GetUuid(), ""));
+        req.data.reset(new VmInterfaceNewFlowDropData(drop_new_flows));
+        vm_table->agent()->interface_table()->Enqueue(&req);
+    }
+}
+
 void VmEntry::SendObjectLog(AgentLogEvent::type event) const {
     VmObjectLogInfo info;
     string str;

src/vnsw/agent/oper/vm.h

@@ -33,6 +33,12 @@ struct VmData : public AgentOperDBData {
 class VmEntry : AgentRefCount<VmEntry>, public AgentOperDBEntry {
 public:
     static const int kVectorIncreaseSize = 16;
+
+    // kDropNewFlowsRecoveryThreshold is set to 90% of the Max flows for a
+    // VM this value represents that recovery from the drop new flows will
+    // happen only when total number of flow fall below this value
+    static const int kDropNewFlowsRecoveryThreshold = 90;
+
     VmEntry(const uuid &id);
     virtual ~VmEntry();
 
@@ -53,24 +59,25 @@ class VmEntry : AgentRefCount<VmEntry>, public AgentOperDBEntry {
     bool DBEntrySandesh(Sandesh *sresp, std::string &name) const;
 
     uint32_t flow_count() const { return flow_count_; }
-    void update_flow_count(int val) const {
-        int tmp = flow_count_.fetch_and_add(val);
-        if (val < 0)
-            assert(tmp >= val);
-    }
+    void update_flow_count(int val) const;
 
     uint32_t linklocal_flow_count() const { return linklocal_flow_count_; }
     void update_linklocal_flow_count(int val) const {
         int tmp = linklocal_flow_count_.fetch_and_add(val);
         if (val < 0)
             assert(tmp >= val);
     }
+
+    bool drop_new_flows() const { return drop_new_flows_; }
+
 private:
+    void SetInterfacesDropNewFlows(bool drop_new_flows) const;
     friend class VmTable;
     uuid uuid_;
     std::string name_;
     mutable tbb::atomic<int> flow_count_;
     mutable tbb::atomic<int> linklocal_flow_count_;
+    mutable bool drop_new_flows_;
     DISALLOW_COPY_AND_ASSIGN(VmEntry);
 };
 

src/vnsw/agent/oper/vm_interface.cc

@@ -55,12 +55,12 @@ using namespace boost::uuids;
 using namespace autogen;
 
 VmInterface::VmInterface(const boost::uuids::uuid &uuid) :
-    Interface(Interface::VM_INTERFACE, uuid, "", NULL), vm_(NULL),
+    Interface(Interface::VM_INTERFACE, uuid, "", NULL), vm_(NULL, this),
     vn_(NULL), primary_ip_addr_(0), mdata_ip_(NULL), subnet_bcast_addr_(0),
     primary_ip6_addr_(), vm_mac_(MacAddress::kZeroMac), policy_enabled_(false),
     mirror_entry_(NULL), mirror_direction_(MIRROR_RX_TX), cfg_name_(""),
-    fabric_port_(true), need_linklocal_ip_(false), dhcp_enable_(true),
-    do_dhcp_relay_(false), vm_name_(),
+    fabric_port_(true), need_linklocal_ip_(false), drop_new_flows_(false),
+    dhcp_enable_(true), do_dhcp_relay_(false), vm_name_(),
     vm_project_uuid_(nil_uuid()), vxlan_id_(0), bridging_(false),
     layer3_forwarding_(true), flood_unknown_unicast_(false),
     mac_set_(false), ecmp_(false), ecmp6_(false), disable_policy_(false),
@@ -93,12 +93,12 @@ VmInterface::VmInterface(const boost::uuids::uuid &uuid,
                          uint16_t tx_vlan_id, uint16_t rx_vlan_id,
                          Interface *parent, const Ip6Address &a6,
                          DeviceType device_type, VmiType vmi_type) :
-    Interface(Interface::VM_INTERFACE, uuid, name, NULL), vm_(NULL),
+    Interface(Interface::VM_INTERFACE, uuid, name, NULL), vm_(NULL, this),
     vn_(NULL), primary_ip_addr_(addr), mdata_ip_(NULL), subnet_bcast_addr_(0),
     primary_ip6_addr_(a6), vm_mac_(mac), policy_enabled_(false),
     mirror_entry_(NULL), mirror_direction_(MIRROR_RX_TX), cfg_name_(""),
-    fabric_port_(true), need_linklocal_ip_(false), dhcp_enable_(true),
-    do_dhcp_relay_(false), vm_name_(vm_name),
+    fabric_port_(true), need_linklocal_ip_(false), drop_new_flows_(false),
+    dhcp_enable_(true), do_dhcp_relay_(false), vm_name_(vm_name),
     vm_project_uuid_(vm_project_uuid), vxlan_id_(0),
     bridging_(false), layer3_forwarding_(true),
     flood_unknown_unicast_(false), mac_set_(false),
@@ -1912,6 +1912,13 @@ bool VmInterface::CopyConfig(const InterfaceTable *table,
             ret = true;
         }
 
+        bool drop_new_flows =
+            (vm_.get() != NULL) ? vm_->drop_new_flows() : false;
+        if (drop_new_flows_ != drop_new_flows) {
+            drop_new_flows_ = drop_new_flows;
+            ret = true;
+        }
+
         VrfEntry *vrf = table->FindVrfRef(data->vrf_name_);
         if (vrf_.get() != vrf) {
             vrf_ = vrf;
@@ -2496,6 +2503,25 @@ bool VmInterfaceHealthCheckData::OnResync(const InterfaceTable *table,
     return vmi->UpdateIsHealthCheckActive();
 }
 
+VmInterfaceNewFlowDropData::VmInterfaceNewFlowDropData(bool drop_new_flows) :
+    VmInterfaceData(NULL, NULL, DROP_NEW_FLOWS, Interface::TRANSPORT_INVALID),
+    drop_new_flows_(drop_new_flows) {
+}
+
+VmInterfaceNewFlowDropData::~VmInterfaceNewFlowDropData() {
+}
+
+bool VmInterfaceNewFlowDropData::OnResync(const InterfaceTable *table,
+                                          VmInterface *vmi,
+                                          bool *force_update) const {
+    if (vmi->drop_new_flows_ != drop_new_flows_) {
+        vmi->drop_new_flows_ = drop_new_flows_;
+        return true;
+    }
+
+    return false;
+}
+
 /////////////////////////////////////////////////////////////////////////////
 // VM Port Entry utility routines
 /////////////////////////////////////////////////////////////////////////////

src/vnsw/agent/oper/vm_interface.h

@@ -506,6 +506,7 @@ class VmInterface : public Interface {
     const MacAddress &vm_mac() const { return vm_mac_; }
     bool fabric_port() const { return fabric_port_; }
     bool need_linklocal_ip() const { return  need_linklocal_ip_; }
+    bool drop_new_flows() const { return drop_new_flows_; }
     bool dhcp_enable_config() const { return dhcp_enable_; }
     void set_dhcp_enable_config(bool dhcp_enable) {
         dhcp_enable_= dhcp_enable;
@@ -701,6 +702,7 @@ class VmInterface : public Interface {
     friend struct VmInterfaceMirrorData;
     friend struct VmInterfaceGlobalVrouterData;
     friend struct VmInterfaceHealthCheckData;
+    friend struct VmInterfaceNewFlowDropData;
 
     bool IsMetaDataIPActive() const;
     bool IsIpv4Active() const;
@@ -856,7 +858,7 @@ class VmInterface : public Interface {
     void CopyEcmpLoadBalance(EcmpLoadBalance &ecmp_load_balance);
     void UpdateCommonNextHop();
 
-    VmEntryRef vm_;
+    VmEntryBackRef vm_;
     VnEntryRef vn_;
     Ip4Address primary_ip_addr_;
     std::auto_ptr<MetaDataIp> mdata_ip_;
@@ -869,6 +871,7 @@ class VmInterface : public Interface {
     std::string cfg_name_;
     bool fabric_port_;
     bool need_linklocal_ip_;
+    bool drop_new_flows_;
     // DHCP flag - set according to the dhcp option in the ifmap subnet object.
     // It controls whether the vrouter sends the DHCP requests from VM interface
     // to agent or if it would flood the request in the VN.
@@ -978,7 +981,8 @@ struct VmInterfaceData : public InterfaceData {
         IP_ADDR,
         OS_OPER_STATE,
         GLOBAL_VROUTER,
-        HEALTH_CHECK
+        HEALTH_CHECK,
+        DROP_NEW_FLOWS
     };
 
     VmInterfaceData(Agent *agent, IFMapNode *node, Type type,
@@ -1166,4 +1170,13 @@ struct VmInterfaceHealthCheckData : public VmInterfaceData {
                           bool *force_update) const;
 };
 
+struct VmInterfaceNewFlowDropData : public VmInterfaceData {
+    VmInterfaceNewFlowDropData(bool drop_new_flows);
+    virtual ~VmInterfaceNewFlowDropData();
+    virtual bool OnResync(const InterfaceTable *table, VmInterface *vmi,
+                          bool *force_update) const;
+
+    bool drop_new_flows_;
+};
+
 #endif // vnsw_agent_vm_interface_hpp

src/vnsw/agent/pkt/test/test_pkt_flow_limits.cc

@@ -617,9 +617,12 @@ TEST_F(FlowTest, FlowLimit_1) {
     FlowStatsTimerStartStop(agent_, true);
     CreateFlow(flow, 4);
     client->WaitForIdle();
-    int nh_id = agent_->interface_table()->FindInterface
-        (vmi_3->id())->flow_key_nh()->id();
+    VmInterface *intf = static_cast<VmInterface*>(agent_->interface_table()->\
+                                                  FindInterface(vmi_3->id()));
+    int nh_id = intf->flow_key_nh()->id();
     EXPECT_EQ(4U, get_flow_proto()->FlowCount());
+    // Validate interface is set to drop new flows after flow limit is reached
+    EXPECT_TRUE(intf->drop_new_flows());
     FlowEntry *fe = FlowGet(VrfGet("vrf3")->vrf_id(), vm4_ip, vm1_ip,
                             IPPROTO_TCP, 300, 200, nh_id);
     EXPECT_TRUE(fe != NULL && fe->is_flags_set(FlowEntry::ShortFlow) == true &&
@@ -631,6 +634,10 @@ TEST_F(FlowTest, FlowLimit_1) {
     DeleteRoute("vrf5", vm4_ip);
     DeleteRoute("vrf3", vm1_ip);
     client->WaitForIdle();
+    FlushFlowTable();
+    client->WaitForIdle();
+    // Validate interface is not dropping new flows after flows age-out
+    EXPECT_FALSE(intf->drop_new_flows());
     client->WaitForIdle();
     agent_->set_max_vm_flows(vm_flows);
 }

src/vnsw/agent/vrouter/ksync/interface_ksync.cc

@@ -41,6 +41,7 @@ InterfaceKSyncEntry::InterfaceKSyncEntry(InterfaceKSyncObject *obj,
                                          const InterfaceKSyncEntry *entry,
                                          uint32_t index) :
     KSyncNetlinkDBEntry(index), analyzer_name_(entry->analyzer_name_),
+    drop_new_flows_(entry->drop_new_flows_),
     dhcp_enable_(entry->dhcp_enable_),
     fd_(kInvalidIndex),
     flow_key_nh_id_(entry->flow_key_nh_id_),
@@ -80,6 +81,7 @@ InterfaceKSyncEntry::InterfaceKSyncEntry(InterfaceKSyncObject *obj,
                                          const Interface *intf) :
     KSyncNetlinkDBEntry(kInvalidIndex),
     analyzer_name_(),
+    drop_new_flows_(false),
     dhcp_enable_(true),
     fd_(-1),
     flow_key_nh_id_(0),
@@ -210,6 +212,11 @@ bool InterfaceKSyncEntry::Sync(DBEntry *e) {
             ret = true;
         }
 
+        if (drop_new_flows_ != vm_port->drop_new_flows()) {
+            drop_new_flows_ = vm_port->drop_new_flows();
+            ret = true;
+        }
+
         if (dhcp_enable_ != vm_port->dhcp_enabled()) {
             dhcp_enable_ = vm_port->dhcp_enabled();
             ret = true;
@@ -540,6 +547,9 @@ int InterfaceKSyncEntry::Encode(sandesh_op::type op, char *buf, int buf_len) {
     case Interface::VM_INTERFACE: {
         if (vmi_device_type_ == VmInterface::TOR)
             return 0;            
+        if (drop_new_flows_) {
+            flags |= VIF_FLAG_DROP_NEW_FLOWS;
+        }
         if (dhcp_enable_) {
             flags |= VIF_FLAG_DHCP_ENABLED;
         }

src/vnsw/agent/vrouter/ksync/interface_ksync.h

@@ -67,6 +67,7 @@ class InterfaceKSyncEntry : public KSyncNetlinkDBEntry {
     virtual int DeleteMsg(char *buf, int buf_len);
     virtual KSyncEntry *UnresolvedReference();
     void FillObjectLog(sandesh_op::type op, KSyncIntfInfo &info) const;
+    bool drop_new_flows() const {return drop_new_flows_;}
     bool dhcp_enable() const {return dhcp_enable_;}
     bool layer3_forwarding() const {return layer3_forwarding_;}
     bool bridging() const {return bridging_;}
@@ -78,6 +79,7 @@ class InterfaceKSyncEntry : public KSyncNetlinkDBEntry {
     int Encode(sandesh_op::type op, char *buf, int buf_len);
 
     string analyzer_name_;
+    bool drop_new_flows_;
     bool dhcp_enable_;
     uint32_t fd_;       // FD opened for this
     uint32_t flow_key_nh_id_;