Fix breakage for smartswitch usecase

Refer to sonic-net/sonic-buildimage#19638
The previous PR (#3979)
inadvertantly caused the above issue.
This commit fixes that by connecting to zmq from fpmsyncd using the
vrf as well (if one is configured)

Author: venkit-nexthop

fpmsyncd/fpmsyncd.cpp

@@ -69,6 +69,31 @@ static bool eoiuFlagsSet(Table &bgpStateTable)
     return true;
 }
 
+std::string get_zmq_vrf(DBConnector &cfgDb) {
+    std::shared_ptr<std::string> mgmt_vrf_enabled = nullptr;
+
+    try
+    {
+        swss::DBConnector config_db("CONFIG_DB", 0);
+        mgmt_vrf_enabled = config_db.hget("MGMT_VRF_CONFIG|vrf_global",
+                                          "mgmtVrfEnabled");
+    }
+    catch (const std::runtime_error &e)
+    {
+        return std::string();
+    }
+
+    if (!mgmt_vrf_enabled)
+    {
+        return std::string();
+    }
+
+    if( *mgmt_vrf_enabled != "true") {
+        return std::string();
+    }
+    return std::string("mgmt");
+}
+
 int main(int argc, char **argv)
 {
     swss::Logger::linkToDbNative("fpmsyncd");
@@ -83,7 +108,8 @@ int main(int argc, char **argv)
     std::unique_ptr<NotificationConsumer> routeResponseChannel;
 
     RedisPipeline pipeline(&db, ROUTE_SYNC_PPL_SIZE);
-    RouteSync sync(&pipeline);
+    auto zmqVrf = get_zmq_vrf(cfgDb);
+    RouteSync sync(&pipeline, zmqVrf);
 
     DBConnector stateDb("STATE_DB", 0);
     Table bgpStateTable(&stateDb, STATE_BGP_TABLE_NAME);

fpmsyncd/routesync.cpp

@@ -145,9 +145,9 @@ static decltype(auto) makeNlAddr(const T& ip)
 }
 
 
-RouteSync::RouteSync(RedisPipeline *pipeline) :
+RouteSync::RouteSync(RedisPipeline *pipeline, const std::string& zmqVrf) :
     // When the feature ORCH_NORTHBOND_ROUTE_ZMQ_ENABLED is enabled, route events must be sent to orchagent via the ZMQ channel.
-    m_zmqClient(create_local_zmq_client(ORCH_NORTHBOND_ROUTE_ZMQ_ENABLED, false)),
+    m_zmqClient(create_local_zmq_client(ORCH_NORTHBOND_ROUTE_ZMQ_ENABLED, false, zmqVrf)),
     m_routeTable(createProducerStateTable(pipeline, APP_ROUTE_TABLE_NAME, true, m_zmqClient)),
     m_nexthop_groupTable(pipeline, APP_NEXTHOP_GROUP_TABLE_NAME, true),
     m_label_routeTable(createProducerStateTable(pipeline, APP_LABEL_ROUTE_TABLE_NAME, true, m_zmqClient)),

fpmsyncd/routesync.h

@@ -178,7 +178,7 @@ class RouteSync : public NetMsg
 public:
     enum { MAX_ADDR_SIZE = 64 };
 
-    RouteSync(RedisPipeline *pipeline);
+    RouteSync(RedisPipeline *pipeline, const std::string& zmqVrf = "");
 
     virtual void onMsg(int nlmsg_type, struct nl_object *obj);
 

lib/orch_zmq_config.cpp

@@ -103,12 +103,12 @@ bool swss::get_feature_status(std::string feature, bool default_value)
     return *enabled == "true";
 }
 
-std::shared_ptr<swss::ZmqClient> swss::create_local_zmq_client(std::string feature, bool default_value)
+std::shared_ptr<swss::ZmqClient> swss::create_local_zmq_client(std::string feature, bool default_value, std::string vrf)
 {
     auto enable = get_feature_status(feature, default_value);
     if (enable) {
-        SWSS_LOG_NOTICE("Feature %s enabled, Create ZMQ client : %s", feature.c_str(), ZMQ_LOCAL_ADDRESS);
-        return create_zmq_client(ZMQ_LOCAL_ADDRESS);
+        SWSS_LOG_NOTICE("Feature %s enabled, Create ZMQ client : %s, vrf: %s", feature.c_str(), ZMQ_LOCAL_ADDRESS, vrf.c_str());
+        return create_zmq_client(ZMQ_LOCAL_ADDRESS, vrf);
     }
 
     return nullptr;

lib/orch_zmq_config.h

@@ -37,7 +37,7 @@ std::shared_ptr<ZmqServer> create_zmq_server(std::string zmq_address, std::strin
 
 bool get_feature_status(std::string feature, bool default_value);
 
-std::shared_ptr<swss::ZmqClient> create_local_zmq_client(std::string feature, bool default_value);
+std::shared_ptr<swss::ZmqClient> create_local_zmq_client(std::string feature, bool default_value, std::string vrf="");
 
 std::shared_ptr<swss::ProducerStateTable> createProducerStateTable(DBConnector *db, const std::string &tableName, std::shared_ptr<swss::ZmqClient> zmqClient);
 

orchagent/main.cpp

@@ -80,7 +80,7 @@ uint32_t create_switch_timeout = 0;
 
 void usage()
 {
-    cout << "usage: orchagent [-h] [-r record_type] [-d record_location] [-f swss_rec_filename] [-j sairedis_rec_filename] [-b batch_size] [-m MAC] [-i INST_ID] [-s] [-z mode] [-k bulk_size] [-q zmq_server_address] [-c mode] [-t create_switch_timeout] [-v VRF] [-I heart_beat_interval] [-R] [-M]" << endl;
+    cout << "usage: orchagent [-h] [-r record_type] [-d record_location] [-f swss_rec_filename] [-j sairedis_rec_filename] [-b batch_size] [-m MAC] [-i INST_ID] [-s] [-z mode] [-k bulk_size] [-q zmq_server_address] [-c mode] [-t create_switch_timeout] [-v ZMQVRF] [-I heart_beat_interval] [-R] [-M]" << endl;
     cout << "    -h: display this message" << endl;
     cout << "    -r record_type: record orchagent logs with type (default 3)" << endl;
     cout << "                    Bit 0: sairedis.rec, Bit 1: swss.rec, Bit 2: responsepublisher.rec. For example:" << endl;
@@ -101,7 +101,7 @@ void usage()
     cout << "    -q zmq_server_address: ZMQ server address (default disable ZMQ)" << endl;
     cout << "    -c counter mode (traditional|asic_db), default: asic_db" << endl;
     cout << "    -t Override create switch timeout, in sec" << endl;
-    cout << "    -v vrf: VRF name (default empty)" << endl;
+    cout << "    -v zmqvrf: ZMQVRF name (default empty)" << endl;
     cout << "    -I heart_beat_interval: Heart beat interval in millisecond (default 10)" << endl;
     cout << "    -R enable the ring thread feature" << endl;
     cout << "    -M enable SAI MACSec POST" << endl;
@@ -367,7 +367,7 @@ int main(int argc, char **argv)
     string swss_rec_filename = Recorder::SWSS_FNAME;
     string sairedis_rec_filename = Recorder::SAIREDIS_FNAME;
     string zmq_server_address = "";
-    string vrf;
+    string zmq_vrf;
     string responsepublisher_rec_filename = Recorder::RESPPUB_FNAME;
     int record_type = 3; // Only swss and sairedis recordings enabled by default.
     long heartBeatInterval = HEART_BEAT_INTERVAL_MSECS_DEFAULT;
@@ -471,7 +471,7 @@ int main(int argc, char **argv)
         case 'v':
             if (optarg)
             {
-                vrf = optarg;
+                zmq_vrf = optarg;
             }
             break;
         case 'I':
@@ -545,8 +545,8 @@ int main(int argc, char **argv)
     }
     else
     {
-        SWSS_LOG_NOTICE("The ZMQ channel on the northbound side of orchagent has been initialized: %s, %s", zmq_server_address.c_str(), vrf.c_str());
-        zmq_server = create_zmq_server(zmq_server_address);
+        SWSS_LOG_NOTICE("The ZMQ channel on the northbound side of orchagent has been initialized: %s, %s", zmq_server_address.c_str(), zmq_vrf.c_str());
+        zmq_server = create_zmq_server(zmq_server_address, zmq_vrf);
     }
 
     // Get switch_type