Make openbmp support redis subscription/population on sonic.

Server/CMakeLists.txt

@@ -15,7 +15,6 @@ if(ENABLE_REDIS)
     add_definitions(-DREDIS_ENABLED)
 endif()
 
-
 # Find and set the env for the mysql c++ connector
 set(HINT_ROOT_DIR
         "${HINT_ROOT_DIR}"
@@ -40,30 +39,57 @@ find_library(LIBYAML_CPP_LIBRARY
         lib64
         lib)
 
-find_path(LIBRDKAFKA_INCLUDE_DIR
-        librdkafka/rdkafkacpp.h
-        HINTS
-        ${HINT_ROOT_DIR}
-        PATH_SUFFIXES
-        include)
-
-find_library(LIBRDKAFKA_LIBRARY
-        NAMES
-        librdkafka.a rdkafka
-        HINTS
-        ${HINT_ROOT_DIR}
-        PATH_SUFFIXES
-        lib64
-        lib)
-
-find_library(LIBRDKAFKA_CPP_LIBRARY
-        NAMES
-        librdkafka++.a rdkafka++
-        HINTS
-        ${HINT_ROOT_DIR}
-        PATH_SUFFIXES
-        lib64
-        lib)
+if (NOT ENABLE_REDIS)
+    find_path(LIBRDKAFKA_INCLUDE_DIR
+            librdkafka/rdkafkacpp.h
+            HINTS
+            ${HINT_ROOT_DIR}
+            PATH_SUFFIXES
+            include)
+
+    find_library(LIBRDKAFKA_LIBRARY
+            NAMES
+            librdkafka.a rdkafka
+            HINTS
+            ${HINT_ROOT_DIR}
+            PATH_SUFFIXES
+            lib64
+            lib)
+
+    find_library(LIBRDKAFKA_CPP_LIBRARY
+            NAMES
+            librdkafka++.a rdkafka++
+            HINTS
+            ${HINT_ROOT_DIR}
+            PATH_SUFFIXES
+            lib64
+            lib)
+else ()
+    find_path(LIBSWSSCOMMON_INCLUDE_DIR
+            swss/dbconnector.h
+            HINTS
+            ${HINT_ROOT_DIR}
+            PATH_SUFFIXES
+            include)
+
+    find_library(LIBHIREDIS_LIBRARY
+            NAMES
+            hiredis
+            HINTS
+            ${HINT_ROOT_DIR}
+            PATH_SUFFIXES
+            lib64
+            lib)
+
+    find_library(LIBSWSSCOMMON_LIBRARY
+            NAMES
+            swsscommon
+            HINTS
+            ${HINT_ROOT_DIR}
+            PATH_SUFFIXES
+            lib64
+            lib)
+endif ()
 
 find_library(LIBRT_LIBRARY
         NAMES
@@ -74,10 +100,23 @@ find_library(LIBRT_LIBRARY
         lib64
         lib)
 
-if (NOT LIBRDKAFKA_INCLUDE_DIR OR NOT LIBRDKAFKA_LIBRARY OR NOT LIBRDKAFKA_CPP_LIBRARY)
-	Message (FATAL_ERROR "Librdkafka was not found, cannot proceed.  Visit https://github.com/edenhill/librdkafka for details on how to install it.")
-#else ()
-#	Message ("lib = " ${LIBRDKAFKA_LIBRARY})
+if (ENABLE_REDIS)
+    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()
+else ()
+    if (NOT LIBRDKAFKA_INCLUDE_DIR OR NOT LIBRDKAFKA_LIBRARY OR NOT LIBRDKAFKA_CPP_LIBRARY)
+        Message (FATAL_ERROR "Librdkafka was not found, cannot proceed.  Visit https://github.com/edenhill/librdkafka for details on how to install it.")
+    else ()
+        Message ("lib = " ${LIBRDKAFKA_LIBRARY})
+    endif()
 endif()
 
 if (NOT LIBYAML_CPP_INCLUDE_DIR OR NOT LIBYAML_CPP_LIBRARY)
@@ -89,19 +128,18 @@ if (NOT LIBRT_LIBRARY AND NOT MACOSX)
 endif()
 
 # Update the include dir
-include_directories(${LIBRDKAFKA_INCLUDE_DIR} ${LIBYAML_CPP_INCLUDE_DIR} src/ src/bmp src/bgp src/bgp/linkstate src/kafka)
+if (NOT ENABLE_REDIS)
+    include_directories(${LIBRDKAFKA_INCLUDE_DIR} ${LIBYAML_CPP_INCLUDE_DIR} src/ src/bmp src/bgp src/bgp/linkstate src/kafka)
+else()
+    include_directories(${LIBSWSSCOMMON_INCLUDE_DIR} ${LIBYAML_CPP_INCLUDE_DIR} src/ src/bmp src/bgp src/bgp/linkstate src/redis)
+endif()
 #link_directories(${LIBRDKAFKA_LIBRARY})
 
 
 # Define the source files to compile
 set (SRC_FILES
 	src/bmp/BMPListener.cpp
 	src/bmp/BMPReader.cpp
-	src/kafka/MsgBusImpl_kafka.cpp
-	src/kafka/KafkaEventCallback.cpp
-	src/kafka/KafkaDeliveryReportCallback.cpp
-    src/kafka/KafkaTopicSelector.cpp
-    src/kafka/KafkaPeerPartitionerCallback.cpp
 	src/openbmp.cpp
 	src/bmp/parseBMP.cpp
 	src/md5.cpp
@@ -121,36 +159,51 @@ set (SRC_FILES
     src/bgp/linkstate/MPLinkStateAttr.cpp
     )
 
+# Add specific files used
+if (NOT ENABLE_REDIS)
+    # Add Kafka-specific source files
+    file(GLOB KAFKA_FILES src/kafka/MsgBusImpl_kafka.cpp src/kafka/KafkaEventCallback.cpp src/kafka/KafkaDeliveryReportCallback.cpp src/kafka/KafkaTopicSelector.cpp src/kafka/KafkaPeerPartitionerCallback.cpp)
+    list(APPEND SRC_FILES ${KAFKA_FILES})
+else ()
+    # Add Redis-specific source files
+    file(GLOB REDIS_FILES src/RedisManager.cpp src/redis/MsgBusImpl_redis.cpp)
+    list(APPEND SRC_FILES ${REDIS_FILES})
+endif ()
+
 # Disable warnings
 add_definitions ("-Wno-unused-result")
 
-# Add C++11
+# Add C++14
 if ("${CMAKE_CXX_COMPILER_ID}" STREQUAL "Clang" OR CMAKE_COMPILER_IS_GNUCXX)
     include(CheckCXXCompilerFlag)
-    check_cxx_compiler_flag(--std=c++11 SUPPORTS_STD_CXX11)
+    check_cxx_compiler_flag(--std=c++14 SUPPORTS_STD_CXX14)
     check_cxx_compiler_flag(--std=c++0x SUPPORTS_STD_CXX01)
-    if(SUPPORTS_STD_CXX11)
-        set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} --std=c++11")
-        set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} --std=c++11")
+    if(SUPPORTS_STD_CXX14)
+        set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} --std=c++14")
+        set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} --std=c++14")
     elseif(SUPPORTS_STD_CXX01)
         set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} --std=c++0x")
         set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} --std=c++0x")
     else()
-        message(ERROR "Compiler does not support --std=c++11 or --std=c++0x. Upgrade gcc 4.7 or greater")
+        message(ERROR "Compiler does not support --std=c++14 or --std=c++0x. Upgrade gcc 4.7 or greater")
     endif()
 endif()
 
 # Set the libs to link
-set (LIBS pthread ${LIBYAML_CPP_LIBRARY} ${LIBRDKAFKA_CPP_LIBRARY} ${LIBRDKAFKA_LIBRARY} z ${SSL_LIBS} dl)
+if (NOT ENABLE_REDIS)
+    set (LIBS pthread ${LIBYAML_CPP_LIBRARY} ${LIBRDKAFKA_CPP_LIBRARY} ${LIBRDKAFKA_LIBRARY} z ${SSL_LIBS} dl zstd)
+else ()
+    set (LIBS pthread ${LIBYAML_CPP_LIBRARY} z ${SSL_LIBS} dl)
+endif ()
 
 # Set the binary
 add_executable (openbmpd ${SRC_FILES})
 
 # Link the binary
-target_link_libraries (openbmpd ${LIBS})
+target_link_libraries (openbmpd ${LIBS} ${LIBSWSSCOMMON_LIBRARY})
 
 if (LIBRT_LIBRARY)
-    target_link_libraries(openbmpd ${LIBRT_LIBRARY})
+    target_link_libraries(openbmpd ${LIBRT_LIBRARY} ${LIBSWSSCOMMON_LIBRARY})
 endif()
 
 # Install the binary and configs

Server/src/Config.cpp

@@ -21,7 +21,10 @@
 #include <boost/algorithm/string/replace.hpp>
 
 #include "Config.h"
+
+#ifndef REDIS_ENABLED
 #include "kafka/KafkaTopicSelector.h"
+#endif
 
 /*********************************************************************//**
  * Constructor for class
@@ -62,6 +65,7 @@ Config::Config() {
      *      The keys match the configuration node/vars. Topic name nodes will be ignored if
      *      not initialized here.
      */
+#ifndef REDIS_ENABLED
     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;
@@ -74,6 +78,7 @@ Config::Config() {
     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;
+#endif
 }
 
 /*********************************************************************//**

Server/src/RedisManager.cpp

@@ -0,0 +1,154 @@
+/*
+ * 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() {
+    exit_ = false;
+}
+
+/*********************************************************************//**
+ * Constructor for class
+ ***********************************************************************/
+RedisManager::~RedisManager() {
+}
+
+
+/*********************************************************************
+ * Setup for this class
+ *
+ * \param [in] logPtr     logger pointer
+ ***********************************************************************/
+void RedisManager::Setup(Logger *logPtr) {
+    logger = logPtr;
+    if (!swss::SonicDBConfig::isInit()) {
+        swss::SonicDBConfig::initialize();
+    }
+
+    stateDb_ =  std::make_shared<swss::DBConnector>(BMP_DB_NAME, 0, false);
+    separator_ = swss::SonicDBConfig::getSeparator(BMP_DB_NAME);
+}
+
+
+
+/**
+ * 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;
+    }
+    std::unique_ptr<swss::Table> stateBMPTable = std::make_unique<swss::Table>(stateDb_.get(), table);
+    std::ostringstream oss;
+    for (const auto& key : keys) {
+        oss << key << separator_;
+    }
+    std::string fullKey = oss.str();
+    fullKey.pop_back();
+
+    DEBUG("RedisManager WriteBMPTable key = %s", fullKey.c_str());
+
+    stateBMPTable->set(fullKey, fieldValues);
+    return true;
+}
+
+
+/**
+ * RemoveEntityFromBMPTable
+ *
+ * \param [in] keys             Reference to various keys
+ */
+bool RedisManager::RemoveEntityFromBMPTable(const std::vector<std::string>& keys) {
+
+    for (const auto& key : keys) {
+        DEBUG("RedisManager RemoveEntityFromBMPTable key = %s", key.c_str());
+    }
+    stateDb_->del(keys);
+    return true;
+}
+
+
+/**
+ * ExitRedisManager
+ *
+ * \param [in] N/A
+ */
+void RedisManager::ExitRedisManager() {
+    exit_ = true;
+}
+
+
+/**
+ * InitBMPConfig, read config_db for table enablement setting.
+ *
+ * \param [in] N/A
+ */
+bool RedisManager::InitBMPConfig() {
+    std::shared_ptr<swss::DBConnector> cfgDb =
+        std::make_shared<swss::DBConnector>("CONFIG_DB", 0, false);
+    std::unique_ptr<swss::Table> cfgTable = std::make_unique<swss::Table>(cfgDb.get(), BMP_CFG_TABLE_NAME);
+    std::vector<swss::FieldValueTuple> fvt;
+    cfgTable->get(BMP_CFG_TABLE_KEY, fvt);
+    for (const auto& item : fvt) {
+        if (item.second == "true") {
+            enabledTables_.insert(item.first);
+        }
+    }
+    return true;
+}
+
+
+/**
+ * Reset ResetBMPTable, this will flush redis
+ *
+ * \param [in] table    Reference to table name BGP_NEIGHBOR_TABLE/BGP_RIB_OUT_TABLE/BGP_RIB_IN_TABLE
+ */
+void RedisManager::ResetBMPTable(const std::string & table) {
+
+    LOG_INFO("RedisManager ResetBMPTable %s", table.c_str());
+    std::unique_ptr<swss::Table> stateBMPTable = std::make_unique<swss::Table>(stateDb_.get(), table);
+    std::vector<std::string> keys;
+    stateBMPTable->getKeys(keys);
+    stateDb_->del(keys);
+}
+
+
+
+/**
+ * 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);
+    }
+}