Completely working SnabbSwitch integration. 12 mpps with 4 NIC's

README.md

@@ -12,7 +12,7 @@ Detailed reference in Russian: [link](https://github.com/FastVPSEestiOu/fastnetm
 
 License: GPLv2
 
-FastNetMon - A high performance DoS/DDoS load analyzer built on top of multiple packet capture engines (NetFlow, IPFIX, sFLOW, netmap, PF_RING, PCAP).
+FastNetMon - A high performance DoS/DDoS load analyzer built on top of multiple packet capture engines (NetFlow, IPFIX, sFLOW, SnabbSwitch, netmap, PF_RING, PCAP).
 
 What can we do? We can detect hosts in our own network with a large amount of packets per second/bytes per second or flow per second incoming or outgoing from certain hosts. And we can call an external script which can notify you, switch off a server or blackhole the client.
 
@@ -33,7 +33,7 @@ Supported packet capture engines:
 - NetFlow v5, v9
 - IPFIX
 - ![sFLOW](http://sflow.org/images/sflowlogo.gif) v4 (dev branch only), v5
-- Port mirror/SPAN capture with PF_RING (with ZC/DNA mode support [need license](http://www.ntop.org/products/pf_ring/)), NETMAP and PCAP
+- Port mirror/SPAN capture with PF_RING (with ZC/DNA mode support [need license](http://www.ntop.org/products/pf_ring/)), SnabbSwitch, NETMAP and PCAP
 
 You could look [comparison table](https://github.com/FastVPSEestiOu/fastnetmon/blob/master/docs/CAPTURE_BACKENDS.md) for all available packet capture engines.
 
@@ -46,6 +46,7 @@ Features:
 - Full integration with [Graphite](docs/GRAPHITE_INTEGRATION.md) and [InfluxDB](docs/INFLUXDB_INTEGRATION.md)
 - Deep packet inspection for attack traffic
 - netmap support (open source; wire speed processing; only Intel hardware NICs or any hypervisor VM type)
+- SnabbSwitch support (open source, very flexible, LUA driven, very-very-very fast)
 - Could filter out NetFLOW v5 flows or sFLOW packets with script implemented in LUA (useful for port exclude)
 - Supports L2TP decapsulation, VLAN untagging and MPLS processing in mirror mode 
 - Can work on server/soft-router

src/CMakeLists.txt

@@ -130,6 +130,7 @@ add_library(ipfix_rfc STATIC ipfix_rfc.cpp)
 add_library(fastnetmon_packet_parser STATIC fastnetmon_packet_parser.c)
 
 # -DENABLE_SNABBSWITCH_SUPPORT=ON ..
+# Please also comment out line: set(ENABLE_LUA_SUPPORT yes) if you want SnabbSwitch support
 if (ENABLE_SNABBSWITCH_SUPPORT) 
     add_definitions(-DSNABB_SWITCH)
     add_library(snabbswitch_plugin STATIC snabbswitch_plugin/snabbswitch_collector.cpp)
@@ -315,7 +316,11 @@ target_link_libraries(fastnetmon ipfix_rfc)
 target_link_libraries(fastnetmon fast_library)
 
 if (ENABLE_PFRING_SUPPORT)
-target_link_libraries(fastnetmon pfring_plugin)
+    target_link_libraries(fastnetmon pfring_plugin)
+endif()
+
+if (ENABLE_SNABBSWITCH_SUPPORT)
+    target_link_libraries(fastnetmon snabbswitch_plugin)
 endif()
 
 target_link_libraries(fastnetmon sflow_plugin netflow_plugin pcap_plugin example_plugin netmap_plugin)

src/fastnetmon.conf

@@ -72,6 +72,12 @@ pfring_sampling_ratio = 1
 # Netmap traffic capture (very fast but need patched drivers)
 mirror_netmap = off
 
+# SnabbSwitch traffic capture
+mirror_snabbswitch = off
+
+# We use PCI-e addresses here instead of OS device names. You could find they in lspci output
+interfaces_snabbswitch = 0000:04:00.0,0000:04:00.1,0000:03:00.0,0000:03:00.1
+
 # Port mirroring could be sampled
 netmap_sampling_ratio = 1
 

src/fastnetmon.cpp

@@ -46,6 +46,10 @@
 #include "pfring_plugin/pfring_collector.h"
 #endif
 
+#ifdef SNABB_SWITCH
+#include "snabbswitch_plugin/snabbswitch_collector.h"
+#endif
+
 // Yes, maybe it's not an good idea but with this we can guarantee working code in example plugin
 #include "example_plugin/example_collector.h"
 
@@ -185,6 +189,7 @@ void init_global_ban_settings() {
 
 bool enable_conection_tracking = true;
 
+bool enable_snabbswitch_collection = false;
 bool enable_data_collection_from_mirror = true;
 bool enable_netmap_collection = false;
 bool enable_sflow_collection = false;
@@ -997,6 +1002,10 @@ bool load_configuration_file() {
         }
     }
 
+    if (configuration_map.count("mirror_snabbswitch") != 0) {
+        enable_snabbswitch_collection = configuration_map["mirror_snabbswitch"] == "on";
+    } 
+
     if (enable_netmap_collection && enable_data_collection_from_mirror) {
         logger << log4cpp::Priority::ERROR << "You have enabled pfring and netmap data collection "
                                               "from mirror which strictly prohibited, please "
@@ -2489,6 +2498,12 @@ int main(int argc, char** argv) {
         packet_capture_plugin_thread_group.add_thread(new boost::thread(start_netmap_collection, process_packet));
     }
 
+#ifdef SNABB_SWITCH 
+    if (enable_snabbswitch_collection) {
+        packet_capture_plugin_thread_group.add_thread(new boost::thread(start_snabbswitch_collection, process_packet));
+    }
+#endif
+
     if (enable_sflow_collection) {
         packet_capture_plugin_thread_group.add_thread(new boost::thread(start_sflow_collection, process_packet));
     }

src/snabbswitch_plugin/snabbswitch_collector.cpp

@@ -38,7 +38,7 @@ extern "C" {
 #endif
 
 // This code defined in SnabbSwitch
-int start_snabb_switch(int snabb_argc, char **snabb_argv);
+int start_snabb_switch(int snabb_argc, const char **snabb_argv);
 
 #ifdef __cplusplus
 }
@@ -75,7 +75,6 @@ struct firehose_rdesc {
 bool parse_raw_packet_to_simple_packet(u_char* buffer, int len, simple_packet& packet);
 
 void firehose_packet(const char *pciaddr, char *data, int length) {
-    //__sync_fetch_and_add(&received_packets, 1);
     simple_packet packet;
 
     if (!parse_raw_packet_to_simple_packet((u_char*)data, length, packet)) {
@@ -122,21 +121,34 @@ void start_snabbswitch_collection(process_packet_pointer func_ptr) {
     std::vector<std::string> interfaces_for_capture;
     boost::split(interfaces_for_capture, interfaces_list, boost::is_any_of(","), boost::token_compress_on);
 
+    if (interfaces_for_capture.size()  == 0) {
+        logger << log4cpp::Priority::ERROR << "Please specify list of PCI-e addresses for SnabbSwitch capture";
+    }
+
     logger << log4cpp::Priority::INFO << "SnabbSwitch will listen on " << interfaces_for_capture.size() << " interfaces";
-   
-    // TODO read this from configureation! 
-    char* cli_arguments[] = {
-        "snabb", // emulate call of standard application
-        "firehose",
-        "--input",
-        "0000:03:00.0",
-        "--input",
-        "0000:03:00.1",
-        "weird_data"
-    };
+  
+    boost::thread_group snabbswitch_main_threads;
 
-    int cli_number_of_arguments = sizeof(cli_arguments) / sizeof(char*);
+    for (std::vector<std::string>::iterator interface = interfaces_for_capture.begin(); 
+        interface != interfaces_for_capture.end(); ++interface) {
+     
+        // We could specify multiple NIC's for single thread with multiple --input
+        const char* cli_arguments[5];
 
-    start_snabb_switch(cli_number_of_arguments, cli_arguments); 
+        cli_arguments[0] = "snabb"; // emulate call of standard application
+        cli_arguments[1] = "firehose";
+        cli_arguments[2] = "--input";
+        cli_arguments[3] = interface->c_str();
+        cli_arguments[4] ="weird_data";
+
+        int cli_number_of_arguments = sizeof(cli_arguments) / sizeof(char*);
+    
+        logger << log4cpp::Priority::INFO << "We are starting SnabbSwitch instance for PCIe interface " << *interface; 
+        snabbswitch_main_threads.add_thread( new boost::thread(start_snabb_switch, cli_number_of_arguments, cli_arguments) );
+        // We should sleep here because init code of SnabbSwitch is not thread safe
+        sleep(10); 
+    }
+
+    snabbswitch_main_threads.join_all();
 }