Warm reboot: store FDB entries and warm start

orchagent/fdborch.cpp

@@ -19,13 +19,14 @@ extern CrmOrch *        gCrmOrch;
 
 const int fdborch_pri = 20;
 
-FdbOrch::FdbOrch(DBConnector *db, string tableName, PortsOrch *port) :
-    Orch(db, tableName, fdborch_pri),
+FdbOrch::FdbOrch(TableConnector applDbConnector, TableConnector stateDbConnector, PortsOrch *port) :
+    Orch(applDbConnector.first, applDbConnector.second, fdborch_pri),
     m_portsOrch(port),
-    m_table(Table(db, tableName))
+    m_table(applDbConnector.first, applDbConnector.second),
+    m_fdbStateTable(stateDbConnector.first, stateDbConnector.second)
 {
     m_portsOrch->attach(this);
-    m_flushNotificationsConsumer = new NotificationConsumer(db, "FLUSHFDBREQUEST");
+    m_flushNotificationsConsumer = new NotificationConsumer(applDbConnector.first, "FLUSHFDBREQUEST");
     auto flushNotifier = new Notifier(m_flushNotificationsConsumer, this, "FLUSHFDBREQUEST");
     Orch::addExecutor(flushNotifier);
 
@@ -36,6 +37,73 @@ FdbOrch::FdbOrch(DBConnector *db, string tableName, PortsOrch *port) :
     Orch::addExecutor(fdbNotifier);
 }
 
+bool FdbOrch::bake()
+{
+    Orch::bake();
+
+    auto consumer = dynamic_cast<Consumer *>(getExecutor(APP_FDB_TABLE_NAME));
+    if (consumer == NULL)
+    {
+        SWSS_LOG_ERROR("No consumer %s in Orch", APP_FDB_TABLE_NAME);
+        return false;
+    }
+
+    size_t refilled = consumer->refillToSync(&m_fdbStateTable);
+    SWSS_LOG_NOTICE("Add warm input FDB State: %s, %zd", APP_FDB_TABLE_NAME, refilled);
+    return true;
+}
+
+bool FdbOrch::storeFdbEntryState(const FdbUpdate& update)
+{
+    const FdbEntry& entry = update.entry;
+    const Port& port = update.port;
+    sai_vlan_id_t vlan_id = port.m_port_vlan_id;
+    const MacAddress& mac = entry.mac;
+    string portName = port.m_alias;
+
+    // ref: https://github.com/Azure/sonic-swss/blob/master/doc/swss-schema.md#fdb_table
+    string key = "Vlan" + to_string(vlan_id) + ":" + mac.to_string();
+
+    if (update.add)
+    {
+        auto inserted = m_entries.insert(entry);
+
+        SWSS_LOG_DEBUG("FdbOrch notification: mac %s was inserted into bv_id 0x%lx",
+                        entry.mac.to_string().c_str(), entry.bv_id);
+
+        if (!inserted.second)
+        {
+            SWSS_LOG_INFO("FdbOrch notification: mac %s is duplicate", entry.mac.to_string().c_str());
+            return false;
+        }
+
+        // Write to StateDb
+        std::vector<FieldValueTuple> fvs;
+        fvs.push_back(FieldValueTuple("port", portName));
+        fvs.push_back(FieldValueTuple("type", "dynamic"));
+        m_fdbStateTable.set(key, fvs);
+
+        gCrmOrch->incCrmResUsedCounter(CrmResourceType::CRM_FDB_ENTRY);
+        return true;
+    }
+    else
+    {
+        size_t erased = m_entries.erase(entry);
+        SWSS_LOG_DEBUG("FdbOrch notification: mac %s was removed from bv_id 0x%lx", entry.mac.to_string().c_str(), entry.bv_id);
+
+        if (erased == 0)
+        {
+            return false;
+        }
+
+        // Remove in StateDb
+        m_fdbStateTable.del(key);
+
+        gCrmOrch->decCrmResUsedCounter(CrmResourceType::CRM_FDB_ENTRY);
+        return true;
+    }
+}
+
 void FdbOrch::update(sai_fdb_event_t type, const sai_fdb_entry_t* entry, sai_object_id_t bridge_port_id)
 {
     SWSS_LOG_ENTER();
@@ -62,22 +130,7 @@ void FdbOrch::update(sai_fdb_event_t type, const sai_fdb_entry_t* entry, sai_obj
         }
 
         update.add = true;
-
-        {
-            auto ret = m_entries.insert(update.entry);
-
-            SWSS_LOG_DEBUG("FdbOrch notification: mac %s was inserted into bv_id 0x%lx",
-                            update.entry.mac.to_string().c_str(), entry->bv_id);
-
-            if (ret.second)
-            {
-                gCrmOrch->incCrmResUsedCounter(CrmResourceType::CRM_FDB_ENTRY);
-            }
-            else
-            {
-                SWSS_LOG_INFO("FdbOrch notification: mac %s is duplicate", update.entry.mac.to_string().c_str());
-            }
-        }
+        storeFdbEntryState(update);
 
         for (auto observer: m_observers)
         {
@@ -89,16 +142,7 @@ void FdbOrch::update(sai_fdb_event_t type, const sai_fdb_entry_t* entry, sai_obj
     case SAI_FDB_EVENT_AGED:
     case SAI_FDB_EVENT_MOVE:
         update.add = false;
-
-        {
-            auto ret = m_entries.erase(update.entry);
-            SWSS_LOG_DEBUG("FdbOrch notification: mac %s was removed from bv_id 0x%lx", update.entry.mac.to_string().c_str(), entry->bv_id);
-
-            if (ret)
-            {
-                gCrmOrch->decCrmResUsedCounter(CrmResourceType::CRM_FDB_ENTRY);
-            }
-        }
+        storeFdbEntryState(update);
 
         for (auto observer: m_observers)
         {
@@ -121,13 +165,12 @@ void FdbOrch::update(sai_fdb_event_t type, const sai_fdb_entry_t* entry, sai_obj
                 update.entry.mac = itr->mac;
                 update.entry.bv_id = itr->bv_id;
                 update.add = false;
+                itr++;
 
-                itr = m_entries.erase(itr);
+                storeFdbEntryState(update);
 
                 SWSS_LOG_DEBUG("FdbOrch notification: mac %s was removed", update.entry.mac.to_string().c_str());
 
-                gCrmOrch->decCrmResUsedCounter(CrmResourceType::CRM_FDB_ENTRY);
-
                 for (auto observer: m_observers)
                 {
                     observer->update(SUBJECT_TYPE_FDB_CHANGE, &update);

orchagent/fdborch.h

@@ -34,14 +34,15 @@ typedef unordered_map<string, vector<SavedFdbEntry>> fdb_entries_by_port_t;
 class FdbOrch: public Orch, public Subject, public Observer
 {
 public:
-    
-    FdbOrch(DBConnector *db, string tableName, PortsOrch *port);
+
+    FdbOrch(TableConnector applDbConnector, TableConnector stateDbConnector, PortsOrch *port);
 
     ~FdbOrch()
     {
         m_portsOrch->detach(this);
     }
 
+    bool bake() override;
     void update(sai_fdb_event_t, const sai_fdb_entry_t *, sai_object_id_t);
     void update(SubjectType type, void *cntx);
     bool getPort(const MacAddress&, uint16_t, Port&);
@@ -51,6 +52,7 @@ class FdbOrch: public Orch, public Subject, public Observer
     set<FdbEntry> m_entries;
     fdb_entries_by_port_t saved_fdb_entries;
     Table m_table;
+    Table m_fdbStateTable;
     NotificationConsumer* m_flushNotificationsConsumer;
     NotificationConsumer* m_fdbNotificationConsumer;
 
@@ -60,6 +62,8 @@ class FdbOrch: public Orch, public Subject, public Observer
     void updateVlanMember(const VlanMemberUpdate&);
     bool addFdbEntry(const FdbEntry&, const string&, const string&);
     bool removeFdbEntry(const FdbEntry&);
+
+    bool storeFdbEntryState(const FdbUpdate& update);
 };
 
 #endif /* SWSS_FDBORCH_H */

orchagent/orchdaemon.cpp

@@ -69,7 +69,11 @@ bool OrchDaemon::init()
 
     gCrmOrch = new CrmOrch(m_configDb, CFG_CRM_TABLE_NAME);
     gPortsOrch = new PortsOrch(m_applDb, ports_tables);
-    gFdbOrch = new FdbOrch(m_applDb, APP_FDB_TABLE_NAME, gPortsOrch);
+
+    TableConnector applDbFdb(m_applDb, APP_FDB_TABLE_NAME);
+    TableConnector stateDbFdb(m_stateDb, STATE_FDB_TABLE_NAME);
+    gFdbOrch = new FdbOrch(applDbFdb, stateDbFdb, gPortsOrch);
+
     gIntfsOrch = new IntfsOrch(m_applDb, APP_INTF_TABLE_NAME);
     gNeighOrch = new NeighOrch(m_applDb, APP_NEIGH_TABLE_NAME, gIntfsOrch);
     gRouteOrch = new RouteOrch(m_applDb, APP_ROUTE_TABLE_NAME, gNeighOrch);
@@ -454,7 +458,7 @@ bool OrchDaemon::warmRestoreValidation()
         }
     }
     WarmStart::setWarmStartState("orchagent", WarmStart::RESTORED);
-    return true;
+    return ts.empty();
 }
 
 /*

tests/conftest.py

@@ -2,6 +2,8 @@
 import os.path
 import re
 import time
+import json
+import redis
 import docker
 import pytest
 import commands
@@ -117,7 +119,7 @@ def __del__(self):
             os.system("ip netns delete %s" % self.nsname)
 
     def runcmd(self, cmd):
-        os.system("ip netns exec %s %s" % (self.nsname, cmd))
+        return os.system("ip netns exec %s %s" % (self.nsname, cmd))
 
     def runcmd_async(self, cmd):
         return subprocess.Popen("ip netns exec %s %s" % (self.nsname, cmd), shell=True)
@@ -267,6 +269,167 @@ def copy_file(self, path, filename):
         self.ctn.put_archive(path, tarstr.getvalue())
         tarstr.close()
 
+    def is_table_entry_exists(self, db, table, keyregex, attributes):
+        tbl = swsscommon.Table(db, table)
+        keys = tbl.getKeys()
+
+        exists = False
+        extra_info = []
+        key_found = False
+        for key in keys:
+            key_found = re.match(keyregex, key)
+
+            status, fvs = tbl.get(key)
+            assert status, "Error reading from table %s" % table
+
+            d_attributes = dict(attributes)
+            for k, v in fvs:
+                if k in d_attributes and d_attributes[k] == v:
+                    del d_attributes[k]
+
+            if len(d_attributes) != 0:
+                exists = False
+                extra_info.append("Desired attributes %s was not found for key %s" % (str(d_attributes), key))
+            else:
+                exists = True
+                break
+
+        if not key_found:
+            exists = False
+            extra_info.append("Desired key with parameters %s was not found" % str(key_values))
+
+        return exists, extra_info
+
+    def is_fdb_entry_exists(self, db, table, key_values, attributes):
+        tbl =  swsscommon.Table(db, table)
+        keys = tbl.getKeys()
+
+        exists = False
+        extra_info = []
+        key_found = False
+        for key in keys:
+            try:
+                d_key = json.loads(key)
+            except ValueError:
+                d_key = json.loads('{' + key + '}')
+
+            for k, v in key_values:
+                if k not in d_key or v != d_key[k]:
+                    continue
+
+            key_found = True
+
+            status, fvs = tbl.get(key)
+            assert status, "Error reading from table %s" % table
+
+            d_attributes = dict(attributes)
+            for k, v in fvs:
+                if k in d_attributes and d_attributes[k] == v:
+                    del d_attributes[k]
+
+            if len(d_attributes) != 0:
+                exists = False
+                extra_info.append("Desired attributes %s was not found for key %s" % (str(d_attributes), key))
+            else:
+                exists = True
+                break
+
+        if not key_found:
+            exists = False
+            extra_info.append("Desired key with parameters %s was not found" % str(key_values))
+
+        return exists, extra_info
+
+    def create_vlan(self, vlan):
+        tbl = swsscommon.Table(self.cdb, "VLAN")
+        fvs = swsscommon.FieldValuePairs([("vlanid", vlan)])
+        tbl.set("Vlan" + vlan, fvs)
+        time.sleep(1)
+
+    def create_vlan_member(self, vlan, interface):
+        tbl = swsscommon.Table(self.cdb, "VLAN_MEMBER")
+        fvs = swsscommon.FieldValuePairs([("tagging_mode", "untagged")])
+        tbl.set("Vlan" + vlan + "|" + interface, fvs)
+        time.sleep(1)
+
+    def set_interface_status(self, interface, admin_status):
+        if interface.startswith("PortChannel"):
+            tbl_name = "PORTCHANNEL"
+        elif interface.startswith("Vlan"):
+            tbl_name = "VLAN"
+        else:
+            tbl_name = "PORT"
+        tbl = swsscommon.Table(self.cdb, tbl_name)
+        fvs = swsscommon.FieldValuePairs([("admin_status", "up")])
+        tbl.set(interface, fvs)
+        time.sleep(1)
+
+    def add_ip_address(self, interface, ip):
+        if interface.startswith("PortChannel"):
+            tbl_name = "PORTCHANNEL_INTERFACE"
+        elif interface.startswith("Vlan"):
+            tbl_name = "VLAN_INTERFACE"
+        else:
+            tbl_name = "INTERFACE"
+        tbl = swsscommon.Table(self.cdb, tbl_name)
+        fvs = swsscommon.FieldValuePairs([("NULL", "NULL")])
+        tbl.set(interface + "|" + ip, fvs)
+        time.sleep(1)
+
+    def add_neighbor(self, interface, ip, mac):
+        tbl = swsscommon.ProducerStateTable(self.pdb, "NEIGH_TABLE")
+        fvs = swsscommon.FieldValuePairs([("neigh", mac),
+                                          ("family", "IPv4")])
+        tbl.set(interface + ":" + ip, fvs)
+        time.sleep(1)
+
+    def setup_db(self):
+        self.pdb = swsscommon.DBConnector(0, self.redis_sock, 0)
+        self.adb = swsscommon.DBConnector(1, self.redis_sock, 0)
+        self.cdb = swsscommon.DBConnector(4, self.redis_sock, 0)
+        self.sdb = swsscommon.DBConnector(6, self.redis_sock, 0)
+
+    def getCrmCounterValue(self, key, counter):
+        counters_db = swsscommon.DBConnector(swsscommon.COUNTERS_DB, self.redis_sock, 0)
+        crm_stats_table = swsscommon.Table(counters_db, 'CRM')
+
+        for k in crm_stats_table.get(key)[1]:
+            if k[0] == counter:
+                return int(k[1])
+
+    def setReadOnlyAttr(self, obj, attr, val):
+        db = swsscommon.DBConnector(swsscommon.ASIC_DB, self.redis_sock, 0)
+        tbl = swsscommon.Table(db, "ASIC_STATE:{0}".format(obj))
+        keys = tbl.getKeys()
+
+        assert len(keys) == 1
+
+        swVid = keys[0]
+        r = redis.Redis(unix_socket_path=self.redis_sock, db=swsscommon.ASIC_DB)
+        swRid = r.hget("VIDTORID", swVid)
+
+        assert swRid is not None
+
+        ntf = swsscommon.NotificationProducer(db, "SAI_VS_UNITTEST_CHANNEL")
+        fvp = swsscommon.FieldValuePairs()
+        ntf.send("enable_unittests", "true", fvp)
+        fvp = swsscommon.FieldValuePairs([(attr, val)])
+        key = "SAI_OBJECT_TYPE_SWITCH:" + swRid
+
+        ntf.send("set_ro", key, fvp)
+
+    # start processes in SWSS
+    def start_swss(self):
+        self.runcmd(['sh', '-c', 'supervisorctl start orchagent; supervisorctl start portsyncd; supervisorctl start intfsyncd; \
+            supervisorctl start neighsyncd; supervisorctl start intfmgrd; supervisorctl start vlanmgrd; \
+            supervisorctl start buffermgrd; supervisorctl start arp_update'])
+
+    # stop processes in SWSS
+    def stop_swss(self):
+        self.runcmd(['sh', '-c', 'supervisorctl stop orchagent; supervisorctl stop portsyncd; supervisorctl stop intfsyncd; \
+            supervisorctl stop neighsyncd;  supervisorctl stop intfmgrd; supervisorctl stop vlanmgrd; \
+            supervisorctl stop buffermgrd; supervisorctl stop arp_update'])
+
 @pytest.yield_fixture(scope="module")
 def dvs(request):
     name = request.config.getoption("--dvsname")