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/openbmpd.conf

@@ -85,6 +85,8 @@ debug:
   bgp:     false       # BGP related
   msgbus:  false       # Kafka/message bus - this will enable librdkafka debugging as well
 
+redis:
+  multiAsic:     false       # SONiC multiASIC related
 
 kafka:
 

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
@@ -34,6 +37,9 @@ Config::Config() {
     debug_bgp           = false;
     debug_bmp           = false;
     debug_msgbus        = false;
+#ifdef REDIS_ENABLED
+    redis_multiAsic     = false;
+#endif
     bmp_buffer_size     = 15 * 1024 * 1024; // 15MB
     svr_ipv6            = false;
     svr_ipv4            = true;
@@ -62,6 +68,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 +81,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
 }
 
 /*********************************************************************//**
@@ -104,6 +112,10 @@ void Config::load(const char *cfg_filename) {
                         parseDebug(node);
                     else if (key.compare("kafka") == 0)
                         parseKafka(node);
+#ifdef REDIS_ENABLED
+                    else if (key.compare("redis") == 0)
+                        parseRedis(node);
+#endif
                     else if (key.compare("mapping") == 0)
                         parseMapping(node);
 
@@ -578,7 +590,26 @@ void Config::parseKafka(const YAML::Node &node) {
     }
 }
 
+#ifdef REDIS_ENABLED
+/**
+ * Parse the Redis configuration
+ *
+ * \param [in] node     Reference to the yaml NODE
+ */
+void Config::parseRedis(const YAML::Node &node) {
+    if (!redis_multiAsic and node["multiAsic"]) {
+        try {
+            redis_multiAsic = node["multiAsic"].as<bool>();
 
+            if (redis_multiAsic)
+                std::cout << "   Config: redis multiAsic : " << redis_multiAsic << std::endl;
+
+        } catch (YAML::TypedBadConversion<bool> err) {
+            printWarning("redis.multiAsic is not of type boolean", node["multiAsic"]);
+        }
+    }
+}
+#endif
 
 /**
  * Parse the kafka topics configuration

Server/src/Config.h

@@ -46,6 +46,9 @@ class Config {
     bool        debug_bgp;
     bool        debug_bmp;
     bool        debug_msgbus;
+#ifdef REDIS_ENABLED
+    bool        redis_multiAsic;
+#endif
 
     int         heartbeat_interval;      ///< Heartbeat interval in seconds for collector updates
     int   	tx_max_bytes;            ///< Maximum transmit message size
@@ -150,6 +153,15 @@ class Config {
      */
     void parseKafka(const YAML::Node &node);
 
+#ifdef REDIS_ENABLED
+    /**
+     * Parse the redis configuration
+     *
+     * \param [in] node     Reference to the yaml NODE
+     */
+    void parseRedis(const YAML::Node &node);
+#endif
+
     /**
      * Parse the kafka topics configuration
      *