Warm reboot: Support portsyncd process warm restart

portsyncd/Makefile.am

@@ -1,4 +1,4 @@
-INCLUDES = -I $(top_srcdir)
+INCLUDES = -I $(top_srcdir) -I $(top_srcdir)/warmrestart
 
 bin_PROGRAMS = portsyncd
 
@@ -8,7 +8,7 @@ else
 DBGFLAGS = -g
 endif
 
-portsyncd_SOURCES = portsyncd.cpp linksync.cpp
+portsyncd_SOURCES = portsyncd.cpp linksync.cpp $(top_srcdir)/warmrestart/warm_restart.cpp $(top_srcdir)/warmrestart/warm_restart.h
 
 portsyncd_CFLAGS = $(DBGFLAGS) $(AM_CFLAGS) $(CFLAGS_COMMON)
 portsyncd_CPPFLAGS = $(DBGFLAGS) $(AM_CFLAGS) $(CFLAGS_COMMON)

portsyncd/linksync.cpp

@@ -13,6 +13,7 @@
 
 #include "linkcache.h"
 #include "portsyncd/linksync.h"
+#include "warm_restart.h"
 
 #include <iostream>
 #include <set>
@@ -41,52 +42,55 @@ LinkSync::LinkSync(DBConnector *appl_db, DBConnector *state_db) :
     m_portTable(appl_db, APP_PORT_TABLE_NAME),
     m_statePortTable(state_db, STATE_PORT_TABLE_NAME)
 {
-    /* See the comments for g_portSet in portsyncd.cpp */
-    for (string port : g_portSet)
+    if (!WarmStart::isWarmStart())
     {
-        vector<FieldValueTuple> temp;
-        if (m_portTable.get(port, temp))
+        /* See the comments for g_portSet in portsyncd.cpp */
+        for (string port : g_portSet)
         {
-            for (auto it : temp)
+            vector<FieldValueTuple> temp;
+            if (m_portTable.get(port, temp))
             {
-                if (fvField(it) == "admin_status")
+                for (auto it : temp)
                 {
-                    g_portSet.erase(port);
-                    break;
+                    if (fvField(it) == "admin_status")
+                    {
+                        g_portSet.erase(port);
+                        break;
+                    }
                 }
             }
         }
-    }
-
-    struct if_nameindex *if_ni, *idx_p;
-    if_ni = if_nameindex();
-    if (if_ni == NULL)
-    {
-        return;
-    }
 
-    for (idx_p = if_ni; ! (idx_p->if_index == 0 && idx_p->if_name == NULL); idx_p++)
-    {
-        string key = idx_p->if_name;
-        if (key.compare(0, INTFS_PREFIX.length(), INTFS_PREFIX))
+        struct if_nameindex *if_ni, *idx_p;
+        if_ni = if_nameindex();
+        if (if_ni == NULL)
         {
-            continue;
+            return;
         }
 
-        m_ifindexOldNameMap[idx_p->if_index] = key;
-
-        /* Bring down the existing kernel interfaces */
-        string cmd, res;
-        SWSS_LOG_INFO("Bring down old interface %s(%d)", key.c_str(), idx_p->if_index);
-        cmd = "ip link set " + key + " down";
-        try
-        {
-            swss::exec(cmd, res);
-        }
-        catch (...)
+        for (idx_p = if_ni; ! (idx_p->if_index == 0 && idx_p->if_name == NULL); idx_p++)
         {
-            /* Ignore error in this flow ; */
-            SWSS_LOG_WARN("Failed to bring down old interface %s(%d)", key.c_str(), idx_p->if_index);
+            string key = idx_p->if_name;
+            if (key.compare(0, INTFS_PREFIX.length(), INTFS_PREFIX))
+            {
+                continue;
+            }
+
+            m_ifindexOldNameMap[idx_p->if_index] = key;
+
+            /* Bring down the existing kernel interfaces */
+            string cmd, res;
+            SWSS_LOG_INFO("Bring down old interface %s(%d)", key.c_str(), idx_p->if_index);
+            cmd = "ip link set " + key + " down";
+            try
+            {
+                swss::exec(cmd, res);
+            }
+            catch (...)
+            {
+                /* Ignore error in this flow ; */
+                SWSS_LOG_WARN("Failed to bring down old interface %s(%d)", key.c_str(), idx_p->if_index);
+            }
         }
     }
 }

portsyncd/portsyncd.cpp

@@ -6,6 +6,7 @@
 #include <set>
 #include <map>
 #include <list>
+#include <sys/stat.h>
 #include "dbconnector.h"
 #include "select.h"
 #include "netdispatcher.h"
@@ -14,6 +15,7 @@
 #include "portsyncd/linksync.h"
 #include "subscriberstatetable.h"
 #include "exec.h"
+#include "warm_restart.h"
 
 using namespace std;
 using namespace swss;
@@ -42,6 +44,7 @@ void handlePortConfigFile(ProducerStateTable &p, string file);
 void handlePortConfigFromConfigDB(ProducerStateTable &p, DBConnector &cfgDb);
 void handleVlanIntfFile(string file);
 void handlePortConfig(ProducerStateTable &p, map<string, KeyOpFieldsValuesTuple> &port_cfg_map);
+void checkPortInitDone(DBConnector *appl_db);
 
 int main(int argc, char **argv)
 {
@@ -72,6 +75,20 @@ int main(int argc, char **argv)
     ProducerStateTable p(&appl_db, APP_PORT_TABLE_NAME);
     SubscriberStateTable portCfg(&cfgDb, CFG_PORT_TABLE_NAME);
 
+    WarmStart::checkWarmStart("portsyncd");
+    if (WarmStart::isWarmStart())
+    {
+        checkPortInitDone(&appl_db);
+        /* If PortInitDone already set, clear the init port config buffer */
+        if(g_init)
+        {
+            cout << "portsyncd warm start" << endl;
+            deque<KeyOpFieldsValuesTuple> vkco;
+            portCfg.pops(vkco);
+        }
+
+    }
+
     LinkSync sync(&appl_db, &state_db);
     NetDispatcher::getInstance().registerMessageHandler(RTM_NEWLINK, &sync);
     NetDispatcher::getInstance().registerMessageHandler(RTM_DELLINK, &sync);
@@ -84,15 +101,20 @@ int main(int argc, char **argv)
         netlink.registerGroup(RTNLGRP_LINK);
         cout << "Listen to link messages..." << endl;
 
-        if (!port_config_file.empty())
+        /* For portsyncd warm start, don't process init port config again if PortInitDone set */
+        if (!g_init)
         {
-            handlePortConfigFile(p, port_config_file);
-        } else {
-            handlePortConfigFromConfigDB(p, cfgDb);
+            if (!port_config_file.empty())
+            {
+                handlePortConfigFile(p, port_config_file);
+            } else {
+                handlePortConfigFromConfigDB(p, cfgDb);
+            }
         }
 
         s.addSelectable(&netlink);
         s.addSelectable(&portCfg);
+
         while (true)
         {
             Selectable *temps;
@@ -314,3 +336,17 @@ void handlePortConfig(ProducerStateTable &p, map<string, KeyOpFieldsValuesTuple>
         }
     }
 }
+
+void checkPortInitDone(DBConnector *appl_db)
+{
+    std::unique_ptr<Table>  portTable = std::unique_ptr<Table>(new Table(appl_db, APP_PORT_TABLE_NAME));
+    std::vector<FieldValueTuple> vfv;
+
+    if (portTable->get("PortInitDone", vfv))
+    {
+        if (!vfv.empty())
+        {
+            g_init = true;
+        }
+    }
+}

tests/test_warm_reboot.py

@@ -0,0 +1,172 @@
+from swsscommon import swsscommon
+import os
+import re
+import time
+import json
+
+# Get restart count of all processes supporting warm restart
+def swss_get_RestartCount(state_db):
+    restart_count = {}
+    warmtbl = swsscommon.Table(state_db, swsscommon.STATE_WARM_RESTART_TABLE_NAME)
+    keys = warmtbl.getKeys()
+    assert  len(keys) !=  0
+    for key in keys:
+        (status, fvs) = warmtbl.get(key)
+        assert status == True
+        for fv in fvs:
+            if fv[0] == "restart_count":
+                restart_count[key] = int(fv[1])
+    print(restart_count)
+    return restart_count
+
+# function to check the restart count incremented by 1 for all processes supporting warm restart
+def swss_check_RestartCount(state_db, restart_count):
+    warmtbl = swsscommon.Table(state_db, swsscommon.STATE_WARM_RESTART_TABLE_NAME)
+    keys = warmtbl.getKeys()
+    print(keys)
+    assert  len(keys) > 0
+    for key in keys:
+        (status, fvs) = warmtbl.get(key)
+        assert status == True
+        for fv in fvs:
+            if fv[0] == "restart_count":
+                assert int(fv[1]) == restart_count[key] + 1
+            elif fv[0] == "state":
+                assert fv[1] == "reconciled"
+
+# function to check the restart count incremented by 1 for a single process
+def swss_app_check_RestartCount_single(state_db, restart_count, name):
+    warmtbl = swsscommon.Table(state_db, swsscommon.STATE_WARM_RESTART_TABLE_NAME)
+    keys = warmtbl.getKeys()
+    print(keys)
+    print(restart_count)
+    assert  len(keys) > 0
+    for key in keys:
+        if key != name:
+            continue
+        (status, fvs) = warmtbl.get(key)
+        assert status == True
+        for fv in fvs:
+            if fv[0] == "restart_count":
+                assert int(fv[1]) == restart_count[key] + 1
+            elif fv[0] == "state":
+                assert fv[1] == "reconciled"
+
+def check_port_oper_status(appl_db, port_name, state):
+    portTbl = swsscommon.Table(appl_db, "PORT_TABLE")
+    (status, fvs) = portTbl.get(port_name)
+    assert status == True
+
+    oper_status = "unknown"
+    for v in fvs:
+        if v[0] == "oper_status":
+            oper_status = v[1]
+            break
+    assert oper_status == state
+
+def create_entry(tbl, key, pairs):
+    fvs = swsscommon.FieldValuePairs(pairs)
+    tbl.set(key, fvs)
+     # FIXME: better to wait until DB create them
+    time.sleep(1)
+def create_entry_tbl(db, table, key, pairs):
+    tbl = swsscommon.Table(db, table)
+    create_entry(tbl, key, pairs)
+def del_entry_tbl(db, table, key):
+    tbl = swsscommon.Table(db, table)
+    tbl._del(key)
+def create_entry_pst(db, table, key, pairs):
+    tbl = swsscommon.ProducerStateTable(db, table)
+    create_entry(tbl, key, pairs)
+def how_many_entries_exist(db, table):
+    tbl =  swsscommon.Table(db, table)
+    return len(tbl.getKeys())
+
+
+def test_PortSyncdWarmRestart(dvs):
+
+    conf_db = swsscommon.DBConnector(swsscommon.CONFIG_DB, dvs.redis_sock, 0)
+    appl_db = swsscommon.DBConnector(swsscommon.APPL_DB, dvs.redis_sock, 0)
+    state_db = swsscommon.DBConnector(swsscommon.STATE_DB, dvs.redis_sock, 0)
+
+    # enable warm restart
+    # TODO: use cfg command to config it
+    create_entry_tbl(
+        conf_db,
+        swsscommon.CFG_WARM_RESTART_TABLE_NAME, "swss",
+        [
+            ("enable", "true"),
+        ]
+    )
+
+    dvs.runcmd("ifconfig Ethernet16  up")
+    dvs.runcmd("ifconfig Ethernet20  up")
+
+    time.sleep(1)
+
+    dvs.runcmd("ifconfig Ethernet16 11.0.0.1/29 up")
+    dvs.runcmd("ifconfig Ethernet20 11.0.0.9/29 up")
+
+    dvs.servers[4].runcmd("ip link set down dev eth0") == 0
+    dvs.servers[4].runcmd("ip link set up dev eth0") == 0
+    dvs.servers[4].runcmd("ifconfig eth0 11.0.0.2/29")
+    dvs.servers[4].runcmd("ip route add default via 11.0.0.1")
+
+    dvs.servers[5].runcmd("ip link set down dev eth0") == 0
+    dvs.servers[5].runcmd("ip link set up dev eth0") == 0
+    dvs.servers[5].runcmd("ifconfig eth0 11.0.0.10/29")
+    dvs.servers[5].runcmd("ip route add default via 11.0.0.9")
+
+    time.sleep(1)
+
+    # Ethernet port oper status should be up
+    check_port_oper_status(appl_db, "Ethernet16", "up")
+    check_port_oper_status(appl_db, "Ethernet20", "up")
+
+    # Ping should work between servers via vs vlan interfaces
+    ping_stats = dvs.servers[4].runcmd("ping -c 1 11.0.0.10")
+    time.sleep(1)
+
+    neighTbl = swsscommon.Table(appl_db, "NEIGH_TABLE")
+    (status, fvs) = neighTbl.get("Ethernet16:11.0.0.2")
+    assert status == True
+
+    (status, fvs) = neighTbl.get("Ethernet20:11.0.0.10")
+    assert status == True
+
+    restart_count = swss_get_RestartCount(state_db)
+
+    # restart portsyncd
+    dvs.runcmd(['sh', '-c', 'pkill -x portsyncd; cp /var/log/swss/sairedis.rec /var/log/swss/sairedis.rec.b; echo > /var/log/swss/sairedis.rec'])
+    dvs.runcmd(['sh', '-c', 'supervisorctl start portsyncd'])
+    time.sleep(2)
+
+     # No create/set/remove operations should be passed down to syncd for portsyncd warm restart
+    num = dvs.runcmd(['sh', '-c', 'grep \|c\| /var/log/swss/sairedis.rec | wc -l'])
+    assert num == '0\n'
+    num = dvs.runcmd(['sh', '-c', 'grep \|s\| /var/log/swss/sairedis.rec | wc -l'])
+    assert num == '0\n'
+    num = dvs.runcmd(['sh', '-c', 'grep \|r\| /var/log/swss/sairedis.rec | wc -l'])
+    assert num == '0\n'
+
+    #new ip on server 5
+    dvs.servers[5].runcmd("ifconfig eth0 11.0.0.11/29")
+
+    # Ping should work between servers via vs Ethernet interfaces
+    ping_stats = dvs.servers[4].runcmd("ping -c 1 11.0.0.11")
+
+    # new neighbor learn on VS
+    (status, fvs) = neighTbl.get("Ethernet20:11.0.0.11")
+    assert status == True
+
+    # Port state change reflected in appDB correctly
+    dvs.servers[6].runcmd("ip link set down dev eth0") == 0
+    dvs.servers[6].runcmd("ip link set up dev eth0") == 0
+    time.sleep(1)
+
+    check_port_oper_status(appl_db, "Ethernet16", "up")
+    check_port_oper_status(appl_db, "Ethernet20", "up")
+    check_port_oper_status(appl_db, "Ethernet24", "up")
+
+
+    swss_app_check_RestartCount_single(state_db, restart_count, "portsyncd")