Something went wrong. Try again.
Mirrored from GitHub github.com/roostorg/osprey
Something went wrong. Try again.
3.7 kB · 119 lines
Shell
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120#!/bin/bash
# Configuration variablesKAFKA_BROKER=${KAFKA_BROKER:-"localhost:9092"}KAFKA_TOPIC=${KAFKA_TOPIC:-"test-data"}KAFKA_CONFIG_FILE=${KAFKA_CONFIG_FILE:-""}
# Function to generate random user IDgenerate_random_user_id() { echo "user_$(shuf -i 100-9999 -n 1)"}
# Function to generate current timestamp in ISO 8601 format with nanosecondsgenerate_timestamp() { date -u +"%Y-%m-%dT%H:%M:%S.%NZ"}
# Initialize action_id counteraction_id=1
# words in postwords=(hello the quick brown fox jumps over lazy dog and cat runs fast);SCRIPT_DIR=$( cd -- "$( dirname -- "${BASH_SOURCE[0]}" )" &> /dev/null && pwd )
# Function to generate a single JSON actiongenerate_action() { local text="${words[RANDOM % ${#words[@]}]} ${words[RANDOM % ${#words[@]}]} ${words[RANDOM % ${#words[@]}]} ${words[RANDOM % ${#words[@]}]} ${words[RANDOM % ${#words[@]}]}." local timestamp=$(generate_timestamp) local user_id=$(generate_random_user_id) local ip_address="192.168.1.$(shuf -i 1-254 -n 1)"
local sed_commands=() sed_commands+=("s/\$text/$text/g") sed_commands+=("s/\$timestamp/$timestamp/g") sed_commands+=("s/\$user_id/$user_id/g") sed_commands+=("s/\$ip_address/$ip_address/g") sed_commands+=("s/\$action_id/$action_id/g")
# Apply all sed commands local cmd="sed" for sed_cmd in "${sed_commands[@]}"; do cmd="$cmd -e '$sed_cmd'" done eval "$cmd" "$SCRIPT_DIR/template.json"}
# Function to build kafka-console-producer command as an array (avoids# eval'ing environment-controlled values as shell code)build_kafka_command() { kafka_cmd=(kafka-console-producer --broker-list "$KAFKA_BROKER" --topic "$KAFKA_TOPIC")
if [ -n "$KAFKA_CONFIG_FILE" ]; then kafka_cmd+=(--producer.config "$KAFKA_CONFIG_FILE") fi}
# Directory + FIFO used to stream messages into a single long-lived producer# process. Using our own mktemp -d directory avoids writing into shared /tmp.FIFO_DIR="$(mktemp -d)"FIFO="$FIFO_DIR/kafka_producer_fifo"producer_pid=""
# Function to handle cleanup on script terminationcleanup() { echo echo "Stopping data generation..." # Close our write end of the FIFO so the producer sees EOF and exits exec 3>&- 2>/dev/null if [ -n "$producer_pid" ]; then # Give the producer a chance to exit on its own after seeing EOF for _ in $(seq 1 20); do kill -0 "$producer_pid" 2>/dev/null || break sleep 0.1 done if kill -0 "$producer_pid" 2>/dev/null; then kill "$producer_pid" 2>/dev/null fi wait "$producer_pid" 2>/dev/null fi rm -rf "$FIFO_DIR" exit 0}
# Set up signal handlers for graceful shutdowntrap cleanup SIGINT SIGTERM
# Main executionecho "Generating actions every second to Kafka topic '$KAFKA_TOPIC'..."echo "Kafka broker: $KAFKA_BROKER"if [ -n "$KAFKA_CONFIG_FILE" ]; then echo "Using config file: $KAFKA_CONFIG_FILE"fiecho "Press Ctrl+C to stop..."echo
# Build the kafka command (populates the kafka_cmd array)build_kafka_command
# Start a single long-lived producer reading from the FIFO, instead of# spawning a brand-new JVM producer process per message.mkfifo "$FIFO" || { echo "Failed to create FIFO at $FIFO" >&2; exit 1; }"${kafka_cmd[@]}" < "$FIFO" &producer_pid=$!
# Open the FIFO for writing on fd 3 and keep it open for the life of the# script, so the producer never sees EOF between messages.exec 3> "$FIFO"
# Infinite loop to generate and send actionswhile true; do action=$(generate_action) echo -e "Sending $action" echo -e "$action" >&3
# Increment action_id in the main shell ((action_id++)) sleep 1done