[fpmsyncd] Implement pending route suppression feature

fpmsyncd/fpminterface.h

@@ -0,0 +1,27 @@
+#pragma once
+
+#include <swss/selectable.h>
+#include <libnl3/netlink/netlink.h>
+
+#include "fpm/fpm.h"
+
+namespace swss
+{
+
+/**
+ * @brief FPM zebra communication interface
+ */
+class FpmInterface : public Selectable
+{
+public:
+    virtual ~FpmInterface() = default;
+
+    /**
+     * @brief Send netlink message through FPM socket
+     * @param msg Netlink message
+     * @return True on success, otherwise false is returned
+     */
+    virtual bool send(nlmsghdr* nl_hdr) = 0;
+};
+
+}

fpmsyncd/fpmlink.cpp

@@ -158,11 +158,17 @@ FpmLink::FpmLink(RouteSync *rsync, unsigned short port) :
 
     m_server_up = true;
     m_messageBuffer = new char[m_bufSize];
+    m_sendBuffer = new char[m_bufSize];
+
+    m_routesync->onFpmConnected(*this);
 }
 
 FpmLink::~FpmLink()
 {
+    m_routesync->onFpmDisconnected();
+
     delete[] m_messageBuffer;
+    delete[] m_sendBuffer;
     if (m_connected)
         close(m_connection_socket);
     if (m_server_up)
@@ -277,3 +283,36 @@ void FpmLink::processFpmMessage(fpm_msg_hdr_t* hdr)
         nlmsg_free(msg);
     }
 }
+
+bool FpmLink::send(nlmsghdr* nl_hdr)
+{
+    fpm_msg_hdr_t hdr{};
+
+    size_t len = fpm_msg_align(sizeof(hdr) + nl_hdr->nlmsg_len);
+
+    if (len > m_bufSize)
+    {
+        SWSS_LOG_THROW("Message length %zu is greater then the send buffer size %d", len, m_bufSize);
+    }
+
+    hdr.version = FPM_PROTO_VERSION;
+    hdr.msg_type = FPM_MSG_TYPE_NETLINK;
+    hdr.msg_len = htons(static_cast<uint16_t>(len));
+
+    memcpy(m_sendBuffer, &hdr, sizeof(hdr));
+    memcpy(m_sendBuffer + sizeof(hdr), nl_hdr, nl_hdr->nlmsg_len);
+
+    size_t sent = 0;
+    while (sent != len)
+    {
+        auto rc = ::send(m_connection_socket, m_sendBuffer + sent, len - sent, 0);
+        if (rc == -1)
+        {
+            SWSS_LOG_ERROR("Failed to send FPM message: %s", strerror(errno));
+            return false;
+        }
+        sent += rc;
+    }
+
+    return true;
+}

fpmsyncd/fpmlink.h

@@ -11,13 +11,13 @@
 #include <unistd.h>
 #include <exception>
 
-#include "selectable.h"
 #include "fpm/fpm.h"
+#include "fpmsyncd/fpminterface.h"
 #include "fpmsyncd/routesync.h"
 
 namespace swss {
 
-class FpmLink : public Selectable {
+class FpmLink : public FpmInterface {
 public:
     const int MSG_BATCH_SIZE;
     FpmLink(RouteSync *rsync, unsigned short port = FPM_DEFAULT_PORT);
@@ -41,10 +41,13 @@ class FpmLink : public Selectable {
 
     void processFpmMessage(fpm_msg_hdr_t* hdr);
 
+    bool send(nlmsghdr* nl_hdr) override;
+
 private:
     RouteSync *m_routesync;
     unsigned int m_bufSize;
     char *m_messageBuffer;
+    char *m_sendBuffer;
     unsigned int m_pos;
 
     bool m_connected;

fpmsyncd/fpmsyncd.cpp

@@ -4,10 +4,14 @@
 #include "select.h"
 #include "selectabletimer.h"
 #include "netdispatcher.h"
+#include "netlink.h"
+#include "notificationconsumer.h"
+#include "subscriberstatetable.h"
 #include "warmRestartHelper.h"
 #include "fpmsyncd/fpmlink.h"
 #include "fpmsyncd/routesync.h"
 
+#include <netlink/route/route.h>
 
 using namespace std;
 using namespace swss;
@@ -47,21 +51,47 @@ static bool eoiuFlagsSet(Table &bgpStateTable)
 int main(int argc, char **argv)
 {
     swss::Logger::linkToDbNative("fpmsyncd");
+
+    const auto routeResponseChannelName = std::string("APPL_DB_") + APP_ROUTE_TABLE_NAME + "_RESPONSE_CHANNEL";
+
     DBConnector db("APPL_DB", 0);
+    DBConnector cfgDb("CONFIG_DB", 0);
+    SubscriberStateTable deviceMetadataTableSubscriber(&cfgDb, CFG_DEVICE_METADATA_TABLE_NAME);
+    Table deviceMetadataTable(&cfgDb, CFG_DEVICE_METADATA_TABLE_NAME);
+    DBConnector applStateDb("APPL_STATE_DB", 0);
+    std::unique_ptr<NotificationConsumer> routeResponseChannel;
+
     RedisPipeline pipeline(&db);
     RouteSync sync(&pipeline);
 
     DBConnector stateDb("STATE_DB", 0);
     Table bgpStateTable(&stateDb, STATE_BGP_TABLE_NAME);
 
+    NetLink netlink;
+
+    netlink.registerGroup(RTNLGRP_LINK);
+
     NetDispatcher::getInstance().registerMessageHandler(RTM_NEWROUTE, &sync);
     NetDispatcher::getInstance().registerMessageHandler(RTM_DELROUTE, &sync);
+    NetDispatcher::getInstance().registerMessageHandler(RTM_NEWLINK, &sync);
+    NetDispatcher::getInstance().registerMessageHandler(RTM_DELLINK, &sync);
+
+    rtnl_route_read_protocol_names(DefaultRtProtoPath);
+
+    std::string suppressionEnabledStr;
+    deviceMetadataTable.hget("localhost", "suppress-fib-pending", suppressionEnabledStr);
+    if (suppressionEnabledStr == "enabled")
+    {
+        routeResponseChannel = std::make_unique<NotificationConsumer>(&applStateDb, routeResponseChannelName);
+        sync.setSuppressionEnabled(true);
+    }
 
     while (true)
     {
         try
         {
             FpmLink fpm(&sync);
+
             Select s;
             SelectableTimer warmStartTimer(timespec{0, 0});
             // Before eoiu flags detected, check them periodically. It also stop upon detection of reconciliation done.
@@ -80,6 +110,13 @@ int main(int argc, char **argv)
             cout << "Connected!" << endl;
 
             s.addSelectable(&fpm);
+            s.addSelectable(&netlink);
+            s.addSelectable(&deviceMetadataTableSubscriber);
+
+            if (sync.isSuppressionEnabled())
+            {
+                s.addSelectable(routeResponseChannel.get());
+            }
 
             /* If warm-restart feature is enabled, execute 'restoration' logic */
             bool warmStartEnabled = sync.m_warmStartHelper.checkAndStart();
@@ -139,11 +176,8 @@ int main(int argc, char **argv)
                         SWSS_LOG_NOTICE("Warm-Restart EOIU hold timer expired.");
                     }
 
-                    if (sync.m_warmStartHelper.inProgress())
-                    {
-                        sync.m_warmStartHelper.reconcile();
-                        SWSS_LOG_NOTICE("Warm-Restart reconciliation processed.");
-                    }
+                    sync.onWarmStartEnd(applStateDb);
+
                     // remove the one-shot timer.
                     s.removeSelectable(temps);
                     pipeline.flush();
@@ -182,6 +216,67 @@ int main(int argc, char **argv)
                         s.removeSelectable(&eoiuCheckTimer);
                     }
                 }
+                else if (temps == &deviceMetadataTableSubscriber)
+                {
+                    std::deque<KeyOpFieldsValuesTuple> keyOpFvsQueue;
+                    deviceMetadataTableSubscriber.pops(keyOpFvsQueue);
+
+                    for (const auto& keyOpFvs: keyOpFvsQueue)
+                    {
+                        const auto& key = kfvKey(keyOpFvs);
+                        const auto& op = kfvOp(keyOpFvs);
+                        const auto& fvs = kfvFieldsValues(keyOpFvs);
+
+                        if (op != SET_COMMAND)
+                        {
+                            continue;
+                        }
+
+                        if (key != "localhost")
+                        {
+                            continue;
+                        }
+
+                        for (const auto& fv: fvs)
+                        {
+                            const auto& field = fvField(fv);
+                            const auto& value = fvValue(fv);
+
+                            if (field != "suppress-fib-pending")
+                            {
+                                continue;
+                            }
+
+                            bool shouldEnable = (value == "enabled");
+
+                            if (shouldEnable && !sync.isSuppressionEnabled())
+                            {
+                                routeResponseChannel = std::make_unique<NotificationConsumer>(&applStateDb, routeResponseChannelName);
+                                sync.setSuppressionEnabled(true);
+                                s.addSelectable(routeResponseChannel.get());
+                            }
+                            else if (!shouldEnable && sync.isSuppressionEnabled())
+                            {
+                                sync.setSuppressionEnabled(false);
+                                s.removeSelectable(routeResponseChannel.get());
+                                routeResponseChannel.reset();
+                            }
+                        } // end for fvs
+                    } // end for keyOpFvsQueue
+                }
+                else if (routeResponseChannel && (temps == routeResponseChannel.get()))
+                {
+                    std::deque<KeyOpFieldsValuesTuple> notifications;
+                    routeResponseChannel->pops(notifications);
+
+                    for (const auto& notification: notifications)
+                    {
+                        const auto& key = kfvKey(notification);
+                        const auto& fieldValues = kfvFieldsValues(notification);
+
+                        sync.onRouteResponse(key, fieldValues);
+                    }
+                }
                 else if (!warmStartEnabled || sync.m_warmStartHelper.isReconciled())
                 {
                     pipeline.flush();