New input/process plugin to accept IPFIX flow records on a socket

CMakeLists.txt

@@ -12,14 +12,16 @@ include(cmake/installation.cmake)
 
 set(CMAKE_MODULE_PATH ${CMAKE_SOURCE_DIR}/cmake/modules)
 
+option(ENABLE_INPUT_SOCK            "Enable build of input SOCK plugin"                       OFF)
 option(ENABLE_INPUT_PCAP            "Enable build of input PCAP plugin"                       OFF)
 option(ENABLE_INPUT_DPDK            "Enable build of input DPDK plugin"                       OFF)
 option(ENABLE_INPUT_NFB             "Enable build of input NFB plugin"                        OFF)
 option(ENABLE_OUTPUT_UNIREC         "Enable build of output UNIREC plugin"                    OFF)
+option(ENABLE_OUTPUT_LZ4            "Enable Build of output LZ4 compression"                  OFF)
 option(ENABLE_PROCESS_EXPERIMENTAL  "Enable build of experimental process plugins"            OFF)
 option(ENABLE_MILLISECONDS_TIMESTAMP "Compile ipfixprobe with miliseconds timestamp precesion" OFF)
 option(ENABLE_NEMEA                 "Enable build of NEMEA plugins"                           OFF)
-
+option(ENABLE_UNWIND                "Enable Build with Lib unwind"                            OFF)
 option(ENABLE_RPMBUILD              "Enable build of RPM package"                             ON)
 option(ENABLE_TESTS                 "Build tests (make test)"                                 OFF)
 

cmake/dependencies.cmake

@@ -3,8 +3,13 @@ find_package(PkgConfig REQUIRED)
 
 find_package(Threads REQUIRED)
 find_package(Atomic REQUIRED)
-find_package(Unwind REQUIRED)
-find_package(LZ4 REQUIRED)
+if (ENABLE_UNWIND)
+	find_package(Unwind REQUIRED)
+endif()
+
+if (ENABLE_OUTPUT_LZ4)
+	find_package(LZ4 REQUIRED)
+endif()
 find_package(OpenSSL REQUIRED)
 
 if (ENABLE_INPUT_PCAP)

include/ipfixprobe/flowifc.hpp

@@ -253,9 +253,12 @@ struct Flow : public Record {
 	uint32_t dst_packets;
 	uint8_t src_tcp_flags;
 	uint8_t dst_tcp_flags;
+	uint32_t drop_packets;
+	uint32_t ing_phy_interface;
 
 	uint8_t ip_version;
 
+	uint8_t ip_tos;
 	uint8_t ip_proto;
 	uint16_t src_port;
 	uint16_t dst_port;

include/ipfixprobe/ipfix-elements.hpp

@@ -89,6 +89,8 @@ namespace ipxp {
 #define OUTPUT_INTERFACE(F) F(0, 14, 2, nullptr)
 #define FLOW_END_REASON(F) F(0, 136, 1, &flow.end_reason)
 #define FLOW_ID(F) F(0, 148, 8, &flow.flow_hash)
+#define DROPS(F)                      F(0,      133,    4,   &flow.drop_packets)
+#define ING_PHY_INTERFACE(F)          F(0,      252,    4,   &flow.ing_phy_interface)
 
 #define ETHERTYPE(F) F(0, 256, 2, nullptr)
 
@@ -100,9 +102,10 @@ namespace ipxp {
 #define L3_PROTO(F) F(0, 60, 1, &flow.ip_version)
 #define L3_IPV4_ADDR_SRC(F) F(0, 8, 4, &flow.src_ip.v4)
 #define L3_IPV4_ADDR_DST(F) F(0, 12, 4, &flow.dst_ip.v4)
-#define L3_IPV4_TOS(F) F(0, 5, 1, nullptr)
+#define L3_IPV4_TOS(F) F(0, 5, 1, &flow.ip_tos)
 #define L3_IPV6_ADDR_SRC(F) F(0, 27, 16, &flow.src_ip.v6)
 #define L3_IPV6_ADDR_DST(F) F(0, 28, 16, &flow.dst_ip.v6)
+#define L3_IPV6_TOS(F) F(0, 5, 1, &flow.ip_tos)
 #define L3_IPV4_IDENTIFICATION(F) F(0, 54, 2, nullptr)
 #define L3_IPV4_FRAGMENT(F) F(0, 88, 2, nullptr)
 #define L3_IPV4_TTL(F) F(0, 192, 1, nullptr)
@@ -343,7 +346,8 @@ namespace ipxp {
 	F(L3_IPV4_ADDR_SRC)                                                                            \
 	F(L3_IPV4_ADDR_DST)                                                                            \
 	F(L2_SRC_MAC)                                                                                  \
-	F(L2_DST_MAC)
+	F(L2_DST_MAC)                                                                                  \
+	F(L3_IPV4_TOS)
 
 #define BASIC_TMPLT_V6(F)                                                                          \
 	F(FLOW_END_REASON)                                                                             \
@@ -363,7 +367,8 @@ namespace ipxp {
 	F(L3_IPV6_ADDR_SRC)                                                                            \
 	F(L3_IPV6_ADDR_DST)                                                                            \
 	F(L2_SRC_MAC)                                                                                  \
-	F(L2_DST_MAC)
+	F(L2_DST_MAC)                                                                                  \
+	F(L3_IPV6_TOS)
 
 #define IPFIX_HTTP_TEMPLATE(F)                                                                     \
 	F(HTTP_USERAGENT)                                                                              \
@@ -583,6 +588,10 @@ namespace ipxp {
 
 #define IPFIX_MPLS_TEMPLATE(F) F(MPLS_TOP_LABEL_STACK_SECTION)
 
+#define IPFIX_SOCKPKTINFO_TEMPLATE(F) \
+	F(ING_PHY_INTERFACE) \
+	F(DROPS)
+
 /**
  * List of all known templated.
  *
@@ -616,7 +625,8 @@ namespace ipxp {
 	IPFIX_ICMP_TEMPLATE(F)                                                                         \
 	IPFIX_VLAN_TEMPLATE(F)                                                                         \
 	IPFIX_NETTISA_TEMPLATE(F)                                                                      \
-	IPFIX_FLOW_HASH_TEMPLATE(F)
+	IPFIX_FLOW_HASH_TEMPLATE(F)                                                                    \
+	IPFIX_SOCKPKTINFO_TEMPLATE(F)
 
 /**
  * Helper macro, convert FIELD into its name as a C literal.

include/ipfixprobe/packet.hpp

@@ -100,6 +100,12 @@ struct Packet : public Record {
 	uint16_t buffer_size; /**< Size of buffer */
 
 	bool source_pkt; /**< Direction of packet from flow point of view */
+	uint32_t pkt_cnt; /**< Number of packets on a flow - input plugin 'sock' */
+	uint32_t drop_cnt; /**< Number of dropped packets on a flow - input plugin 'sock' */
+	uint64_t byte_cnt; /**< Bytes received on a flow - input plugin 'sock' */
+	uint32_t source_interface; /**< source interface for the flow  - input plugin 'sock' */
+	uint8_t end_reason; /**< Flow end reason - input plugin 'sock' */
+	struct timeval end_ts; /**< Flow end time - input plugin 'sock' */
 
 	/**
 	 * \brief Constructor.
@@ -142,6 +148,12 @@ struct Packet : public Record {
 		, buffer(nullptr)
 		, buffer_size(0)
 		, source_pkt(true)
+		, pkt_cnt(0)
+		, drop_cnt(0)
+		, byte_cnt(0)
+		, source_interface(0)
+		, end_reason (0)
+		, end_ts({0, 0})
 	{
 	}
 };

init/config2args.py

@@ -39,6 +39,8 @@ def process_input_plugin(config):
         return process_input_dpdk_plugin(settings)
     if plugin == "dpdk_ring":
         return process_input_dpdk_ring_plugin(settings)
+    if plugin == "sock":
+        return process_input_sock_plugin(settings)
     if plugin == "raw":
         return process_input_raw_plugin(settings)
     if plugin == "ndp":
@@ -292,6 +294,19 @@ def process_input_raw_plugin(settings):
 
     return " ".join(params)
 
+def process_input_sock_plugin(settings):
+    params = ['-i "sock']
+
+    if settings is None:
+        raise ValueError("Settings for sock plugin cannot be empty.")
+
+    sock = settings.get("sock")
+    if sock is None:
+        raise ValueError("sock must be specified in the sock plugin configuration.")
+
+    params.append(f"sock={sock}")
+
+    return " ".join(params)
 
 def process_process_plugins(config):
     process_plugins = config.get("process_plugins", [])

init/link0.conf.example

@@ -1,7 +1,7 @@
 # Input plugin configuration (input_plugin)
 input_plugin:
   # IMPORTANT: Only one input plugin can be specified. Choose one of the following options:
-  # raw, pcap_file, pcap_live, ndp, dpdk_ring, or dpdk.
+  # raw, pcap_file, pcap_live, ndp, dpdk_ring, or dpdk, sock.
 
   raw:
     interface: eth0  # Network interface name to capture traffic from [required]
@@ -35,6 +35,9 @@ input_plugin:
     eal_opts: null  # EAL options (null = default options)
     mtu: null  # Maximum Transmission Unit (defaults to RTE_ETHER_MAX_LEN)
 
+  sock:
+    sock: Unix domain socket path [required]
+
 # Storage configuration (storage)
 storage:
   cache:

init/schema.json

@@ -32,6 +32,26 @@
             "raw"
           ]
         },
+        {
+          "type": "object",
+          "properties": {
+            "sock": {
+              "type": "object",
+              "properties": {
+                "path": {
+                  "type": "string"
+                }
+              },
+              "required": [
+                "path"
+              ],
+              "additionalProperties": false
+            }
+          },
+          "required": [
+            "sock"
+          ]
+        },
         {
           "type": "object",
           "properties": {

pkg/rpm/CMakeLists.txt

@@ -22,6 +22,10 @@ if (ENABLE_INPUT_PCAP)
 	list(APPEND RPMBUILD_ARGS "--with" "input_pcap")
 endif()
 
+if (ENABLE_INPUT_SOCK)
+	list(APPEND RPMBUILD_ARGS "--with" "input_sock")
+endif()
+
 if (ENABLE_INPUT_DPDK)
 	list(APPEND RPMBUILD_ARGS "--with" "input_dpdk")
 endif()

pkg/rpm/ipfixprobe-msec.spec.in

@@ -29,18 +29,25 @@ BuildRequires: gcc-c++ >= 10
 
 BuildRequires: make
 BuildRequires: cmake >= 3.12
+%if %{with unwind}
 BuildRequires: libunwind-devel
+%endif
+
 %if 0%{?rhel} <= 9
 BuildRequires: gcc-toolset-14-libatomic-devel
 %endif
 BuildRequires: pkgconfig
+%if %{with lz4}
 BuildRequires: lz4-devel
+%endif
 BuildRequires: openssl-devel
 BuildRequires: git
 
 Requires: libatomic
 Requires: fuse3
+%if %{with lz4}
 Requires: lz4
+%endif
 Requires: openssl
 Requires: python3
 Requires: python3-pyyaml

pkg/rpm/ipfixprobe-nemea.spec.in

@@ -32,19 +32,26 @@ BuildRequires: gcc-c++ >= 10
 
 BuildRequires: make
 BuildRequires: cmake >= 3.12
+%if %{with unwind}
 BuildRequires: libunwind-devel
+%endif
+
 %if 0%{?rhel} <= 9
 BuildRequires: gcc-toolset-14-libatomic-devel
 %endif
 BuildRequires: pkgconfig
+%if %{with lz4}
 BuildRequires: lz4-devel
+%endif
 BuildRequires: openssl-devel
 BuildRequires: nemea-framework-devel
 BuildRequires: git
 
 Requires: libatomic
 Requires: fuse3
+%if %{with lz4}
 Requires: lz4
+%endif
 Requires: openssl
 Requires: python3
 Requires: python3-pyyaml

pkg/rpm/ipfixprobe.spec.in

@@ -2,6 +2,7 @@
 %bcond_with input_dpdk
 %bcond_with input_nfb
 %bcond_with process_experimental
+%bcond_with input_sock
 
 %global _unitdir %{_prefix}/lib/systemd/system
 
@@ -34,18 +35,25 @@ BuildRequires: gcc-c++ >= 10
 
 BuildRequires: make
 BuildRequires: cmake >= 3.12
+%if %{with unwind}
 BuildRequires: libunwind-devel
+%endif
+
 %if 0%{?rhel} <= 9
 BuildRequires: gcc-toolset-14-libatomic-devel
 %endif
 BuildRequires: pkgconfig
+%if %{with lz4}
 BuildRequires: lz4-devel
+%endif
 BuildRequires: openssl-devel
 BuildRequires: git
 
 Requires: libatomic
 Requires: fuse3
+%if %{with lz4}
 Requires: lz4
+%endif
 Requires: openssl
 Requires: python3
 Requires: python3-pyyaml
@@ -64,6 +72,16 @@ BuildRequires: libpcap-devel
 Input plugin for libpcap.
 %endif
 
+%if %{with input_sock}
+%package input-sock
+Summary: Input plugin to read flow records from a unix domain socket using libsock.
+Requires: libsock
+BuildRequires: libsock-devel
+
+%description input-sock
+Input plugin for libsock.
+%endif
+
 %if %{with input_dpdk}
 %package input-dpdk
 Summary: Input plugin to read packets from interfaces using dpdk.
@@ -104,7 +122,7 @@ Experimental process plugins.
 %if 0%{?rhel} == 8
 source /opt/rh/gcc-toolset-14/enable
 %endif
-%cmake -DCMAKE_BUILD_TYPE=Release %{?with_input_pcap:-DENABLE_INPUT_PCAP=ON} %{?with_input_dpdk:-DENABLE_INPUT_DPDK=ON} %{?with_input_nfb:-DENABLE_INPUT_NFB=ON} %{?with_process_experimental: -DENABLE_PROCESS_EXPERIMENTAL=ON}
+%cmake -DCMAKE_BUILD_TYPE=Release %{?with_unwind:-DENABLE_UNWIND=ON} %{?with_output_lz4:-DENABLE_OUTPUT_LZ4=ON} %{?with_input_pcap:-DENABLE_INPUT_PCAP=ON} %{?with_input_sock:-DENABLE_INPUT_SOCK=ON} %{?with_input_dpdk:-DENABLE_INPUT_DPDK=ON} %{?with_input_nfb:-DENABLE_INPUT_NFB=ON} %{?with_process_experimental: -DENABLE_PROCESS_EXPERIMENTAL=ON}
 %cmake_build
 
 %install
@@ -157,6 +175,12 @@ source /opt/rh/gcc-toolset-14/enable
 %{_libdir}/ipfixprobe/input/libipfixprobe-input-pcap.so
 %endif
 
+%if %{with input_sock}
+%files input-sock
+%{_libdir}/ipfixprobe/input/libipfixprobe-input-sock.so
+%{_libdir}/ipfixprobe/process/libipfixprobe-process-sockpktinfo.so
+%endif
+
 %if %{with input_nfb}
 %files input-nfb
 %{_libdir}/ipfixprobe/input/libipfixprobe-input-nfb.so

src/core/CMakeLists.txt

@@ -30,10 +30,17 @@ target_link_libraries(ipfixprobe-core
 	telemetry::telemetry
 	telemetry::appFs
 	atomic::atomic
-	unwind::unwind
 	${CMAKE_DL_LIBS}
 )
 
+if (ENABLE_UNWIND)
+target_link_libraries(ipfixprobe-core unwind::unwind)
+
+target_compile_definitions(ipfixprobe-core PRIVATE
+	WITH_UNWIND
+)
+endif()
+
 add_executable(ipfixprobe main.cpp)
 target_link_libraries(ipfixprobe ${CORE_LIB})
 target_link_options(ipfixprobe PRIVATE -Wl,--export-dynamic)

src/core/ipfixprobe.cpp

@@ -65,7 +65,9 @@ void signal_handler(int sig)
 {
 	(void) sig;
 	if (sig == SIGSEGV || sig == SIGABRT) {
+#ifdef WITH_UNWIND
 		st_dump(STDERR_FILENO, sig);
+#endif
 		exit(EXIT_FAILURE);
 	}
 	stop = 1;
@@ -77,9 +79,7 @@ void register_handlers()
 	signal(SIGINT, signal_handler);
 	signal(SIGSEGV, signal_handler);
 	signal(SIGABRT, signal_handler);
-#ifdef WITH_NEMEA
 	signal(SIGPIPE, SIG_IGN);
-#endif
 }
 
 void error(std::string msg)

src/core/stacktrace.cpp

@@ -36,6 +36,7 @@
 #define UNW_LOCAL_ONLY
 #include "stacktrace.hpp"
 
+#ifdef WITH_UNWIND
 #include <libunwind.h>
 
 namespace ipxp {
@@ -192,3 +193,4 @@ void st_dump(int fd, int sig)
 }
 
 } // namespace ipxp
+#endif