github-mirrors/logstash
1
0
Fork 0
mirror of https://github.com/elastic/logstash.git synced 2025-04-24 06:37:19 -04:00
Code Issues Projects Releases Packages Wiki Activity

#7091 Dockerized Kafka IT fixture

Browse source 
Fixes #7093
This commit is contained in:
Armin 2017-05-12 20:08:29 +02:00 • committed by Armin Braun
parent 0aea12ce00
commit 2ac72226d3
10 changed files with 83 additions and 98 deletions
Download patch file Download diff file Expand all files Collapse all files

6
.travis.yml
View file

@ -1,4 +1,6 @@
sudo: false
sudo: required
services:
- docker
language: ruby
cache:
directories:
@ -15,7 +17,7 @@ env:
- INTEGRATION=true FEATURE_FLAG=persistent_queues
before_install:
# Force bundler 1.12.5 because version 1.13 has issues, see https://github.com/fastlane/fastlane/issues/6065#issuecomment-246044617
- gem uninstall -i /home/travis/.rvm/gems/jruby-1.7.25@global bundler
- yes | gem uninstall -q -i /home/travis/.rvm/gems/jruby-1.7.25@global bundler
- gem install bundler -v 1.12.5 --no-rdoc --no-ri --no-document --quiet
install:
- rake test:install-core

1
qa/integration/integration_tests.gemspec
View file

@ -21,4 +21,5 @@ Gem::Specification.new do |s|
s.add_development_dependency 'logstash-devutils'
s.add_development_dependency 'flores'
s.add_development_dependency 'rubyzip'
s.add_development_dependency 'docker-api'
end

18
qa/integration/services/kafka_dockerized/Dockerfile Normal file
View file

@ -0,0 +1,18 @@
FROM debian:stretch
ENV KAFKA_HOME /kafka
ENV KAFKA_LOGS_DIR="/kafka-logs"
ENV KAFKA_VERSION 0.10.2.1
ENV _JAVA_OPTIONS "-Djava.net.preferIPv4Stack=true"
ENV TERM=linux
RUN apt-get update && apt-get install -y curl openjdk-8-jdk netcat
RUN mkdir -p ${KAFKA_LOGS_DIR} && mkdir -p ${KAFKA_HOME} && curl -s -o $INSTALL_DIR/kafka.tgz \
"http://ftp.wayne.edu/apache/kafka/${KAFKA_VERSION}/kafka_2.11-${KAFKA_VERSION}.tgz" && \
tar xzf ${INSTALL_DIR}/kafka.tgz -C ${KAFKA_HOME} --strip-components 1
ADD run.sh /run.sh
EXPOSE 9092
EXPOSE 2181

34
qa/integration/services/kafka_dockerized/run.sh Normal file
View file

@ -0,0 +1,34 @@
#!/bin/bash
KAFKA_TOPIC=logstash_topic_plain
wait_for_port() {
count=20
port=$1
while ! nc -z localhost $port && [[ $count -ne 0 ]]; do
count=$(( $count - 1 ))
[[ $count -eq 0 ]] && return 1
sleep 0.5
done
# just in case, one more time
nc -z localhost $port
}
echo "Starting ZooKeeper"
${KAFKA_HOME}/bin/zookeeper-server-start.sh ${KAFKA_HOME}/config/zookeeper.properties &
wait_for_port 2181
echo "Starting Kafka broker"
mkdir -p ${KAFKA_LOGS_DIR}
${KAFKA_HOME}/bin/kafka-server-start.sh ${KAFKA_HOME}/config/server.properties \
--override delete.topic.enable=true --override advertised.host.name=127.0.0.1 \
--override logs.dir=${KAFKA_LOGS_DIR} --override log.flush.interval.ms=200 &
wait_for_port 9092
${KAFKA_HOME}/bin/kafka-topics.sh --create --partitions 1 --replication-factor 1 --topic ${KAFKA_TOPIC} --zookeeper 127.0.0.1:2181
${KAFKA_HOME}/bin/kafka-console-producer.sh --topic ${KAFKA_TOPIC} --broker-list 127.0.0.1:9092 < /how_sample.input
echo "Kafka load status code $?"
tail -f /dev/null

23
qa/integration/services/kafka_service.rb
View file

@ -1,7 +1,30 @@
require_relative "service"
require "docker"
class KafkaService < Service
def initialize(settings)
@kafka_image = Docker::Image.build_from_dir(File.expand_path("../kafka_dockerized", __FILE__))
.insert_local(
'localPath' => File.join(TestSettings::FIXTURES_DIR, "how_sample.input"),
'outputPath' => '/')
super("kafka", settings)
end
def setup
@kafka_container = Docker::Container.create(:Image => @kafka_image.id,
:HostConfig => {
:PortBindings => {
'9092/tcp' => [{ :HostPort => '9092' }],
'2181/tcp' => [{ :HostPort => '2181' }]
}
}, :Cmd => ["/bin/bash", "-l", "/run.sh"])
@kafka_container.start
super()
end
def teardown
@kafka_container.kill(:signal => "SIGHUP")
@kafka_container.delete(:force => true, :volumes => true)
super()
end
end

71
qa/integration/services/kafka_setup.sh
View file

@ -1,71 +0,0 @@
#!/bin/bash
set -ex
current_dir="$(dirname "$0")"
export _JAVA_OPTIONS="-Djava.net.preferIPv4Stack=true"
source "$current_dir/helpers.sh"
if [ -n "${KAFKA_VERSION+1}" ]; then
echo "KAFKA_VERSION is $KAFKA_VERSION"
version=$KAFKA_VERSION
else
version=0.10.2.1
fi
KAFKA_HOME=$INSTALL_DIR/kafka
KAFKA_TOPIC=logstash_topic_plain
KAFKA_MESSAGES=37
KAFKA_LOGS_DIR=/tmp/ls_integration/kafka-logs
setup_kafka() {
local version=$1
if [ ! -d $KAFKA_HOME ]; then
echo "Downloading Kafka version $version"
curl -s -o $INSTALL_DIR/kafka.tgz "http://ftp.wayne.edu/apache/kafka/$version/kafka_2.11-$version.tgz"
mkdir $KAFKA_HOME && tar xzf $INSTALL_DIR/kafka.tgz -C $KAFKA_HOME --strip-components 1
rm $INSTALL_DIR/kafka.tgz
fi
}
start_kafka() {
echo "Starting ZooKeeper"
$KAFKA_HOME/bin/zookeeper-server-start.sh -daemon $KAFKA_HOME/config/zookeeper.properties
wait_for_port 2181
echo "Starting Kafka broker"
mkdir -p ${KAFKA_LOGS_DIR}
$KAFKA_HOME/bin/kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties --override delete.topic.enable=true --override advertised.host.name=127.0.0.1 --override log.dir=${KAFKA_LOGS_DIR} --override log.flush.interval.ms=200
wait_for_port 9092
}
wait_for_messages() {
local count=10
local read_lines=0
echo "Checking if Kafka topic has been populated with data"
while [[ $read_lines -ne $KAFKA_MESSAGES ]] && [[ $count -ne 0 ]]; do
read_lines=`$KAFKA_HOME/bin/kafka-console-consumer.sh --topic $KAFKA_TOPIC --new-consumer --bootstrap-server localhost:9092 --from-beginning --max-messages $KAFKA_MESSAGES --timeout-ms 10000 | wc -l`
count=$(( $count - 1 ))
[[ $count -eq 0 ]] && return 1
sleep 0.5
ls -lrt $KAFKA_LOGS_DIR/$KAFKA_TOPIC-0/
done
echo "Kafka topic has been populated with test data"
}
setup_install_dir
setup_kafka $version
start_kafka
# Set up topics
$KAFKA_HOME/bin/kafka-topics.sh --create --partitions 1 --replication-factor 1 --topic $KAFKA_TOPIC --zookeeper localhost:2181
# check topic got created
num_topic=`$KAFKA_HOME/bin/kafka-topics.sh --list --zookeeper localhost:2181 | grep $KAFKA_TOPIC | wc -l`
[[ $num_topic -eq 1 ]]
# Add test messages to the newly created topic
cp $current_dir/../fixtures/how_sample.input $KAFKA_HOME
[[ ! -s how_sample.input ]]
$KAFKA_HOME/bin/kafka-console-producer.sh --topic $KAFKA_TOPIC --broker-list localhost:9092 < $KAFKA_HOME/how_sample.input
echo "Kafka load status code $?"
# Wait until broker has all messages
wait_for_messages
echo "Kafka Setup complete"

22
qa/integration/services/kafka_teardown.sh
View file

@ -1,22 +0,0 @@
#!/bin/bash
set -ex
current_dir="$(dirname "$0")"
source "$current_dir/helpers.sh"
KAFKA_HOME=$INSTALL_DIR/kafka
stop_kafka() {
echo "Stopping Kafka broker"
$KAFKA_HOME/bin/kafka-server-stop.sh
echo "Stopping zookeeper"
$KAFKA_HOME/bin/zookeeper-server-stop.sh
}
# delete test topic
echo "Deleting test topic in Kafka"
$KAFKA_HOME/bin/kafka-topics.sh --delete --topic logstash_topic_plain --zookeeper localhost:2181 --if-exists
stop_kafka
rm -rf /tmp/ls_integration/kafka-logs
rm -rf /tmp/zookeeper

2
qa/integration/services/service.rb
View file

@ -22,7 +22,7 @@ class Service
def teardown
puts "Tearing down #{@name} service"
if File.exists?(@setup_script)
if File.exists?(@teardown_script)
`#{@teardown_script}`
else
puts "Teardown script not found for #{@name}"

2
qa/integration/specs/kafka_input_spec.rb
View file

@ -25,7 +25,7 @@ describe "Test Kafka Input" do
end
try(num_retries) do
count = File.foreach(@fixture.actual_output).inject(0) {|c, line| c+1}
count = File.foreach(@fixture.actual_output).inject(0) {|c, _| c+1}
expect(count).to eq(num_events)
end
end

2
qa/integration/specs/monitoring_api_spec.rb
View file

@ -2,7 +2,7 @@ require_relative '../framework/fixture'
require_relative '../framework/settings'
require_relative '../services/logstash_service'
require "logstash/devutils/rspec/spec_helper"
require"stud/try"
require "stud/try"
describe "Test Monitoring API" do
before(:all) {