Add Redis population support class

Server/CMakeLists.txt

@@ -33,14 +33,23 @@ find_library(LIBYAML_CPP_LIBRARY
         lib64
         lib)
 
-
 find_path(LIBSWSSCOMMON_INCLUDE_DIR
         sonic-swss-common/common/dbconnector.h
         HINTS
         ${HINT_ROOT_DIR}
         PATH_SUFFIXES
         include)
 
+
+find_library(LIBHIREDIS_LIBRARY
+        NAMES
+        libhiredis.a hiredis
+        HINTS
+        ${HINT_ROOT_DIR}
+        PATH_SUFFIXES
+        lib64
+        lib)
+
 find_library(LIBSWSSCOMMON_LIBRARY
         NAMES
         libswsscommon.a libswsscommon
@@ -60,13 +69,6 @@ find_library(LIBRT_LIBRARY
         lib)
 
 
-if (NOT LIBSWSSCOMMON_INCLUDE_DIR OR NOT LIBSWSSCOMMON_LIBRARY)
-	Message (FATAL_ERROR "swsscommon was not found, cannot proceed.  Visit https://github.com/sonic-net/sonic-swss-common for details on how to install it.")
-#else ()
-#	Message ("lib = " ${LIBSWSSCOMMON_LIBRARY})
-endif()
-
-
 if (NOT LIBYAML_CPP_INCLUDE_DIR OR NOT LIBYAML_CPP_LIBRARY)
     Message (FATAL_ERROR "Libyaml-cpp was not found, cannot proceed.  Visit https://github.com/jbeder/yaml-cpp for install details.")
 endif()
@@ -75,11 +77,21 @@ if (NOT LIBRT_LIBRARY AND NOT MACOSX)
     Message (FATAL_ERROR "librt was not found, cannot proceed.")
 endif()
 
+if (NOT LIBHIREDIS_LIBRARY)
+	Message (FATAL_ERROR "Libhiredis was not found, cannot proceed.  Visit https://github.com/redis/hiredis for details on how to install it.")
+#else ()
+#	Message ("lib = " ${LIBHIREDIS_LIBRARY})
+endif()
 
+if (NOT LIBSWSSCOMMON_INCLUDE_DIR OR NOT LIBSWSSCOMMON_LIBRARY)
+	Message (FATAL_ERROR "swsscommon was not found, cannot proceed.  Visit https://github.com/sonic-net/sonic-swss-common for details on how to install it.")
+#else ()
+#	Message ("lib = " ${LIBSWSSCOMMON_LIBRARY})
+endif()
 
 # Update the include dir
-include_directories(${LIBSWSSCOMMON_INCLUDE_DIR} ${LIBYAML_CPP_INCLUDE_DIR} src/ src/bmp src/bgp src/bgp/linkstate )
-
+include_directories(${LIBSWSSCOMMON_INCLUDE_DIR} ${LIBYAML_CPP_INCLUDE_DIR} src/ src/bmp src/bgp src/bgp/linkstate)
+#link_directories(${LIBRDKAFKA_LIBRARY})
 
 # Define the source files to compile
 set (SRC_FILES
@@ -90,6 +102,7 @@ set (SRC_FILES
 	src/md5.cpp
 	src/Logger.cpp
     src/Config.cpp
+    src/RedisManager.cpp
 	src/client_thread.cpp
 	src/bgp/parseBGP.cpp
 	src/bgp/NotificationMsg.cpp
@@ -102,6 +115,7 @@ set (SRC_FILES
     src/bgp/EVPN.cpp
     src/bgp/linkstate/MPLinkState.cpp
     src/bgp/linkstate/MPLinkStateAttr.cpp
+    src/redis/MsgBusImpl_redis.cpp
     )
 
 # Disable warnings
@@ -124,7 +138,7 @@ if ("${CMAKE_CXX_COMPILER_ID}" STREQUAL "Clang" OR CMAKE_COMPILER_IS_GNUCXX)
 endif()
 
 # Set the libs to link
-set (LIBS pthread ${LIBYAML_CPP_LIBRARY} ${LIBSWSSCOMMON_LIBRARY} z ${SSL_LIBS} dl)
+set (LIBS pthread ${LIBYAML_CPP_LIBRARY} ${LIBHIREDIS_LIBRARY} ${LIBSWSSCOMMON_LIBRARY} z ${SSL_LIBS} dl)
 
 # Set the binary
 add_executable (openbmpd ${SRC_FILES})
@@ -138,4 +152,4 @@ endif()
 
 # Install the binary and configs
 install(TARGETS openbmpd DESTINATION bin COMPONENT binaries)
-install(FILES openbmpd.conf DESTINATION etc/openbmp/ COMPONENT config)
+install(FILES openbmpd.conf DESTINATION etc/openbmp/ COMPONENT config)

Server/src/Config.cpp

@@ -57,6 +57,25 @@ Config::Config() {
     pat_enabled		= false;
     bzero(admin_id, sizeof(admin_id));
 
+    /*
+     * Initialized the kafka topic names
+     *      The keys match the configuration node/vars. Topic name nodes will be ignored if
+     *      not initialized here.
+     */
+    /*
+    topic_names_map[MSGBUS_TOPIC_VAR_COLLECTOR]        = MSGBUS_TOPIC_COLLECTOR;
+    topic_names_map[MSGBUS_TOPIC_VAR_ROUTER]           = MSGBUS_TOPIC_ROUTER;
+    topic_names_map[MSGBUS_TOPIC_VAR_PEER]             = MSGBUS_TOPIC_PEER;
+    topic_names_map[MSGBUS_TOPIC_VAR_BMP_STAT]         = MSGBUS_TOPIC_BMP_STAT;
+    topic_names_map[MSGBUS_TOPIC_VAR_BMP_RAW]          = MSGBUS_TOPIC_BMP_RAW;
+    topic_names_map[MSGBUS_TOPIC_VAR_BASE_ATTRIBUTE]   = MSGBUS_TOPIC_BASE_ATTRIBUTE;
+    topic_names_map[MSGBUS_TOPIC_VAR_UNICAST_PREFIX]   = MSGBUS_TOPIC_UNICAST_PREFIX;
+    topic_names_map[MSGBUS_TOPIC_VAR_LS_NODE]          = MSGBUS_TOPIC_LS_NODE;
+    topic_names_map[MSGBUS_TOPIC_VAR_LS_LINK]          = MSGBUS_TOPIC_LS_LINK;
+    topic_names_map[MSGBUS_TOPIC_VAR_LS_PREFIX]        = MSGBUS_TOPIC_LS_PREFIX;
+    topic_names_map[MSGBUS_TOPIC_VAR_L3VPN]            = MSGBUS_TOPIC_L3VPN;
+    topic_names_map[MSGBUS_TOPIC_VAR_EVPN]             = MSGBUS_TOPIC_EVPN;
+    */
 }
 
 /*********************************************************************//**

Server/src/RedisManager.cpp

@@ -0,0 +1,236 @@
+/*
+ * Copyright (c) 2024 Microsoft, Inc. and others.  All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ *
+ */
+
+#include "RedisManager.h"
+
+
+/*********************************************************************//**
+ * Constructor for class
+ ***********************************************************************/
+RedisManager::RedisManager() : stateDb_(BMP_DB_NAME, 0, true) {
+    swss::SonicDBConfig::initialize();
+    swss::SonicDBConfig::initializeGlobalConfig();
+    separator_ = swss::SonicDBConfig::getSeparator(BMP_DB_NAME);
+    exit_ = false;
+}
+
+/*********************************************************************//**
+ * Constructor for class
+ ***********************************************************************/
+RedisManager::~RedisManager() {
+    if (!exit_) {
+        exit_ = true;
+        for (auto& threadPtr : threadList_) {
+            threadPtr->join();
+        }
+    }
+}
+
+
+/*********************************************************************
+ * Setup logger for this class
+ *
+ * \param [in] logPtr     logger pointer
+ ***********************************************************************/
+void RedisManager::Setup(Logger *logPtr, BMPListener::ClientInfo *client) {
+    logger = logPtr;
+    client_ = client;
+}
+
+
+
+/**
+ * Get Key separator for deletion
+ *
+ * \param [in] N/A
+ */
+std::string RedisManager::GetKeySeparator() {
+    return separator_;
+}
+
+
+/**
+ * WriteBMPTable
+ *
+ * \param [in] table            Reference to table name
+ * \param [in] key              Reference to various keys list
+ * \param [in] fieldValues      Reference to field-value pairs
+ */
+bool RedisManager::WriteBMPTable(const std::string& table, const std::vector<std::string>& keys, const std::vector<swss::FieldValueTuple> fieldValues) {
+
+    if (enabledTables_.find(table) == enabledTables_.end()) {
+        LOG_INFO("RedisManager %s is disabled", table.c_str());
+        return false;
+    }
+
+    swss::Table stateBMPTable(&stateDb_, table);
+    std::string fullKey;
+    for (const auto& key : keys) {
+        fullKey += key;
+        fullKey += separator_;
+    }
+    fullKey.erase(fullKey.size() - 1);
+
+    LOG_INFO("RedisManager WriteBMPTable key = %s", fullKey.c_str());
+
+    stateBMPTable.set(fullKey, fieldValues);
+    return true;
+}
+
+
+/**
+ * RemoveBMPTable
+ *
+ * \param [in] keys             Reference to various keys
+ */
+bool RedisManager::RemoveBMPTable(const std::vector<std::string>& keys) {
+
+    stateDb_.del(keys);
+    return true;
+}
+
+/**
+ * DisconnectBMP
+ *
+ * \param [in] N/A
+ */
+void RedisManager::DisconnectBMP() {
+    LOG_INFO("RedisManager DisconnectBMP");
+    close(client_->c_sock);
+    client_->c_sock = 0;
+}
+
+/**
+ * ExitRedisManager
+ *
+ * \param [in] N/A
+ */
+void RedisManager::ExitRedisManager() {
+    exit_ = true;
+    for (auto& threadPtr : threadList_) {
+        threadPtr->join();
+    }
+}
+
+/**
+ * ReadBMPTable, there will be dedicated thread be launched inside and monitor corresponding redis table.
+ *
+ * \param [in] tables             table names to be subscribed.
+ */
+void RedisManager::SubscriberWorker(const std::string& table) {
+    try {
+        swss::DBConnector cfgDb("CONFIG_DB", 0, true);
+
+        swss::SubscriberStateTable conf_table(&cfgDb, table);
+        swss::Select s;
+        s.addSelectable(&conf_table);
+
+        while (!exit_) {
+            swss::Selectable *sel;
+            int ret;
+
+            ret = s.select(&sel, BMP_CFG_TABLE_SELECT_TIMEOUT);
+            if (ret == swss::Select::ERROR) {
+                SWSS_LOG_NOTICE("Error: %s!", strerror(errno));
+                continue;
+            }
+            if (ret == swss::Select::TIMEOUT) {
+                continue;
+            }
+
+            swss::KeyOpFieldsValuesTuple kco;
+            conf_table.pop(kco);
+
+            if (std::get<0>(kco) == "SET") {
+                if (std::get<1>(kco) == "true") {
+                    EnableTable(table);
+                }
+                else {
+                    DisableTable(table);
+                    DisconnectBMP();
+                }
+            }
+            else if (std::get<0>(kco) == "DEL")
+            {
+                LOG_ERR("Config should not be deleted");
+            }
+        }
+    }
+    catch (const exception &e) {
+        LOG_ERR("Runtime error: %s", e.what());
+    }
+}
+
+
+
+/**
+ * ReadBMPTable, there will be dedicated thread be launched inside and monitor corresponding redis table.
+ *
+ * \param [in] tables             table names to be subscribed.
+ */
+bool RedisManager::ReadBMPTable(const std::vector<std::string>& tables) {
+    for (const auto& table : tables) {
+        std::shared_ptr<std::thread> threadPtr = std::make_shared<std::thread>(
+            std::bind(&RedisManager::SubscriberWorker, this, table));
+        threadList_.push_back(threadPtr);
+    }
+    return true;
+}
+
+
+/**
+ * Enable specific Table
+ *
+ * \param [in] table    Reference to table name, like BGP_NEIGHBOR_TABLE/BGP_RIB_OUT_TABLE/BGP_RIB_IN_TABLE
+ */
+bool RedisManager::EnableTable(const std::string & table) {
+    enabledTables_.insert(table);
+    return true;
+}
+
+/**
+ * Enable BGP_Neighbor* Table
+ *
+ * \param [in] table    Reference to table name BGP_NEIGHBOR_TABLE/BGP_RIB_OUT_TABLE/BGP_RIB_IN_TABLE
+ */
+bool RedisManager::DisableTable(const std::string & table) {
+    enabledTables_.erase(table);
+    return ResetBMPTable(table);
+}
+
+
+/**
+ * Reset ResetBMPTable, this will flush redis
+ *
+ * \param [in] table    Reference to table name BGP_NEIGHBOR_TABLE/BGP_RIB_OUT_TABLE/BGP_RIB_IN_TABLE
+ */
+bool RedisManager::ResetBMPTable(const std::string & table) {
+
+    LOG_INFO("RedisManager ResetBMPTable %s", table.c_str());
+    swss::Table stateBMPTable(&stateDb_, table);
+    std::vector<std::string> keys;
+    stateBMPTable.getKeys(keys);
+    stateDb_.del(keys);
+
+    return true;
+}
+
+
+
+/**
+ * Reset all Tables once FRR reconnects to BMP, this will not disable table population
+ *
+ * \param [in] N/A
+ */
+void RedisManager::ResetAllTables() {
+    LOG_INFO("RedisManager ResetAllTables");
+    for (const auto& enabledTable : enabledTables_) {
+        ResetBMPTable(enabledTable);
+    }
+}