mirror of
https://github.com/rsyslog/rsyslog.git
synced 2025-12-16 12:10:46 +01:00
testbench: improve imkafka tests and testbench framework
This commit is contained in:
Rainer Gerhards
parent
877ac178a7
commit
60bbe1d14c
@ -316,6 +316,23 @@ function kafka_check_broken_broker() {
|
||||
fi
|
||||
}
|
||||
|
||||
# inject messages via kafkacat tool (for imkafka tests)
|
||||
# $1 == "--wait" means wait for rsyslog to receive TESTMESSAGES lines in RSYSLOG_OUT_LOG
|
||||
# $TESTMESSAGES contains number of messages to inject
|
||||
# $RANDTOPIC contains topic to produce to
|
||||
function injectmsg_kafkacat() {
|
||||
if [ "$TESTMESSAGES" == "" ]; then
|
||||
printf 'TESTBENCH ERROR: TESTMESSAGES env var not set!\n'
|
||||
error_exit 1
|
||||
fi
|
||||
for ((i=1 ; i<=TESTMESSAGES ; i++)); do
|
||||
printf ' msgnum:%8.8d\n' $i; \
|
||||
done | kafkacat -P -b localhost:29092 -t $RANDTOPIC
|
||||
if [ "$1" == "--wait" ]; then
|
||||
$srcdir/diag.sh wait-file-lines $RSYSLOG_OUT_LOG $TESTMESSAGES ${RETRIES:-200}
|
||||
fi
|
||||
}
|
||||
|
||||
# wait for rsyslogd startup ($1 is the instance)
|
||||
function wait_startup() {
|
||||
wait_rsyslog_startup_pid $1
|
||||
@ -1271,6 +1288,8 @@ function dump_kafka_serverlog() {
|
||||
echo "========================================="
|
||||
cat $dep_work_dir/kafka/logs/server.log
|
||||
echo "========================================="
|
||||
printf 'non-info is:\n'
|
||||
grep --invert-match '^\[.* INFO ' $dep_work_dir/kafka/logs/server.log | grep '^\['
|
||||
fi
|
||||
}
|
||||
|
||||
@ -1284,6 +1303,8 @@ function dump_zookeeper_serverlog() {
|
||||
echo "========================================="
|
||||
cat $dep_work_dir/zk/zookeeper.out
|
||||
echo "========================================="
|
||||
printf 'non-info is:\n'
|
||||
grep --invert-match '^\[.* INFO ' $dep_work_dir/zk/zookeeper.out | grep '^\['
|
||||
}
|
||||
|
||||
|
||||
@ -1500,11 +1521,7 @@ case $1 in
|
||||
;;
|
||||
'wait-file-lines')
|
||||
# $2 filename, $3 expected nbr of lines, $4 nbr of tries
|
||||
if [ "$4" == "" ]; then
|
||||
timeoutend=1
|
||||
else
|
||||
timeoutend=$4
|
||||
fi
|
||||
timeoutend=${4:-1}
|
||||
timecounter=0
|
||||
|
||||
while [ $timecounter -lt $timeoutend ]; do
|
||||
@ -1518,7 +1535,7 @@ case $1 in
|
||||
break
|
||||
else
|
||||
if [ "x$timecounter" == "x$timeoutend" ]; then
|
||||
echo wait-file-lines failed, expected $3 got $count
|
||||
echo wait-file-lines failed, expected $3 got $count after $timeoutend retries
|
||||
shutdown_when_empty
|
||||
wait_shutdown
|
||||
error_exit 1
|
||||
|
||||
@ -7,8 +7,8 @@ check_command_available kafkacat
|
||||
export TESTMESSAGES=100000
|
||||
export RANDTOPIC=$(tr -dc 'a-zA-Z0-9' < /dev/urandom | fold -w 8 | head -n 1)
|
||||
# Set EXTRA_EXITCHECK to dump kafka/zookeeperlogfiles on failure only.
|
||||
#export EXTRA_EXITCHECK=dumpkafkalogs
|
||||
#export EXTRA_EXIT=kafka
|
||||
export EXTRA_EXITCHECK=dumpkafkalogs
|
||||
export EXTRA_EXIT=kafka
|
||||
|
||||
echo Check and Stop previous instances of kafka/zookeeper
|
||||
download_kafka
|
||||
@ -19,10 +19,8 @@ echo Create kafka/zookeeper instance and $RANDTOPIC topic
|
||||
start_zookeeper
|
||||
start_kafka
|
||||
|
||||
# create new topic
|
||||
create_kafka_topic $RANDTOPIC '.dep_wrk' '22181'
|
||||
|
||||
# --- Create imkafka receiver config
|
||||
export RSYSLOG_DEBUGLOG="log"
|
||||
generate_conf
|
||||
add_conf '
|
||||
@ -51,17 +49,7 @@ if ($msg contains "msgnum:") then {
|
||||
export RSTB_DAEMONIZE="YES"
|
||||
startup
|
||||
|
||||
for i in {00000001..00100000}; do
|
||||
echo " msgnum:$i" >> $RSYSLOG_DYNNAME.in
|
||||
done
|
||||
|
||||
echo Inject messages into kafka
|
||||
kafkacat -P -b localhost:29092 -t $RANDTOPIC < $RSYSLOG_DYNNAME.in
|
||||
|
||||
echo Give imkafka some time to start...
|
||||
sleep 5
|
||||
|
||||
echo Stopping sender instance [omkafka]
|
||||
injectmsg_kafkacat --wait
|
||||
shutdown_when_empty
|
||||
wait_shutdown
|
||||
|
||||
|
||||
@ -5,22 +5,18 @@
|
||||
check_command_available kafkacat
|
||||
|
||||
export TESTMESSAGES=1000
|
||||
|
||||
export RANDTOPIC=$(tr -dc 'a-zA-Z0-9' < /dev/urandom | fold -w 8 | head -n 1)
|
||||
|
||||
# Set EXTRA_EXITCHECK to dump kafka/zookeeperlogfiles on failure only.
|
||||
export EXTRA_EXITCHECK=dumpkafkalogs
|
||||
export EXTRA_EXIT=kafka
|
||||
|
||||
export RANDTOPIC=$(tr -dc 'a-zA-Z0-9' < /dev/urandom | fold -w 8 | head -n 1)
|
||||
|
||||
download_kafka
|
||||
stop_zookeeper
|
||||
stop_kafka
|
||||
|
||||
echo Create kafka/zookeeper instance and $RANDTOPIC topic
|
||||
start_zookeeper
|
||||
start_kafka
|
||||
|
||||
# --- Create imkafka receiver config
|
||||
export RSYSLOG_DEBUGLOG="log"
|
||||
generate_conf
|
||||
add_conf '
|
||||
@ -49,9 +45,7 @@ startup
|
||||
|
||||
# We inject messages, even though we know this will not work. The reason
|
||||
# is that we want to ensure we do not get a segfault in such an error case
|
||||
for ((i=1 ; i<=TESTMESSAGES ; i++)); do
|
||||
printf ' msgnum:%8.8d\n' $i; \
|
||||
done | kafkacat -P -b localhost:29092 -t $RANDTOPIC
|
||||
injectmsg_kafkacat
|
||||
|
||||
shutdown_when_empty
|
||||
wait_shutdown
|
||||
|
||||
@ -16,7 +16,6 @@ stop_kafka
|
||||
start_zookeeper
|
||||
start_kafka
|
||||
|
||||
# --- Create imkafka receiver config
|
||||
generate_conf
|
||||
add_conf '
|
||||
main_queue(queue.timeoutactioncompletion="60000" queue.timeoutshutdown="60000")
|
||||
@ -44,12 +43,7 @@ if ($msg contains "msgnum:") then {
|
||||
'
|
||||
startup
|
||||
|
||||
for ((i=1 ; i<=TESTMESSAGES ; i++)); do
|
||||
printf ' msgnum:%8.8d\n' $i; \
|
||||
done | kafkacat -P -b localhost:29092 -t $RANDTOPIC
|
||||
rst_msleep 1000 # give kafkacat a chance to startup (we need to wait anyhow)
|
||||
$srcdir/diag.sh wait-file-lines $RSYSLOG_OUT_LOG $TESTMESSAGES ${RETRIES:-100}
|
||||
|
||||
injectmsg_kafkacat --wait
|
||||
shutdown_when_empty
|
||||
wait_shutdown
|
||||
|
||||
|
||||
@ -5,30 +5,22 @@ echo Init Testbench
|
||||
. ${srcdir:=.}/diag.sh init
|
||||
check_command_available kafkacat
|
||||
|
||||
# *** ==============================================================================
|
||||
export TESTMESSAGES=10000
|
||||
export TESTMESSAGES=100000
|
||||
export TESTMESSAGESFULL=$TESTMESSAGES
|
||||
|
||||
# Generate random topic name
|
||||
export RANDTOPIC=$(tr -dc 'a-zA-Z0-9' < /dev/urandom | fold -w 8 | head -n 1)
|
||||
|
||||
# Set EXTRA_EXITCHECK to dump kafka/zookeeperlogfiles on failure only.
|
||||
export EXTRA_EXITCHECK=dumpkafkalogs
|
||||
export EXTRA_EXIT=kafka
|
||||
echo ===============================================================================
|
||||
echo Check and Stop previous instances of kafka/zookeeper
|
||||
|
||||
export RANDTOPIC=$(tr -dc 'a-zA-Z0-9' < /dev/urandom | fold -w 8 | head -n 1)
|
||||
|
||||
download_kafka
|
||||
stop_zookeeper
|
||||
stop_kafka
|
||||
|
||||
echo Create kafka/zookeeper instance and $RANDTOPIC topic
|
||||
start_zookeeper
|
||||
start_kafka
|
||||
# create new topic
|
||||
create_kafka_topic $RANDTOPIC '.dep_wrk' '22181'
|
||||
|
||||
# --- Create imkafka receiver config
|
||||
export RSYSLOG_DEBUGLOG="log"
|
||||
generate_conf
|
||||
add_conf '
|
||||
main_queue(queue.timeoutactioncompletion="60000" queue.timeoutshutdown="60000")
|
||||
@ -56,34 +48,15 @@ if ($msg contains "msgnum:") then {
|
||||
# ---
|
||||
|
||||
# --- Start imkafka receiver config
|
||||
echo Starting receiver instance [imkafka]
|
||||
startup_vg
|
||||
# ---
|
||||
|
||||
# --- Fill Kafka Server with messages
|
||||
# Can properly be done in a better way?!
|
||||
for i in {00000001..0010000}
|
||||
do
|
||||
echo " msgnum:$i" >> $RSYSLOG_OUT_LOG.in
|
||||
done
|
||||
|
||||
echo Inject messages into kafka
|
||||
kafkacat <$RSYSLOG_OUT_LOG.in -P -b localhost:29092 -t $RANDTOPIC
|
||||
# ---
|
||||
|
||||
echo Give imkafka some time to start...
|
||||
sleep 5
|
||||
|
||||
echo Stopping sender instance [omkafka]
|
||||
injectmsg_kafkacat --wait
|
||||
shutdown_when_empty
|
||||
wait_shutdown_vg
|
||||
check_exit_vg
|
||||
|
||||
# Delete topic to remove old traces before
|
||||
delete_kafka_topic $RANDTOPIC '.dep_wrk' '22181'
|
||||
|
||||
# Do the final sequence check
|
||||
seq_check 1 $TESTMESSAGESFULL -d
|
||||
|
||||
echo success
|
||||
exit_test
|
||||
|
||||
@ -5,31 +5,22 @@ echo Init Testbench
|
||||
. ${srcdir:=.}/diag.sh init
|
||||
check_command_available kafkacat
|
||||
|
||||
# *** ==============================================================================
|
||||
export TESTMESSAGES=100000
|
||||
export TESTMESSAGESFULL=$TESTMESSAGES
|
||||
|
||||
# Generate random topic name
|
||||
export RANDTOPIC=$(tr -dc 'a-zA-Z0-9' < /dev/urandom | fold -w 8 | head -n 1)
|
||||
|
||||
# Set EXTRA_EXITCHECK to dump kafka/zookeeperlogfiles on failure only.
|
||||
export EXTRA_EXITCHECK=dumpkafkalogs
|
||||
export EXTRA_EXIT=kafka
|
||||
echo ===============================================================================
|
||||
echo Check and Stop previous instances of kafka/zookeeper
|
||||
|
||||
export RANDTOPIC=$(tr -dc 'a-zA-Z0-9' < /dev/urandom | fold -w 8 | head -n 1)
|
||||
|
||||
download_kafka
|
||||
stop_zookeeper
|
||||
stop_kafka
|
||||
|
||||
echo Create kafka/zookeeper instance and $RANDTOPIC topic
|
||||
start_zookeeper
|
||||
start_kafka
|
||||
|
||||
# create new topic
|
||||
create_kafka_topic $RANDTOPIC '.dep_wrk' '22181'
|
||||
|
||||
# --- Create imkafka receiver config
|
||||
export RSYSLOG_DEBUGLOG="log"
|
||||
generate_conf
|
||||
add_conf '
|
||||
main_queue(queue.timeoutactioncompletion="60000" queue.timeoutshutdown="60000")
|
||||
@ -57,33 +48,14 @@ if ($msg contains "msgnum:") then {
|
||||
# ---
|
||||
|
||||
# --- Start imkafka receiver config
|
||||
echo Starting receiver instance [imkafka]
|
||||
startup
|
||||
# ---
|
||||
|
||||
# --- Fill Kafka Server with messages
|
||||
# Can properly be done in a better way?!
|
||||
for i in {00000001..00100000}
|
||||
do
|
||||
echo " msgnum:$i" >> $RSYSLOG_OUT_LOG.in
|
||||
done
|
||||
|
||||
echo Inject messages into kafka
|
||||
kafkacat -P -b localhost:29092 -t $RANDTOPIC < $RSYSLOG_OUT_LOG.in
|
||||
# ---
|
||||
|
||||
echo Give imkafka some time to start...
|
||||
sleep 5
|
||||
|
||||
echo Stopping sender instance [omkafka]
|
||||
injectmsg_kafkacat --wait
|
||||
shutdown_when_empty
|
||||
wait_shutdown
|
||||
|
||||
# Delete topic to remove old traces before
|
||||
delete_kafka_topic $RANDTOPIC '.dep_wrk' '22181'
|
||||
|
||||
# Do the final sequence check
|
||||
seq_check 1 $TESTMESSAGESFULL -d
|
||||
|
||||
echo success
|
||||
exit_test
|
||||
|
||||
@ -1,34 +1,24 @@
|
||||
#!/bin/bash
|
||||
# added 2018-08-29 by alorbach
|
||||
# This file is part of the rsyslog project, released under ASL 2.0
|
||||
echo Init Testbench
|
||||
. ${srcdir:=.}/diag.sh init
|
||||
check_command_available kafkacat
|
||||
|
||||
# *** ==============================================================================
|
||||
export TESTMESSAGES=100000
|
||||
export TESTMESSAGESFULL=100000
|
||||
|
||||
# Generate random topic name
|
||||
export RANDTOPIC=$(tr -dc 'a-zA-Z0-9' < /dev/urandom | fold -w 8 | head -n 1)
|
||||
|
||||
# Set EXTRA_EXITCHECK to dump kafka/zookeeperlogfiles on failure only.
|
||||
export EXTRA_EXITCHECK=dumpkafkalogs
|
||||
export EXTRA_EXIT=kafka
|
||||
echo ===============================================================================
|
||||
echo Check and Stop previous instances of kafka/zookeeper
|
||||
#export EXTRA_EXITCHECK=dumpkafkalogs
|
||||
#export EXTRA_EXIT=kafka
|
||||
|
||||
download_kafka
|
||||
stop_zookeeper
|
||||
stop_kafka
|
||||
|
||||
echo Create kafka/zookeeper instance and $RANDTOPIC topic
|
||||
start_zookeeper
|
||||
start_kafka
|
||||
|
||||
# create new topic
|
||||
export RANDTOPIC=$(tr -dc 'a-zA-Z0-9' < /dev/urandom | fold -w 8 | head -n 1)
|
||||
|
||||
create_kafka_topic $RANDTOPIC '.dep_wrk' '22181'
|
||||
|
||||
# --- Create imkafka receiver config
|
||||
export RSYSLOG_DEBUGLOG="log"
|
||||
generate_conf
|
||||
add_conf '
|
||||
@ -139,42 +129,22 @@ if ($msg contains "msgnum:") then {
|
||||
}
|
||||
'
|
||||
|
||||
# Start imkafka receiver config
|
||||
echo Starting receiver instance [imkafka]
|
||||
startup
|
||||
# ---
|
||||
|
||||
# Messure Starttime
|
||||
TIMESTART=$(date +%s.%N)
|
||||
|
||||
# --- Fill Kafka Server with messages
|
||||
# Can properly be done in a better way?!
|
||||
for i in {00000001..00100000}
|
||||
do
|
||||
echo " msgnum:$i" >> $RSYSLOG_OUT_LOG.in
|
||||
done
|
||||
|
||||
echo Inject messages into kafka
|
||||
kafkacat <$RSYSLOG_OUT_LOG.in -P -b localhost:29092 -t $RANDTOPIC
|
||||
# ---
|
||||
|
||||
echo Give imkafka some time to start...
|
||||
sleep 5
|
||||
|
||||
echo Stopping sender instance [omkafka]
|
||||
injectmsg_kafkacat
|
||||
# special case: number of test messages differs from file output
|
||||
$srcdir/diag.sh wait-file-lines $RSYSLOG_OUT_LOG $((TESTMESSAGES * 8)) ${RETRIES:-200}
|
||||
shutdown_when_empty
|
||||
wait_shutdown
|
||||
|
||||
# Messure Endtime
|
||||
TIMEEND=$(date +%s.%N)
|
||||
TIMEDIFF=$(echo "$TIMEEND - $TIMESTART" | bc)
|
||||
echo "*** imkafka time to process all data: $TIMEDIFF seconds!"
|
||||
|
||||
# Delete topic to remove old traces before
|
||||
delete_kafka_topic $RANDTOPIC '.dep_wrk' '22181'
|
||||
|
||||
# Do the final sequence check
|
||||
seq_check 1 $TESTMESSAGESFULL -d
|
||||
seq_check 1 $TESTMESSAGES -d
|
||||
|
||||
echo success
|
||||
exit_test
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user