Merge branch 'kafka_integration_test' of https://github.com/janmejay/rsyslog into tmp

.travis.yml

@@ -38,6 +38,8 @@ addons:
     - flex
     - libcurl4-gnutls-dev
     - python-docutils 
+    - openjdk-7-jdk
+    - wget
     #- libgrok1
     #- libgrok-dev
     #- faketime
@@ -64,7 +66,7 @@ matrix:
        # functions (C99 semantics are really ugly...)
        dist: trusty
      - compiler: "gcc"
-       env: BUILD_FROM_TARBALL="YES", GROK="YES", CHECK="YES", CFLAGS="-g -O2 -W -Wall -Wformat-security -Wshadow -Wcast-align -Wpointer-arith -Wmissing-format-attribute", RS_TESTBENCH_VALGRIND_EXTRA_OPTS="--suppressions=travis/trusty.supp --gen-suppressions=all"
+       env: BUILD_FROM_TARBALL="YES", GROK="YES", KAFKA="YES", CHECK="YES", CFLAGS="-g -O2 -W -Wall -Wformat-security -Wshadow -Wcast-align -Wpointer-arith -Wmissing-format-attribute", RS_TESTBENCH_VALGRIND_EXTRA_OPTS="--suppressions=travis/trusty.supp --gen-suppressions=all"
        dist: trusty
      - compiler: "clang"
        env: CHECK="YES", CFLAGS="-g -O1 -fsanitize=address -fno-color-diagnostics"

configure.ac

@@ -1475,7 +1475,30 @@ AC_ARG_ENABLE(omkafka,
          esac],
         [enable_omkafka=no]
 )
+AC_ARG_ENABLE(kafka_tests,
+        [AS_HELP_STRING([--enable-kafka-tests],[Enable Kafka tests, needs Java @<:@default=no@:>@])],
+        [case "${enableval}" in
+         yes) enable_kafka_tests="yes" ;;
+          no) enable_kafka_tests="no" ;;
+           *) AC_MSG_ERROR(bad value ${enableval} for --enable-kafka-tests) ;;
+         esac],
+        [enable_kafka_tests=no]
+)
+AM_CONDITIONAL(ENABLE_KAFKA_TESTS, test x$enable_kafka_tests = xyes)
+
 if test "x$enable_omkafka" = "xyes"; then
+	if test "x$enable_kafka_tests" = "xyes"; then
+		AX_PROG_JAVAC #we don't need javac, but macro documentation says JAVAC *must* be checked before JAVA
+		AX_PROG_JAVA
+		AC_CHECK_PROG(WGET, [wget], [yes], [no])
+		if test "x${WGET}" = "xno"; then
+			AC_MSG_FAILURE([wget, which is a kafka-tests dependency, not found])
+		fi
+		AC_CHECK_PROG(READLINK, [readlink], [yes], [no])
+		if test "x${READLINK}" = "xno"; then
+			AC_MSG_FAILURE([readlink, which is a kafka-tests dependency, not found])
+		fi
+	fi
 	PKG_CHECK_MODULES([LIBRDKAFKA], [librdkafka],, [
 		AC_CHECK_LIB([rdkafka], [rd_kafka_produce], [
 			AC_MSG_WARN([librdkafka is missing but library present, using -lrdkafka])
@@ -1485,10 +1508,13 @@ if test "x$enable_omkafka" = "xyes"; then
 		])
 	])
 	AC_CHECK_HEADERS([librdkafka/rdkafka.h])
+else
+	if test "x$enable_kafka_tests" = "xyes"; then
+		AC_MSG_ERROR([kafka-tests can not be enabled while omkafka is disabled])
+	fi
 fi
 AM_CONDITIONAL(ENABLE_OMKAFKA, test x$enable_omkafka = xyes)
 
-
 #MONGODB SUPPORT
 
 AC_ARG_ENABLE(ommongodb,
@@ -1893,6 +1919,7 @@ echo "---{ debugging support }---"
 echo "    Testbench enabled:                        $enable_testbench"
 echo "    Extended Testbench enabled:               $enable_extended_tests"
 echo "    MySQL Tests enabled:                      $enable_mysql_tests"
+echo "    Kafka Tests enabled:                      $enable_kafka_tests"
 echo "    Debug mode enabled:                       $enable_debug"
 echo "    Runtime Instrumentation enabled:          $enable_rtinst"
 echo "    (total) debugless mode enabled:           $enable_debugless"

tests/.gitignore

@@ -2,3 +2,5 @@ rscript
 rt_init
 tmp
 mangle_qi
+.dep_cache
+.dep_wrk

tests/Makefile.am

@@ -243,6 +243,13 @@ TESTS +=  \
 	omjournal-basic-no-template.sh
 endif
 
+if ENABLE_OMKAFKA
+if ENABLE_KAFKA_TESTS
+TESTS += \
+	omkafka_static.sh
+endif
+endif
+
 if ENABLE_MYSQL_TESTS
 TESTS +=  \
 	mysql-basic.sh \
@@ -1035,6 +1042,10 @@ EXTRA_DIST= \
 	mysql-actq-mt-withpause.sh \
 	mysql-actq-mt-withpause-vg.sh \
 	testsuites/mysql-actq-mt.conf \
+	testsuites/kafka-server.properties \
+	testsuites/zoo.cfg \
+	omkafka_static.sh \
+	testsuites/omkafka_static.conf \
 	mmpstrucdata.sh \
 	mmpstrucdata-vg.sh \
 	testsuites/mmpstrucdata.conf \

tests/diag.sh

@@ -37,6 +37,24 @@
 #export RSYSLOG_DEBUG="debug nologfuncflow noprintmutexaction nostdout"
 #export RSYSLOG_DEBUGLOG="log"
 TB_TIMEOUT_STARTSTOP=3000 # timeout for start/stop rsyslogd in tenths (!) of a second 3000 => 5 min
+
+#START: ext dependency config
+dep_zk_url=http://www-us.apache.org/dist/zookeeper/zookeeper-3.4.8/zookeeper-3.4.8.tar.gz
+dep_kafka_url=http://www-us.apache.org/dist/kafka/0.9.0.1/kafka_2.11-0.9.0.1.tgz
+dep_cache_dir=$(readlink -f $srcdir/.dep_cache)
+dep_work_dir=$(readlink -f $srcdir/.dep_wrk)
+dep_zk_cached_file=$dep_cache_dir/zookeeper-3.4.8.tar.gz
+dep_kafka_cached_file=$dep_cache_dir/kafka_2.11-0.9.0.1.tgz
+
+dep_kafka_work_dir=$dep_work_dir/kafka
+dep_kafka_dir_xform_pattern='s#^[^/]\+#kafka#g'
+
+dep_zk_work_dir=$dep_work_dir/zk
+dep_zk_dir_xform_pattern='s#^[^/]\+#zk#g'
+
+dep_kafka_log_dump=$(readlink -f $srcdir/rsyslog.out.kafka.log)
+#END: ext dependency config
+
 case $1 in
    'init')	$srcdir/killrsyslog.sh # kill rsyslogd if it runs for some reason
 		if [ -z $RS_SORTCMD ]; then
@@ -441,6 +459,88 @@ case $1 in
 		    exit 77
 		fi
 		;;
+	 'download-kafka')
+		if [ ! -d $dep_cache_dir ]; then
+				echo "Creating dependency cache dir"
+				mkdir $dep_cache_dir
+		fi
+		if [ ! -f $dep_zk_cached_file ]; then
+				echo "Downloading zookeeper"
+				wget $dep_zk_url -O $dep_zk_cached_file
+		fi
+		if [ ! -f $dep_kafka_cached_file ]; then
+				echo "Downloading kafka"
+				wget $dep_kafka_url -O $dep_kafka_cached_file
+		fi
+		;;
+	 'start-kafka')
+		if [ ! -f $dep_zk_cached_file ]; then
+				echo "Dependency-cache does not have zookeeper package, did you download dependencies?"
+				exit 1
+		fi
+		if [ ! -f $dep_kafka_cached_file ]; then
+				echo "Dependency-cache does not have kafka package, did you download dependencies?"
+				exit 1
+		fi
+		if [ ! -d $dep_work_dir ]; then
+				echo "Creating dependency working directory"
+				mkdir -p $dep_work_dir
+		fi
+		if [ -d $dep_kafka_work_dir ]; then
+				(cd $dep_kafka_work_dir && ./bin/kafka-server-stop.sh)
+				./msleep 4000
+		fi
+		if [ -d $dep_zk_work_dir ]; then
+				(cd $dep_zk_work_dir && ./bin/zkServer.sh stop)
+				./msleep 2000
+		fi
+		rm -rf $dep_kafka_work_dir
+		rm -rf $dep_zk_work_dir
+		(cd $dep_work_dir && tar -zxvf $dep_zk_cached_file --xform $dep_zk_dir_xform_pattern --show-transformed-names)
+		(cd $dep_work_dir && tar -zxvf $dep_kafka_cached_file --xform $dep_kafka_dir_xform_pattern --show-transformed-names)
+		cp $srcdir/testsuites/zoo.cfg $dep_zk_work_dir/conf/
+		(cd $dep_zk_work_dir && ./bin/zkServer.sh start)
+		./msleep 4000
+		cp $srcdir/testsuites/kafka-server.properties $dep_kafka_work_dir/config/
+		(cd $dep_kafka_work_dir && ./bin/kafka-server-start.sh -daemon ./config/kafka-server.properties)
+		./msleep 8000
+		;;
+	 'stop-kafka')
+		if [ ! -d $dep_kafka_work_dir ]; then
+				echo "Kafka work-dir does not exist, did you start kafka?"
+				exit 1
+		fi
+		(cd $dep_kafka_work_dir && ./bin/kafka-server-stop.sh)
+		./msleep 4000
+		(cd $dep_zk_work_dir && ./bin/zkServer.sh stop)
+		./msleep 2000
+		rm -rf $dep_kafka_work_dir
+		rm -rf $dep_zk_work_dir
+		;;
+	 'create-kafka-topic')
+		if [ ! -d $dep_kafka_work_dir ]; then
+				echo "Kafka work-dir does not exist, did you start kafka?"
+				exit 1
+		fi
+		if [ "x$2" == "x" ]; then
+				echo "Topic-name not provided."
+				exit 1
+		fi
+		(cd $dep_kafka_work_dir && ./bin/kafka-topics.sh --create --zookeeper localhost:2181/kafka --topic $2 --partitions 2 --replication-factor 1)
+		;;
+	 'dump-kafka-topic')
+		echo "dumping kafka-topic $2"
+		if [ ! -d $dep_kafka_work_dir ]; then
+				echo "Kafka work-dir does not exist, did you start kafka?"
+				exit 1
+		fi
+		if [ "x$2" == "x" ]; then
+				echo "Topic-name not provided."
+				exit 1
+		fi
+
+		(cd $dep_kafka_work_dir && ./bin/kafka-console-consumer.sh --timeout-ms 2000 --from-beginning --zookeeper localhost:2181/kafka --topic $2 > $dep_kafka_log_dump)
+		;;
    'error-exit') # this is called if we had an error and need to abort. Here, we
                 # try to gather as much information as possible. That's most important
 		# for systems like Travis-CI where we cannot debug on the machine itself.

tests/omkafka_static.sh

@@ -0,0 +1,25 @@
+#!/bin/bash
+# added 2016-04-19 by singh.janmejay
+# This file is part of the rsyslog project, released under ASL 2.0
+echo ===============================================================================
+echo \[omkafka_static.sh\]: test for omkafka publishing to statically defined topic
+. $srcdir/diag.sh init
+. $srcdir/diag.sh download-kafka
+. $srcdir/diag.sh start-kafka
+. $srcdir/diag.sh create-kafka-topic 'static'
+. $srcdir/diag.sh startup omkafka_static.conf
+. $srcdir/diag.sh wait-for-stats-flush 'rsyslog.out.stats.log'
+. $srcdir/diag.sh injectmsg  0 10
+. $srcdir/diag.sh wait-queueempty
+. $srcdir/diag.sh wait-for-stats-flush 'rsyslog.out.stats.log'
+echo doing shutdown
+. $srcdir/diag.sh shutdown-when-empty
+echo wait on shutdown
+. $srcdir/diag.sh wait-shutdown
+. $srcdir/diag.sh dump-kafka-topic 'static'
+. $srcdir/diag.sh stop-kafka
+. $srcdir/diag.sh custom-content-check 'msgnum:00000000' 'rsyslog.out.kafka.log'
+. $srcdir/diag.sh custom-content-check 'msgnum:00000001' 'rsyslog.out.kafka.log'
+. $srcdir/diag.sh custom-content-check 'msgnum:00000009' 'rsyslog.out.kafka.log'
+. $srcdir/diag.sh first-column-sum-check 's/.*submitted=\([0-9]\+\)/\1/g' 'omkafka' 'rsyslog.out.stats.log' 10
+. $srcdir/diag.sh exit

tests/testsuites/kafka-server.properties

@@ -0,0 +1,26 @@
+broker.id=0
+listeners=PLAINTEXT://:9092
+
+num.network.threads=1
+
+num.io.threads=1
+
+socket.send.buffer.bytes=102400
+socket.receive.buffer.bytes=102400
+socket.request.max.bytes=104857600
+
+log.dirs=kafka-logs
+num.partitions=1
+num.recovery.threads.per.data.dir=1
+
+log.retention.hours=1
+
+#10 MB
+log.segment.bytes=10485760
+
+log.retention.check.interval.ms=300000
+
+zookeeper.connect=localhost:2181/kafka
+
+zookeeper.connection.timeout.ms=6000
+

tests/testsuites/omkafka_static.conf

@@ -0,0 +1,15 @@
+$IncludeConfig diag-common.conf
+
+module(load="../plugins/omkafka/.libs/omkafka")
+
+ruleset(name="stats") {
+  action(type="omfile" file="./rsyslog.out.stats.log")
+}
+
+module(load="../plugins/impstats/.libs/impstats" interval="1" severity="7" resetCounters="on" Ruleset="stats" bracketing="on")
+
+template(name="outfmt" type="string" string="%msg%\n")
+
+if ($msg contains "msgnum") then {
+	 action(name="kafka-fwd" type="omkafka" topic="static" broker="localhost" template="outfmt" partitions.auto="on")
+}

tests/testsuites/zoo.cfg

@@ -0,0 +1,5 @@
+tickTime=2000
+initLimit=10
+syncLimit=5
+dataDir=zk_data_dir
+clientPort=2181

tests/travis/install.sh

@@ -12,4 +12,4 @@ if [ "$CC" == "clang" ] && [ "$DISTRIB_CODENAME" == "trusty" ]; then CLANG_PKG="
 if [ "$CC" == "clang" ]; then export NO_VALGRIND="--without-valgrind-testbench"; fi
 if [ "$CC" == "clang" ]; then sudo apt-get install -qq $CLANG_PKG ; fi
 if [ "x$KAFKA" == "xYES" ]; then sudo apt-get install -qq librdkafka-dev ; fi
-if [ "x$KAFKA" == "xYES" ]; then export ENABLE_KAFKA="--enable-omkafka" ; fi
+if [ "x$KAFKA" == "xYES" ]; then export ENABLE_KAFKA="--enable-omkafka --enable-kafka-tests" ; fi