Add orchagent pre-warm-restart check mechanism

Signed-off-by: Jipan Yang <[email protected]>

Author: jipanyang

.gitignore

@@ -52,6 +52,7 @@ neighsyncd/neighsyncd
 portsyncd/portsyncd
 orchagent/orchagent
 orchagent/routeresync
+orchagent/orchagent_restart_check
 swssconfig/swssconfig
 swssconfig/swssplayer
 tests/tests

orchagent/Makefile.am

@@ -10,7 +10,7 @@ dist_swss_DATA = \
 		 pfc_detect_barefoot.lua \
 		 pfc_restore.lua
 
-bin_PROGRAMS = orchagent routeresync
+bin_PROGRAMS = orchagent routeresync orchagent_restart_check
 
 if DEBUG
 DBGFLAGS = -ggdb -DDEBUG
@@ -81,3 +81,7 @@ routeresync_SOURCES = routeresync.cpp
 routeresync_CFLAGS = $(DBGFLAGS) $(AM_CFLAGS) $(CFLAGS_COMMON)
 routeresync_CPPFLAGS = $(DBGFLAGS) $(AM_CFLAGS) $(CFLAGS_COMMON)
 routeresync_LDADD = -lswsscommon
+
+orchagent_restart_check_SOURCES = orchagent_restart_check.cpp
+orchagent_restart_check_CPPFLAGS = $(DBGFLAGS) $(AM_CPPFLAGS) $(CFLAGS_COMMON)
+orchagent_restart_check_LDADD = -lhiredis -lswsscommon -lpthread

orchagent/orch.cpp

@@ -185,6 +185,30 @@ void Consumer::drain()
         m_orch->doTask(*this);
 }
 
+string Consumer::dumpTuple(KeyOpFieldsValuesTuple &tuple)
+{
+    string s = getTableName() + getConsumerTable()->getTableNameSeparator() + kfvKey(tuple)
+               + "|" + kfvOp(tuple);
+    for (auto i = kfvFieldsValues(tuple).begin(); i != kfvFieldsValues(tuple).end(); i++)
+    {
+        s += "|" + fvField(*i) + ":" + fvValue(*i);
+    }
+
+    return s;
+}
+
+void Consumer::dumpToSyncTasks(vector<string> &ts)
+{
+    for (auto &tm :m_toSync)
+    {
+        KeyOpFieldsValuesTuple& tuple = tm.second;
+
+        string s = dumpTuple(tuple);
+
+        ts.push_back(s);
+    }
+}
+
 size_t Orch::addExistingData(const string& tableName)
 {
     auto consumer = dynamic_cast<Consumer *>(getExecutor(tableName));
@@ -224,7 +248,7 @@ bool Orch::bake()
         {
             continue;
         }
-        
+
         size_t refilled = consumer->refillToSync();
         SWSS_LOG_NOTICE("Add warm input: %s, %zd", executorName.c_str(), refilled);
     }
@@ -326,6 +350,21 @@ void Orch::doTask()
     }
 }
 
+void Orch::dumpToSyncTasks(vector<string> &ts)
+{
+    for(auto &it : m_consumerMap)
+    {
+        Consumer* consumer = dynamic_cast<Consumer *>(it.second.get());
+        if (consumer == NULL)
+        {
+            SWSS_LOG_DEBUG("Executor is not a Consumer");
+            continue;
+        }
+
+        consumer->dumpToSyncTasks(ts);
+    }
+}
+
 void Orch::logfileReopen()
 {
     gRecordOfs.close();
@@ -347,12 +386,7 @@ void Orch::logfileReopen()
 
 void Orch::recordTuple(Consumer &consumer, KeyOpFieldsValuesTuple &tuple)
 {
-    string s = consumer.getTableName() + ":" + kfvKey(tuple)
-               + "|" + kfvOp(tuple);
-    for (auto i = kfvFieldsValues(tuple).begin(); i != kfvFieldsValues(tuple).end(); i++)
-    {
-        s += "|" + fvField(*i) + ":" + fvValue(*i);
-    }
+    string s = consumer.dumpTuple(tuple);
 
     gRecordOfs << getTimestamp() << "|" << s << endl;
 
@@ -366,13 +400,7 @@ void Orch::recordTuple(Consumer &consumer, KeyOpFieldsValuesTuple &tuple)
 
 string Orch::dumpTuple(Consumer &consumer, KeyOpFieldsValuesTuple &tuple)
 {
-    string s = consumer.getTableName() + ":" + kfvKey(tuple)
-               + "|" + kfvOp(tuple);
-    for (auto i = kfvFieldsValues(tuple).begin(); i != kfvFieldsValues(tuple).end(); i++)
-    {
-        s += "|" + fvField(*i) + ":" + fvValue(*i);
-    }
-
+    string s = consumer.dumpTuple(tuple);
     return s;
 }
 

orchagent/orch.h

@@ -116,6 +116,9 @@ class Consumer : public Executor {
         return getConsumerTable()->getTableName();
     }
 
+    string dumpTuple(KeyOpFieldsValuesTuple &tuple);
+    void dumpToSyncTasks(vector<string> &ts);
+
     size_t refillToSync();
     size_t refillToSync(Table* table);
     void execute();
@@ -124,7 +127,7 @@ class Consumer : public Executor {
     /* Store the latest 'golden' status */
     // TODO: hide?
     SyncMap m_toSync;
-    
+
 protected:
     // Returns: the number of entries added to m_toSync
     size_t addToSync(std::deque<KeyOpFieldsValuesTuple> &entries);
@@ -162,7 +165,7 @@ class Orch
     // Prepare for warm start if Redis contains valid input data
     // otherwise fallback to cold start
     virtual bool bake();
-    
+
     /* Iterate all consumers in m_consumerMap and run doTask(Consumer) */
     void doTask();
 
@@ -173,6 +176,8 @@ class Orch
 
     /* TODO: refactor recording */
     static void recordTuple(Consumer &consumer, KeyOpFieldsValuesTuple &tuple);
+
+    void dumpToSyncTasks(vector<string> &ts);
 protected:
     ConsumerMap m_consumerMap;
 

orchagent/orchagent_restart_check.cpp

@@ -0,0 +1,58 @@
+#include <iostream>
+#include <sstream>
+
+#include <unistd.h>
+#include <getopt.h>
+
+#include "notificationproducer.h"
+#include "notificationconsumer.h"
+#include "select.h"
+#include "logger.h"
+
+int main(int argc, char **argv)
+{
+    swss::Logger::getInstance().setMinPrio(swss::Logger::SWSS_NOTICE);
+    SWSS_LOG_ENTER();
+
+    std::string op = "orchagent";
+
+    swss::DBConnector db(APPL_DB, swss::DBConnector::DEFAULT_UNIXSOCKET, 0);
+    swss::NotificationProducer restartQuery(&db, "RESTARTCHECK");
+
+    swss::NotificationConsumer restartQueryReply(&db, "RESTARTCHECKREPLY");
+    swss::Select s;
+    s.addSelectable(&restartQueryReply);
+    swss::Selectable *sel;
+
+    std::vector<swss::FieldValueTuple> values;
+    SWSS_LOG_NOTICE("requested %s to do warm restart state check and freeze if ready", op.c_str());
+    restartQuery.send(op, op, values);
+
+    int result = s.select(&sel, 3000);
+    if (result == swss::Select::OBJECT)
+    {
+        std::string op_ret, data;
+
+        restartQueryReply.pop(op_ret, data, values);
+        if (op_ret == "READY")
+        {
+            SWSS_LOG_NOTICE("RESTARTCHECK success, %s is frozen and ready for warm restart", op.c_str());
+            std::cout << "RESTARTCHECK succeeded" << std::endl;
+            return EXIT_SUCCESS;
+        }
+        else
+        {
+            SWSS_LOG_NOTICE("RESTARTCHECK failed, %s is not ready for warm restart", op.c_str());
+        }
+    }
+    else if (result == swss::Select::TIMEOUT)
+    {
+        SWSS_LOG_NOTICE("RESTARTCHECK for %s timed out", op.c_str());
+    }
+    else
+    {
+        SWSS_LOG_NOTICE("RESTARTCHECK for %s error", op.c_str());
+    }
+    std::cout << "RESTARTCHECK failed" << std::endl;
+    return EXIT_FAILURE;
+}

orchagent/orchdaemon.cpp

@@ -3,6 +3,8 @@
 #include "orchdaemon.h"
 #include "logger.h"
 #include <sairedis.h>
+#include <limits.h>
+#include "notificationproducer.h"
 
 #define SAI_SWITCH_ATTR_CUSTOM_RANGE_BASE SAI_SWITCH_ATTR_CUSTOM_RANGE_START
 #include "sairedis.h"
@@ -27,6 +29,7 @@ RouteOrch *gRouteOrch;
 AclOrch *gAclOrch;
 CrmOrch *gCrmOrch;
 BufferOrch *gBufferOrch;
+SwitchOrch *gSwitchOrch;
 
 OrchDaemon::OrchDaemon(DBConnector *applDb, DBConnector *configDb, DBConnector *stateDb) :
         m_applDb(applDb),
@@ -49,7 +52,7 @@ bool OrchDaemon::init()
 
     string platform = getenv("platform") ? getenv("platform") : "";
 
-    SwitchOrch *switch_orch = new SwitchOrch(m_applDb, APP_SWITCH_TABLE_NAME);
+    gSwitchOrch = new SwitchOrch(m_applDb, APP_SWITCH_TABLE_NAME);
 
     const int portsorch_base_pri = 40;
 
@@ -116,7 +119,7 @@ bool OrchDaemon::init()
         CFG_DTEL_EVENT_TABLE_NAME
     };
 
-    m_orchList = { switch_orch, gCrmOrch, gBufferOrch, gPortsOrch, intfs_orch, gNeighOrch, gRouteOrch, copp_orch, tunnel_decap_orch, qos_orch, mirror_orch };
+    m_orchList = { gSwitchOrch, gCrmOrch, gBufferOrch, gPortsOrch, intfs_orch, gNeighOrch, gRouteOrch, copp_orch, tunnel_decap_orch, qos_orch, mirror_orch };
 
     bool initialize_dtel = false;
     if (platform == BFN_PLATFORM_SUBSTRING || platform == VS_PLATFORM_SUBSTRING)
@@ -320,5 +323,69 @@ void OrchDaemon::start()
          * is a good chance to flush the pipeline before next select happened.
          */
         flush();
+
+        /*
+         * Asked to check warm restart readiness.
+         * Not doing this under Select::TIMEOUT condition because of
+         * the existence of finer granularity ExecutableTimer with select
+         */
+        if (gSwitchOrch->checkRestartReady())
+        {
+            bool ret = warmRestartCheckReply();
+            if (ret)
+            {
+                // Orchagent is ready to perform warm restart, stop processing any new db data.
+                // Should sleep here or continue handling timers and etc.??
+                SWSS_LOG_WARN("Orchagent is frozen for warm restart!");
+                sleep(UINT_MAX);
+            }
+        }
+
+    }
+}
+
+/*
+ * Get tasks to sync for consumers of each orch being managed by this orch daemon
+ */
+void OrchDaemon::getTaskToSync(vector<string> &ts)
+{
+    for (Orch *o : m_orchList)
+    {
+        o->dumpToSyncTasks(ts);
     }
 }
+
+/*
+ * Reply with "READY" notification if no pending tasks, and return true.
+ * Ortherwise reply with "NOT_READY" notification and return false.
+ * Further consideration is needed as to when orchagent is treated as warm restart ready.
+ * For now, no pending task should exist in any orch agent.
+ */
+bool OrchDaemon::warmRestartCheckReply()
+{
+    NotificationProducer restartRequestReply(m_applDb, "RESTARTCHECKREPLY");
+    std::vector<swss::FieldValueTuple> values;
+    std::string op = "READY";
+    bool ret = true;
+
+    vector<string> ts;
+    getTaskToSync(ts);
+
+    if (ts.size() != 0)
+    {
+        SWSS_LOG_ERROR("WarmRestart not ready with pending tasks: ");
+        for(auto &s : ts)
+        {
+            SWSS_LOG_NOTICE("%s", s.c_str());
+        }
+        op = "NOT_READY";
+        ret = false;
+    }
+
+    SWSS_LOG_NOTICE("Restart check result: %s", op.c_str());
+
+    restartRequestReply.send(op, op, values);
+    gSwitchOrch->checkRestartReadyDone();
+    return ret;
+}
+

orchagent/orchdaemon.h

@@ -34,6 +34,8 @@ class OrchDaemon
 
     bool init();
     void start();
+    void getTaskToSync(vector<string> &ts);
+    bool warmRestartCheckReply();
 private:
     DBConnector *m_applDb;
     DBConnector *m_configDb;

orchagent/switchorch.cpp

@@ -2,6 +2,7 @@
 
 #include "switchorch.h"
 #include "converter.h"
+#include "notifier.h"
 
 using namespace std;
 using namespace swss;
@@ -27,8 +28,13 @@ const map<string, sai_packet_action_t> packet_action_map =
 };
 
 SwitchOrch::SwitchOrch(DBConnector *db, string tableName) :
-        Orch(db, tableName)
+        Orch(db, tableName),
+        m_db(db)
 {
+    m_restartCheckNotificationConsumer = new NotificationConsumer(db, "RESTARTCHECK");
+    auto restartCheckNotifier = new Notifier(m_restartCheckNotificationConsumer, this);
+    // restartCheckNotifier->setName("RESTARTCHECK");
+    Orch::addExecutor("RESTARTCHECK", restartCheckNotifier);
 }
 
 void SwitchOrch::doTask(Consumer &consumer)
@@ -122,3 +128,25 @@ void SwitchOrch::doTask(Consumer &consumer)
     }
 }
 
+void SwitchOrch::doTask(NotificationConsumer& consumer)
+{
+    SWSS_LOG_ENTER();
+
+    std::string op;
+    std::string data;
+    std::vector<swss::FieldValueTuple> values;
+
+    consumer.pop(op, data, values);
+
+    if (&consumer != m_restartCheckNotificationConsumer)
+    {
+        return;
+    }
+
+    SWSS_LOG_NOTICE("RESTARTCHECK notification for %s ", op.c_str());
+    if (op == "orchagent")
+    {
+        checkRestartReadyState = true;
+    }
+}
+

orchagent/switchorch.h

@@ -6,7 +6,14 @@ class SwitchOrch : public Orch
 {
 public:
     SwitchOrch(DBConnector *db, string tableName);
-
+    bool checkRestartReady() { return checkRestartReadyState; }
+    void checkRestartReadyDone() { checkRestartReadyState = false; }
 private:
     void doTask(Consumer &consumer);
+
+    NotificationConsumer* m_restartCheckNotificationConsumer;
+    void doTask(NotificationConsumer& consumer);
+    // Whether to check readiness of warm restart
+    bool checkRestartReadyState = false;
+    DBConnector *m_db;
 };