Merge remote-tracking branch 'origin/master' into topic/johanna/postgres

CHANGES

@@ -1,4 +1,47 @@
 
+0.2-84 | 2016-05-12 07:10:27 -0700
+
+  * Fix to get kafka plugin to be visible in the docs. (Daniel Thayer)
+
+  * BIT-1586 Fixed thread-safety issues with Kafka Writer. (Nick Allen)
+
+  * Fix a tiny logic error that shows up when shutting down the
+    Kafka Writer. (Seth Hall)
+
+  * BIT-1559 allow users to define different kafka topics for each
+    log-stream. (Nick Allen)
+
+0.2-75 | 2016-05-07 12:15:43 -0700
+
+  * BIT-1543: Updated README for Kafka writer to include compatible
+    librdkafka versions. (Nick Allen)
+
+0.2-72 | 2016-03-10 13:22:26 -0800
+
+  * af_packet: Updated README. (Jan Grashoefer)
+
+  * af_packet: Automatically enable promiscuous mode. Socket is now
+    set to promiscuous mode using PACKET_ADD_MEMBERSHIP. This will not
+    be reflected by the interface flags but will show up in the kernel
+    log. (Jan Grashoefer)
+
+  * af_packet: Fixed initialization of statistics. (Jan Grashoefer)
+
+0.2-66 | 2016-03-10 12:05:13 -0800
+
+  * Add plugin to write log output to Kafka. (Nick Allen)
+
+  * Fix failing ElasticSearch tests due to netcat. (Daniel Thayer)
+
+  * Fix compile error by replacing u_char with uint8_t. (Johanna
+    Amann)
+
+  * Fixed 'abs' ambiguity. (James Swaro)
+
+0.2-59 | 2016-02-08 12:32:23 -0800
+
+  * Capitalize Myricom_ROOT_DIR in configure.plugin. (Keith Butler)
+
 0.2-53 | 2016-01-25 15:56:45 -0800
 
   * Add TCPRS plugin. (James Swaro)

Makefile

@@ -3,8 +3,8 @@
 # building and testing.
 #
 
-build-all: build-elasticsearch build-netmap build-pf_ring build-redis build-myricom build-af_packet build-tcprs build-postgresql
-test-all:  test-elasticsearch test-netmap test-pf_ring test-redis test-myricom test-af_packet test-tcprs test-postgresql
+build-all: build-elasticsearch build-netmap build-pf_ring build-redis build-myricom build-af_packet build-tcprs build-kafka build-postgresql
+test-all:  test-elasticsearch test-netmap test-pf_ring test-redis test-myricom test-af_packet test-tcprs test-kafka test-postgresql
 
 build-elasticsearch:
 	make -C elasticsearch
@@ -27,6 +27,9 @@ build-af_packet:
 build-tcprs:
 	make -C tcprs
 
+build-kafka:
+	make -C kafka
+
 build-postgresql:
 	make -C postgresql
 
@@ -51,5 +54,8 @@ test-af_packet:
 test-tcprs:
 	make -C tcprs test
 
+test-kafka:
+	make -C kafka test
+
 test-postgresql:
 	make -C postgresql test

README

@@ -16,7 +16,9 @@ risk.
 
    af_packet - A packet source providing native AF_Packet support on Linux. <af_packet/README>
 
-   elasticsearch - A log writer adding support for the Apache Lucene-based ElasticSearch database <elasticsearch/README>
+   elasticsearch-deprecated - A log writer adding support for the Apache Lucene-based ElasticSearch database <elasticsearch-deprecated/README>
+
+   kafka - A log writer that sends logging output to Kafka <kafka/README>
 
    myricom - A packet source providing native `Myricom SNF v3` support <myricom/README>
 

VERSION

@@ -1 +1 @@
-0.2-53
+0.2-84

af_packet/README

@@ -11,7 +11,7 @@ Installation
 Make sure the kernel headers are installed and your kernel supports
 PACKET_FANOUT and TPACKET_V3. The following will then compile and
 install the af_packet plugin alongside Bro, assuming it can find the
-kernel headers in a standard location:
+kernel headers in a standard location::
 
     # ./configure && make && make install
 
@@ -28,24 +28,25 @@ this::
     [Constant] AF_Packet::enable_fanout
     [Constant] AF_Packet::fanout_id
 
-To use AF_Packet you need to give the Bro processes the CAP_NET_RAW capability if running as non-root.
-You can set it with the following command (on each sensor, after broctl install).
-
-setcap cap_net_raw+eip <path_to_bro>/bin/bro
-
 Usage
 -----
 
 Once installed, you can use AF_Packet interfaces/ports by prefixing them
 with ``af_packet::`` on the command line. For example, to use AF_Packet
-to monitor interface ``eth0``:
+to monitor interface ``eth0``::
+
+    # bro -i af_packet::eth0
 
-    bro -i af_packet::eth0
+To use AF_Packet, running Bro without root privileges, the Bro processes needs
+the CAP_NET_RAW capability. You can set it with the following command (on each
+sensor, after broctl install)::
 
-The AF_Packet plugin does not enable promiscuous mode on interfaces,
-you'll have to do that yourself. For example, on Linux:
+    # setcap cap_net_raw+eip <path_to_bro>/bin/bro
 
-    ifconfig eth0 promisc
+The AF_Packet plugin automatically enables promiscuous mode on the interfaces.
+As the plugin is using PACKET_ADD_MEMBERSHIP to enter the promiscuous mode
+without interfering others, the PROMISC flag is not touched. To verify that the
+interface entered promiscuous mode you can use ``dmesg``.
 
 To adapt the plugin to your needs, you can set a couple of parameters like
 buffer size. See scripts/init.bro for the default values.

af_packet/README.rst

@@ -0,0 +1 @@
+README

af_packet/src/AF_Packet.cc

@@ -80,6 +80,7 @@ void AF_PacketSource::Open()
 	props.is_live = true;
 	props.link_type = DLT_EN10MB; // Ethernet headers
 
+	memset(&stats, 0, sizeof(stats));
 	num_discarded = 0;
 
 	Opened(props);
@@ -109,9 +110,23 @@ inline bool AF_PacketSource::BindInterface()
 
 inline bool AF_PacketSource::EnablePromiscMode()
 	{
-	//TODO: Set interface to promisc
+	struct ifreq ifr;
+	struct packet_mreq mreq;
+	int ret;
 
-	return true;
+	memset(&ifr, 0, sizeof(ifr));
+	snprintf(ifr.ifr_name, sizeof(ifr.ifr_name), "%s", props.path.c_str());
+
+	ret = ioctl(socket_fd, SIOCGIFINDEX, &ifr);
+	if ( ret < 0 )
+		return false;
+
+	memset(&mreq, 0, sizeof(mreq));
+	mreq.mr_ifindex = ifr.ifr_ifindex;
+	mreq.mr_type = PACKET_MR_PROMISC;
+
+	ret = setsockopt(socket_fd, SOL_PACKET, PACKET_ADD_MEMBERSHIP, &mreq, sizeof(mreq));
+	return (ret >= 0);
 	}
 
 inline bool AF_PacketSource::ConfigureFanoutGroup(bool enabled)
@@ -122,7 +137,7 @@ inline bool AF_PacketSource::ConfigureFanoutGroup(bool enabled)
 		int ret;
 
 		fanout_id = BifConst::AF_Packet::fanout_id;
-		fanout_arg = (fanout_id | (PACKET_FANOUT_HASH << 16));
+		fanout_arg = ((fanout_id & 0xffff) | (PACKET_FANOUT_HASH << 16));
 
 		ret = setsockopt(socket_fd, SOL_PACKET, PACKET_FANOUT,
 			&fanout_arg, sizeof(fanout_arg));
@@ -181,13 +196,9 @@ bool AF_PacketSource::ExtractNextPacket(Packet* pkt)
 	struct tpacket3_hdr *packet = 0;
 	const u_char *data;
 	struct timeval ts;
-	bool ret;
-
 	while ( true )
 		{
-		ret = rx_ring->GetNextPacket(&packet);
-
-		if ( ! ret )
+		if ( ! rx_ring->GetNextPacket(&packet) )
 			return false;
 
 		current_hdr.ts.tv_sec = packet->tp_sec;
@@ -196,6 +207,13 @@ bool AF_PacketSource::ExtractNextPacket(Packet* pkt)
 		current_hdr.len = packet->tp_len;
 		data = (u_char *) packet + packet->tp_mac;
 
+		if ( !ApplyBPFFilter(current_filter, &current_hdr, data) )
+			{
+			++num_discarded;
+			DoneWithPacket();
+			continue;
+			}
+
 		pkt->Init(props.link_type, &current_hdr.ts, current_hdr.caplen, current_hdr.len, data);
 
 		if ( current_hdr.len == 0 || current_hdr.caplen == 0 )
@@ -204,15 +222,12 @@ bool AF_PacketSource::ExtractNextPacket(Packet* pkt)
 			return false;
 			}
 
-		if ( ApplyBPFFilter(current_filter, &current_hdr, data) )
-			break;
-
-		num_discarded++;
+		stats.received++;
+		stats.bytes_received += current_hdr.len;
+		return true;
 		}
 
-	stats.received++;
-	stats.bytes_received += current_hdr.len;
-	return true;
+	return false;
 	}
 
 void AF_PacketSource::DoneWithPacket()
@@ -240,7 +255,7 @@ void AF_PacketSource::Statistics(Stats* s)
 		}
 
 	struct tpacket_stats_v3 tp_stats;
-	socklen_t tp_stats_len;
+	socklen_t tp_stats_len = sizeof (struct tpacket_stats_v3);
 	int ret;
 
 	ret = getsockopt(socket_fd, SOL_PACKET, PACKET_STATISTICS, &tp_stats, &tp_stats_len);

elasticsearch/README → elasticsearch-deprecated/README

@@ -17,10 +17,11 @@ both for distributed indexing and distributed searching.
 Warning
 -------
 
-This writer plugin is still in testing and is not yet recommended for
-production use!  The approach to how logs are handled in the plugin is "fire
-and forget" at this time, there is no error handling if the server fails to
-respond successfully to the insertion request.
+This writer plugin only supports ElasticSearch 1; it will not work with
+ElasticSearch version 2 and above. This writer plugin is deprecated and
+will be removed from the Bro distribution in the future. This plugin is
+experimental and not recommended for production use; it is for example
+missing error handling and may loose messages.
 
 Installing ElasticSearch
 ------------------------

elasticsearch-deprecated/tests/Scripts/run-nc

@@ -0,0 +1,11 @@
+#! /bin/sh
+
+listen_port=$1
+
+# First try the traditional netcat (this is the default netcat on debian)
+nc -l -p $listen_port 2>/dev/null
+
+if [ $? -ne 0 ]; then
+    # Traditional netcat failed, so try the netcat that everyone else uses
+    nc -l $listen_port
+fi

elasticsearch/tests/btest.cfg → elasticsearch-deprecated/tests/btest.cfg

@@ -14,4 +14,5 @@ PATH=`%(testbase)s/Scripts/get-bro-env path`
 TZ=UTC
 LC_ALL=C
 TRACES=%(testbase)s/Traces
+SCRIPTS=%(testbase)s/Scripts
 TMPDIR=%(testbase)s/.tmp

elasticsearch/tests/elasticsearch/redef-fmt.bro → elasticsearch-deprecated/tests/elasticsearch/es-output.bro

@@ -1,11 +1,9 @@
 # @TEST-REQUIRES: which nc
-# @TEST-EXEC: btest-bg-run nc nc -l 9200
+# @TEST-EXEC: btest-bg-run nc $SCRIPTS/run-nc 9200
 # @TEST-EXEC: bro -r $TRACES/syslog-single-udp.trace %INPUT
 # @TEST-EXEC: btest-bg-wait -k 5
 
 # @TEST-EXEC: btest-diff nc/.stdout
 
 @load Bro/ElasticSearch/logs-to-elasticsearch.bro
 redef LogElasticSearch::send_logs += { Conn::LOG };
-
-redef LogElasticSearch::index_name_fmt = "%Y.%m.%d-%H";

elasticsearch/tests/elasticsearch/nsq-output.bro → elasticsearch-deprecated/tests/elasticsearch/nsq-output.bro

@@ -1,5 +1,5 @@
 # @TEST-REQUIRES: which nc
-# @TEST-EXEC: btest-bg-run nc nc -l 4151
+# @TEST-EXEC: btest-bg-run nc $SCRIPTS/run-nc 4151
 # @TEST-EXEC: bro -r $TRACES/syslog-single-udp.trace %INPUT
 # @TEST-EXEC: btest-bg-wait -k 5
 
@@ -9,4 +9,4 @@
 redef LogElasticSearch::send_logs += { Conn::LOG };
 
 redef LogElasticSearch::destination = "nsq";
-redef LogElasticSearch::server_port = 4151;
+redef LogElasticSearch::server_port = 4151;

elasticsearch/tests/elasticsearch/es-output.bro → elasticsearch-deprecated/tests/elasticsearch/redef-fmt.bro

@@ -1,9 +1,11 @@
 # @TEST-REQUIRES: which nc
-# @TEST-EXEC: btest-bg-run nc nc -l 9200
+# @TEST-EXEC: btest-bg-run nc $SCRIPTS/run-nc 9200
 # @TEST-EXEC: bro -r $TRACES/syslog-single-udp.trace %INPUT
 # @TEST-EXEC: btest-bg-wait -k 5
 
 # @TEST-EXEC: btest-diff nc/.stdout
 
 @load Bro/ElasticSearch/logs-to-elasticsearch.bro
-redef LogElasticSearch::send_logs += { Conn::LOG };
+redef LogElasticSearch::send_logs += { Conn::LOG };
+
+redef LogElasticSearch::index_name_fmt = "%Y.%m.%d-%H";

kafka/.gitignore

@@ -0,0 +1,31 @@
+.state
+build
+
+# Compiled Object files
+*.slo
+*.lo
+*.o
+*.obj
+
+# Precompiled Headers
+*.gch
+*.pch
+
+# Compiled Dynamic libraries
+*.so
+*.dylib
+*.dll
+
+# Fortran module files
+*.mod
+
+# Compiled Static libraries
+*.lai
+*.la
+*.a
+*.lib
+
+# Executables
+*.exe
+*.out
+*.app

kafka/CMakeLists.txt

@@ -0,0 +1,26 @@
+cmake_minimum_required(VERSION 2.8)
+project(Plugin)
+include(BroPlugin)
+find_package(LibRDKafka)
+find_package(OpenSSL)
+
+if (LIBRDKAFKA_FOUND AND OPENSSL_FOUND)
+  include_directories(BEFORE ${LibRDKafka_INCLUDE_DIR} ${OpenSSL_INCLUDE_DIR})
+  bro_plugin_begin(BRO KAFKA)
+  bro_plugin_cc(src/KafkaWriter.cc)
+  bro_plugin_cc(src/Plugin.cc)
+  bro_plugin_cc(src/TaggedJSON.cc)
+  bro_plugin_bif(src/kafka.bif)
+  bro_plugin_dist_files(README CHANGES COPYING VERSION)
+  bro_plugin_link_library(${LibRDKafka_LIBRARIES})
+  bro_plugin_link_library(${LibRDKafka_C_LIBRARIES})
+  bro_plugin_link_library(${OpenSSL_LIBRARIES})
+  bro_plugin_end()
+
+elseif (NOT LIBRDKAFKA_FOUND)
+  message(FATAL_ERROR "LibRDKafka not found.")
+
+elseif (NOT OPENSSL_FOUND)
+  message(FATAL_ERROR "OpenSSL not found.")
+
+endif ()