doc: add in/out kafka example with lua filter

Signed-off-by: Thiago Padilha <[email protected]>

Author: tchrono

examples/kafka_filter/.gitignore

@@ -0,0 +1,2 @@
+build/
+kafka/

examples/kafka_filter/Makefile

@@ -0,0 +1,25 @@
+CPU_COUNT := $(shell grep -c ^processor /proc/cpuinfo)
+
+KAFKA_TGT = kafka/done.stamp
+KAFKA_DIR = $(shell dirname $(KAFKA_TGT))
+CMAKE_TGT = build/done.stamp
+FLB_TGT = build/bin/fluent-bit
+
+run: $(KAFKA_TGT) $(FLB_TGT)
+	./kafka-start.sh $(KAFKA_DIR) && tmux a
+
+$(FLB_TGT): $(CMAKE_TGT)
+	make -C build -j$(CPU_COUNT)
+
+$(CMAKE_TGT):
+	rm -rf build
+	mkdir -pv build
+	cd build && cmake -DFLB_DEV=On -DFLB_IN_KAFKA=On -DFLB_OUT_KAFKA=On ../../..
+	touch $@
+
+$(KAFKA_TGT):
+	rm -rf $(KAFKA_DIR)
+	./kafka-setup.sh $(KAFKA_DIR)
+	touch $@
+
+.PHONY: run

examples/kafka_filter/README.md

@@ -0,0 +1,4 @@
+## Instructions
+
+    $ cd examples/kafka_filter
+    $ make

examples/kafka_filter/common.sh

@@ -0,0 +1,6 @@
+set -euo pipefail
+
+die() {
+	echo $* >&2
+	exit 1
+}

examples/kafka_filter/example-data.json

@@ -0,0 +1,5 @@
+{ "name": "kafka", "type": "message bus" }
+{ "name": "fluent-bit", "type": "log processor" }
+{ "name": "kubernetes", "type": "container orchestration system" }
+{ "name": "docker", "type": "container platform" }
+{ "name": "ansible", "type": "software provisioning, configuration management, and application-deployment tool" }

examples/kafka_filter/kafka-setup.sh

@@ -0,0 +1,24 @@
+#!/bin/bash
+
+. common.sh
+
+url="https://dlcdn.apache.org/kafka/3.0.0/kafka_2.13-3.0.0.tgz"
+sha256="a82728166bbccf406009747a25e1fe52dbcb4d575e4a7a8616429b5818cd02d1"
+
+outdir="$1"
+
+[ -z $outdir ] && die "usage: $0 OUTDIR"
+[ -d $outdir ] && die "$outdir already exists"
+
+mkdir -p $outdir
+cd $outdir
+outdir=$(pwd)
+cd -
+rmdir $outdir
+tmpdir=$(mktemp -d)
+cd $tmpdir
+curl -L $url | tee kafka.tgz | sha256sum -c <(echo "$sha256 -")
+tar xf kafka.tgz
+mv kafka_* $outdir
+rm -rf $tmpdir
+find $outdir/config -type f -name '*.properties' | xargs sed -e "s@=/tmp@=$outdir/tmp@" -i 

examples/kafka_filter/kafka-start.sh

@@ -0,0 +1,30 @@
+#!/bin/bash
+
+. common.sh
+
+java --version &> /dev/null || die "please install java 8+"
+
+if tmux has-session -t kafka; then
+	exec tmux attach -t kafka
+fi
+
+kafka_dir="$1"
+[ -z $kafka_dir ] && die "usage: $0 OUTDIR"
+[ ! -d $kafka_dir ] && die "$kafka_dir does not exist"
+[ ! -e $kafka_dir/bin/zookeeper-server-start.sh ] && die "$kafka_dir doesn't contain kafka"
+
+cd $kafka_dir
+tmux new-session -s kafka -n "cli" -d
+tmux new-window -n "services" -t kafka
+tmux split-window -v -t kafka:0
+tmux split-window -v -t kafka:0
+tmux split-window -v -t kafka:1
+tmux send-keys -t kafka:1.0 './bin/zookeeper-server-start.sh config/zookeeper.properties' C-m
+tmux send-keys -t kafka:1.1 'sleep 1 && ./bin/kafka-server-start.sh config/server.properties' C-m
+tmux send-keys -t kafka:0.0 'sleep 3 && ./bin/kafka-topics.sh --create --partitions 1 --replication-factor 1 --topic fb-source --bootstrap-server localhost:9092' C-m
+tmux send-keys -t kafka:0.1 'sleep 3 && ./bin/kafka-topics.sh --create --partitions 1 --replication-factor 1 --topic fb-sink --bootstrap-server localhost:9092' C-m
+tmux send-keys -t kafka:0.0 './bin/kafka-console-consumer.sh --topic fb-sink --from-beginning --bootstrap-server localhost:9092' C-m
+tmux send-keys -t kafka:0.1 'sleep 5' C-m
+tmux send-keys -t kafka:0.1 'cat ../example-data.json | ./bin/kafka-console-producer.sh --topic fb-source --bootstrap-server localhost:9092' C-m
+tmux send-keys -t kafka:0.2 'sleep 4 && ../build/bin/fluent-bit -c ../kafka.conf' C-m
+tmux select-window -t kafka:0.1

examples/kafka_filter/kafka.conf

@@ -0,0 +1,20 @@
+[SERVICE]
+    Flush      0.1
+    Grace      2
+    Log_Level  info
+
+[INPUT]
+    Name kafka
+    brokers localhost:9092
+    topics fb-source
+
+[FILTER]
+    Name    lua
+    Match   *
+    script  kafka.lua
+    call    modify_kafka_message
+
+[OUTPUT]
+    Name kafka
+    brokers localhost:9092
+    topics fb-sink

examples/kafka_filter/kafka.lua

@@ -0,0 +1,7 @@
+local count = 0
+function modify_kafka_message(tag, timestamp, record)
+    count = count + 1
+    local payload = record.payload
+    payload.status = 'processed '..payload.name..', total records: '..tostring(count)
+    return 1, timestamp, payload
+end