Simulate link prober offload

Signed-off-by: Longxiang Lyu <[email protected]>

src/DbInterface.cpp

@@ -33,6 +33,7 @@
 #include "swss/netlink.h"
 #include "swss/macaddress.h"
 #include "swss/select.h"
+#include "swss/json.hpp"
 
 #include "DbInterface.h"
 #include "MuxManager.h"
@@ -42,8 +43,11 @@
 
 namespace mux
 {
+using json = nlohmann::json;
+
 constexpr auto DEFAULT_TIMEOUT_MSEC = 1000;
 std::vector<std::string> DbInterface::mMuxState = {"active", "standby", "unknown", "Error"};
+std::vector<std::string> DbInterface::mLinkProberState = {"Active", "Standby", "Unknown", "Wait", "PeerWait", "PeerActive", "PeerUnknown", "Init", "Up", "Down", "Init", "Up", "Down"};
 std::vector<std::string> DbInterface::mMuxLinkmgrState = {"uninitialized", "unhealthy", "healthy"};
 std::vector<std::string> DbInterface::mMuxMetrics = {"start", "end"};
 std::vector<std::string> DbInterface::mLinkProbeMetrics = {"link_prober_unknown_start", "link_prober_unknown_end", "link_prober_wait_start", "link_prober_active_start", "link_prober_standby_start"};
@@ -112,6 +116,18 @@ void DbInterface::setPeerMuxState(const std::string &portName, mux_state::MuxSta
     ));
 }
 
+void DbInterface::setLinkProberSessionState(const std::string portName, const std::string &sessionId, link_prober::LinkProberState::Label label)
+{
+    MUXLOGDEBUG(boost::format("%s: setting link prober session %s state to %s") % portName % sessionId % mLinkProberState[label]);
+
+    boost::asio::post(mStrand, boost::bind(
+        &DbInterface::handleSetLinkProberSessionState,
+        this,
+        sessionId,
+        label
+    ));
+}
+
 
 //
 // ---> probeMuxState(const std::string &portName)
@@ -308,6 +324,7 @@ void DbInterface::initialize()
     try {
         mAppDbPtr = std::make_shared<swss::DBConnector> ("APPL_DB", 0);
         mStateDbPtr = std::make_shared<swss::DBConnector> ("STATE_DB", 0);
+        mAsicDbPtr = std::make_shared<swss::DBConnector>("ASIC_DB", 0);
 
         mAppDbMuxTablePtr = std::make_shared<swss::ProducerStateTable> (
             mAppDbPtr.get(), APP_MUX_CABLE_TABLE_NAME
@@ -334,6 +351,9 @@ void DbInterface::initialize()
             mStateDbPtr.get(),  STATE_MUX_SWITCH_CAUSE_TABLE_NAME
         );
         mMuxStateTablePtr = std::make_shared<swss::Table> (mStateDbPtr.get(), STATE_MUX_CABLE_TABLE_NAME);
+        mAsicDbNotificationChannelPtr = std::make_shared<swss::NotificationProducer> (
+            mAsicDbPtr.get(), LINK_PROBER_SESSION_STATE_CHANGE_NOTIFICATION_CHANNEL
+        );
 
         mSwssThreadPtr = std::make_shared<boost::thread> (&DbInterface::handleSwssNotification, this);
     }
@@ -420,6 +440,30 @@ void DbInterface::handleSetPeerMuxState(const std::string portName, mux_state::M
     }
 }
 
+//
+// ---> handleSetLinkProberSessionState(const std::string &sessionId, link_prober::LinkProberState::Label label);
+//
+// set link prober session state in ASIC DB to simulate offload
+//
+void DbInterface::handleSetLinkProberSessionState(const std::string &sessionId, link_prober::LinkProberState::Label label)
+{
+    std::string notification;
+    std::vector<swss::FieldValueTuple> entries;
+
+    json j = json::array();
+    json item;
+    item["link_prober_session_id"] = sessionId;
+    item["session_state"] = mLinkProberState[label];
+    j.push_back(item);
+    notification = j.dump();
+
+    mAsicDbNotificationChannelPtr->send(
+        LINK_PROBER_SESSION_STATE_CHANGE_NOTIFICATION,
+        notification,
+        entries
+    );
+}
+
 //
 // ---> handleProbeMuxState(const std::string portName)
 //
@@ -1350,6 +1394,41 @@ void DbInterface::handleMuxStateNotifiction(swss::SubscriberStateTable &statedbP
     processMuxStateNotifiction(entries);
 }
 
+//
+// ---> handleLinkProberSessionStateNotification(std::deque<swss::KeyOpFieldsValuesTuple> &entries);
+//
+// handle link prober session state notification
+//
+void DbInterface::handleLinkProberSessionStateNotification(swss::NotificationConsumer &linkProberNotificationChannel)
+{
+    std::deque<swss::KeyOpFieldsValuesTuple> entries;
+
+    linkProberNotificationChannel.pops(entries);
+    processPeerMuxNotification(entries);
+}
+
+//
+// ---> processLinkProberSessionStateNotification(std::deque<swss::KeyOpFieldsValuesTuple> &entries);
+//
+// process link prober session state notification
+//
+void DbInterface::processLinkProberSessionStateNotification(std::deque<swss::KeyOpFieldsValuesTuple> &entries)
+{
+    for (auto &entry : entries) {
+        std::string key = kfvKey(entry);
+        std::string op = kfvOp(entry);
+
+        json j = json::parse(op);
+        for (uint32_t i = 0; i < j.size(); ++i) {
+            mMuxManagerPtr->handleLinkProberSessionStateNotification(
+                j[i]["link_prober_session_id"],
+                j[i]["session_state"]
+            );
+        }
+    }
+}
+
+
 //
 // ---> processDefaultRouteStateNotification(std::deque<swss::KeyOpFieldsValuesTuple> &entries)
 // 
@@ -1460,6 +1539,7 @@ void DbInterface::handleSwssNotification()
     std::shared_ptr<swss::DBConnector> configDbPtr = std::make_shared<swss::DBConnector> ("CONFIG_DB", 0);
     std::shared_ptr<swss::DBConnector> appDbPtr = std::make_shared<swss::DBConnector> ("APPL_DB", 0);
     std::shared_ptr<swss::DBConnector> stateDbPtr = std::make_shared<swss::DBConnector> ("STATE_DB", 0);
+    std::shared_ptr<swss::DBConnector> asicDbPtr = std::make_shared<swss::DBConnector> ("ASIC_DB", 0);
 
     // For reading Link Prober configurations from the MUX linkmgr table name
     swss::SubscriberStateTable configDbMuxLinkmgrTable(configDbPtr.get(), CFG_MUX_LINKMGR_TABLE_NAME);
@@ -1482,6 +1562,8 @@ void DbInterface::handleSwssNotification()
     swss::SubscriberStateTable stateDbMuxInfoTable(stateDbPtr.get(), MUX_CABLE_INFO_TABLE);
     // for getting peer's admin forwarding state
     swss::SubscriberStateTable stateDbPeerMuxTable(stateDbPtr.get(), STATE_PEER_HW_FORWARDING_STATE_TABLE_NAME);
+    // for getting simulated link prober notifications
+    swss::NotificationConsumer linkProberNotificationChannel(asicDbPtr.get(), LINK_PROBER_SESSION_STATE_CHANGE_NOTIFICATION_CHANNEL);
 
     getTorMacAddress(configDbPtr);
     getVlanNames(configDbPtr);
@@ -1510,6 +1592,9 @@ void DbInterface::handleSwssNotification()
     swssSelect.addSelectable(&stateDbMuxInfoTable);
     swssSelect.addSelectable(&stateDbPeerMuxTable);
     swssSelect.addSelectable(&netlinkNeighbor);
+    if (mMuxManagerPtr->getMuxConfig().getIfEnableSimulateLfdOffload()) {
+        swssSelect.addSelectable(&linkProberNotificationChannel);
+    }
 
     while (mPollSwssNotifcation) {
         swss::Selectable *selectable;
@@ -1545,6 +1630,8 @@ void DbInterface::handleSwssNotification()
             handlePeerLinkStateNotification(stateDbMuxInfoTable);
         } else if (selectable == static_cast<swss::Selectable *> (&stateDbPeerMuxTable)) {
             handlePeerMuxStateNotification(stateDbPeerMuxTable);
+        } else if (selectable == static_cast<swss::Selectable *> (&linkProberNotificationChannel)) {
+            handleLinkProberSessionStateNotification(linkProberNotificationChannel);
         } else if (selectable == static_cast<swss::Selectable *> (&netlinkNeighbor)) {
             continue;
         } else {

src/DbInterface.h

@@ -34,6 +34,8 @@
 #include "swss/dbconnector.h"
 #include "swss/producerstatetable.h"
 #include "swss/subscriberstatetable.h"
+#include "swss/notificationproducer.h"
+#include "swss/notificationconsumer.h"
 #include "swss/warm_restart.h"
 
 #include "link_manager/LinkManagerStateMachineActiveStandby.h"
@@ -55,6 +57,8 @@ namespace mux
 #define STATE_PEER_HW_FORWARDING_STATE_TABLE_NAME   "HW_MUX_CABLE_TABLE_PEER"
 
 #define STATE_MUX_SWITCH_CAUSE_TABLE_NAME "MUX_SWITCH_CAUSE"
+#define LINK_PROBER_SESSION_STATE_CHANGE_NOTIFICATION_CHANNEL "LINK_PROBER_NOTIFICATIONS"
+#define LINK_PROBER_SESSION_STATE_CHANGE_NOTIFICATION "link_prober_session_state_change"
 
 class MuxManager;
 using ServerIpPortMap = std::map<boost::asio::ip::address, std::string>;
@@ -166,6 +170,19 @@ class DbInterface
     */
     void setPeerMuxState(const std::string &portName, mux_state::MuxState::Label label);
 
+    /**
+    *@method setLinkProberSessionState
+    *
+    *@brief set link prober session state in ASIC DB to simulate offload
+    *
+    *@param portName (in)   MUX/port name
+    *@param sessionId (in)  link prober session id
+    *@param label (in)      label of target state
+    *
+    *@return none
+    */
+    void setLinkProberSessionState(const std::string portName, const std::string &sessionId, link_prober::LinkProberState::Label label);
+
     /**
     *@method probeMuxState
     *
@@ -414,6 +431,18 @@ class DbInterface
     */
     virtual void handleSetPeerMuxState(const std::string portName, mux_state::MuxState::Label label);
 
+    /**
+    *@method handleSetLinkProberSessionState
+    *
+    *@brief set link prober session state in ASIC DB to simulate offload
+    *
+    *@param sessionId (in)  link prober session id
+    *@param label (in)      label of target state
+    *
+    *@return none
+    */
+    void handleSetLinkProberSessionState(const std::string &sessionId, link_prober::LinkProberState::Label label);
+
     /**
     *@method handleProbeMuxState
     *
@@ -807,6 +836,26 @@ class DbInterface
     */
     void handleMuxStateNotifiction(swss::SubscriberStateTable &statedbPortTable);
 
+    /**
+    *@method processLinkProberSessionStateNotification
+    *
+    *@brief process link prober session state notification
+    *
+    *@param entries (in) reference to link prober session state notification
+    *
+    *@return none
+    */
+    inline void processLinkProberSessionStateNotification(std::deque<swss::KeyOpFieldsValuesTuple> &entries);
+
+    /**
+    *@method handleLinkProberSessionStateNotification
+    *
+    *@brief handle link prober session state notification
+    *
+    *@return none
+    */
+    void handleLinkProberSessionStateNotification(swss::NotificationConsumer &linkProberNotificationChannel);
+
     /**
     *@method handleSwssNotification
     *
@@ -858,6 +907,7 @@ class DbInterface
 
 private:
     static std::vector<std::string> mMuxState;
+    static std::vector<std::string> mLinkProberState;
     static std::vector<std::string> mMuxLinkmgrState;
     static std::vector<std::string> mMuxMetrics;
     static std::vector<std::string> mLinkProbeMetrics;
@@ -869,6 +919,7 @@ class DbInterface
 
     std::shared_ptr<swss::DBConnector> mAppDbPtr;
     std::shared_ptr<swss::DBConnector> mStateDbPtr;
+    std::shared_ptr<swss::DBConnector> mAsicDbPtr;
     std::shared_ptr<swss::Table> mMuxStateTablePtr;
 
     // for communicating with orchagent
@@ -887,6 +938,8 @@ class DbInterface
     std::shared_ptr<swss::Table> mStateDbLinkProbeStatsTablePtr;
     // for writing mux switch reason to state db 
     std::shared_ptr<swss::Table> mStateDbSwitchCauseTablePtr;
+    // for writing link prober session state change notification
+    std::shared_ptr<swss::NotificationProducer> mAsicDbNotificationChannelPtr;
 
     std::shared_ptr<boost::thread> mSwssThreadPtr;
 

src/LinkMgrdMain.cpp

@@ -71,6 +71,7 @@ int main(int argc, const char* argv[])
     bool measureSwitchover = false;
     bool defaultRoute = false;
     bool linkToSwssLogger = false;
+    bool simulateLfdOffload = false;
 
     program_options::options_description description("linkmgrd options");
     description.add_options()
@@ -95,6 +96,10 @@ int main(int argc, const char* argv[])
          "Link to swss logger instead of using native boost syslog support, this will"
          "set the boost logging level to TRACE and option verbosity is ignored"
          )
+        ("simulate_lfd_offload,s",
+         program_options::bool_switch(&simulateLfdOffload)->default_value(false),
+         "Simulate LFD offload by posting link prober state change notification to Redis"
+         )
     ;
 
     //
@@ -139,7 +144,7 @@ int main(int argc, const char* argv[])
         }
 
         std::shared_ptr<mux::MuxManager> muxManagerPtr = std::make_shared<mux::MuxManager> ();
-        muxManagerPtr->initialize(measureSwitchover, defaultRoute);
+        muxManagerPtr->initialize(measureSwitchover, defaultRoute, simulateLfdOffload);
         muxManagerPtr->run();
         muxManagerPtr->deinitialize();
     }

src/MuxManager.cpp

@@ -76,7 +76,7 @@ void MuxManager::setUseWellKnownMacActiveActive(bool useWellKnownMac)
 //
 // initialize MuxManager class and creates DbInterface instance that reads/listen from/to Redis db
 //
-void MuxManager::initialize(bool enable_feature_measurement, bool enable_feature_default_route)
+void MuxManager::initialize(bool enable_feature_measurement, bool enable_feature_default_route, bool enable_simulate_lfd_offload)
 {
     for (uint8_t i = 0; (mMuxConfig.getNumberOfThreads() > 2) &&
                         (i < mMuxConfig.getNumberOfThreads() - 2); i++) {
@@ -94,6 +94,7 @@ void MuxManager::initialize(bool enable_feature_measurement, bool enable_feature
 
     mMuxConfig.enableSwitchoverMeasurement(enable_feature_measurement);
     mMuxConfig.enableDefaultRouteFeature(enable_feature_default_route);
+    mMuxConfig.enableSimulateLfdOffload(enable_simulate_lfd_offload);
 }
 
 //
@@ -445,6 +446,9 @@ std::shared_ptr<MuxPort> MuxManager::getMuxPortPtrOrThrow(const std::string &por
                     muxPortPtr->setServerMacAddress(address);
                 }
             }
+            if (mMuxConfig.getIfEnableSimulateLfdOffload()) {
+                // register the dispatch table
+            }
             mPortMap.insert({portName, muxPortPtr});
         }
         else {
@@ -591,4 +595,36 @@ void MuxManager::handleTsaEnableNotification(bool enable)
     }
 }
 
+//
+// ---> getMuxPortPtr(const std::string &sessionId)
+//
+// get the mux port that the link prober session belongs to
+//
+inline std::shared_ptr<MuxPort> MuxManager::getMuxPortPtr(const std::string &sessionId)
+{
+    auto pos = sessionId.find('|');
+    if (pos != std::string::npos) {
+        std::string portName = sessionId.substr(0, pos);
+
+        PortMapIterator portMapIterator = mPortMap.find(portName);
+        if (portMapIterator != mPortMap.end()) {
+            return portMapIterator->second;
+        }
+    }
+    return nullptr;
+}
+
+//
+// ---> handleLinkProberSessionStateNotification(const std::string &sessionId, const std::string &sessionState)
+//
+// handle the link prober session state notification from ASIC DB
+//
+void MuxManager::handleLinkProberSessionStateNotification(const std::string &sessionId, const std::string &sessionState)
+{
+    auto muxPortPtr = getMuxPortPtr(sessionId);
+    if (muxPortPtr) {
+        muxPortPtr->handleLinkProberSessionStateNotification(sessionId, sessionState);
+    }
+}
+
 } /* namespace mux */