kafka-podman

kafka-podman.sh

#!/usr/bin/env bash
# Requisitos de config:
#   conf/zk/zoo.cfg
#   conf/kafka/server-1.properties, server-2.properties, server-3.properties
set -euo pipefail

# ======= Estilo =======
if [[ -t 1 ]]; then
  BOLD="\033[1m"; DIM="\033[2m"; RESET="\033[0m"
  RED="\033[31m"; GREEN="\033[32m"; YELLOW="\033[33m"; BLUE="\033[34m"; MAGENTA="\033[35m"; CYAN="\033[36m"
else
  BOLD=""; DIM=""; RESET=""; RED=""; GREEN=""; YELLOW=""; BLUE=""; MAGENTA=""; CYAN=""
fi
ok()   { printf "${GREEN}✅ %s${RESET}\n" "$*"; }
bad()  { printf "${RED}❌ %s${RESET}\n" "$*"; }
info() { printf "${BLUE}ℹ️  %s${RESET}\n" "$*"; }
title(){ printf "${BOLD}${MAGENTA}▶ %s${RESET}\n" "$*"; }
warn() { printf "${YELLOW}⚠️  %s${RESET}\n" "$*"; }

# ======= Variáveis (sobrescreva via ENV) =======
KAFKA_IMG="${KAFKA_IMG:-registry.redhat.io/amq-streams/kafka-39-rhel9:3.0.1-2}"
ZK_IMG="${KAFKA_IMG:-registry.redhat.io/amq-streams/kafka-39-rhel9:3.0.1-2}"

NET="${NET:-kafka-net}"
BASE_DIR="${BASE_DIR:-$PWD}"
CONF_ZK="${CONF_ZK:-$BASE_DIR/conf/zk/zoo.cfg}"
CONF_DIR_KAFKA="${CONF_DIR_KAFKA:-$BASE_DIR/conf/kafka}"

KAFKA_HEAP_OPTS="${KAFKA_HEAP_OPTS:--Xms256m -Xmx512m}"
# KAFKA_GC_LOG_OPTS="${KAFKA_GC_LOG_OPTS:--Xlog:gc*:stderr:time,tags}"
# Desabilita logs de GC para reduzir ruído no console
KAFKA_GC_LOG_OPTS=""
LOG_DIR_ENV="${LOG_DIR_ENV:-/tmp}"


# ======= Helpers =======
usage() {
  cat <<EOF
${BOLD}Uso:${RESET}
  ${CYAN}$0 container-setup${RESET}          ${DIM}# cria rede e volumes (idempotente)${RESET}
  ${CYAN}$0 container-destroy${RESET}        ${DIM}# apaga tudo (containers, volumes, rede)${RESET}

  ${CYAN}$0 start${RESET} [zks|brokers]      ${DIM}# se omitido, sobe zks e brokers${RESET}
  ${CYAN}$0 stop${RESET}  [zks|brokers]      ${DIM}# se omitido, para brokers e depois zks${RESET}
  ${CYAN}$0 status${RESET}                   ${DIM}# health geral (ruok/API)${RESET}
  ${CYAN}$0 logs${RESET}                     ${DIM}# últimos logs de todos${RESET}

  ${CYAN}$0 zk1|zk2|zk3${RESET} ${YELLOW}start|stop|restart|status|logs${RESET}
  ${CYAN}$0 broker1|broker2|broker3${RESET} ${YELLOW}start|stop|restart|status|logs${RESET}

  ${CYAN}$0 topic_create <nome> [partitions] [replication]${RESET}
  ${CYAN}$0 topic_list${RESET}                       ${DIM}# lista todos os tópicos existentes${RESET}
  ${CYAN}$0 produce <topic>${RESET}                  ${DIM}# producer simples (stdin)${RESET}
  ${CYAN}$0 produce_keyed <topic>${RESET}            ${DIM}# producer com chave 'key:value'${RESET}
  ${CYAN}$0 consume <topic> [group]${RESET}          ${DIM}# consumer; se group, entra em CG${RESET}

${BOLD}Vars:${RESET}
  KAFKA_IMG=${DIM}"$KAFKA_IMG"${RESET}
  ZK_IMG    ${DIM}"$ZK_IMG"${RESET}
  NET       ${DIM}"$NET"${RESET}
  CONF_ZK   ${DIM}"$CONF_ZK"${RESET}
  CONF_DIR_KAFKA ${DIM}"$CONF_DIR_KAFKA"${RESET}
EOF
  exit 1
}

require_file() { [[ -f "$1" ]] || { bad "arquivo não encontrado: $1"; exit 1; }; }
ensure_network() {
  if ! podman network inspect "$NET" >/dev/null 2>&1; then
    title "Criando rede ${NET} 🌐"; podman network create "$NET" >/dev/null; ok "Rede ${NET} criada."
  fi
}

zk_name()         { echo "zk$1"; }
zk_vol_data()     { echo "zk$1-data"; }
zk_vol_log()      { echo "zk$1-log"; }
broker_name()     { echo "kafka$1"; }
broker_vol_data() { echo "kafka$1-data"; }
broker_port()     { case "$1" in 1) echo 19092;; 2) echo 19093;; 3) echo 19094;; esac; }

# ======= SETUP / DESTROY =======
container_setup() {
  title "🔧 Criando volumes e rede"
  ensure_network
  for i in 1 2 3; do
    podman volume create "$(zk_vol_data "$i")" >/dev/null 2>&1 || true
    podman volume create "$(zk_vol_log  "$i")" >/dev/null 2>&1 || true
    podman volume create "$(broker_vol_data "$i")" >/dev/null 2>&1 || true
  done
  ok "Ambiente pronto!"
  info "Garanta os arquivos: ${CONF_ZK} e ${CONF_DIR_KAFKA}/server-[1..3].properties"
}
container_destroy() {
  title "💣 Limpando tudo"
  podman rm -f zk1 zk2 zk3 kafka1 kafka2 kafka3 >/dev/null 2>&1 || true
  for v in \
    "$(zk_vol_data 1)" "$(zk_vol_log 1)" \
    "$(zk_vol_data 2)" "$(zk_vol_log 2)" \
    "$(zk_vol_data 3)" "$(zk_vol_log 3)" \
    "$(broker_vol_data 1)" "$(broker_vol_data 2)" "$(broker_vol_data 3)"
  do podman volume rm -f "$v" >/dev/null 2>&1 || true; done
  podman network rm "$NET" >/dev/null 2>&1 || true
  ok "Tudo removido."
}

# ======= ZOOKEEPER =======
zk_start() {
  local id="$1"; ensure_network; require_file "$CONF_ZK"
  local name vdata vlogs; name="$(zk_name "$id")"; vdata="$(zk_vol_data "$id")"; vlogs="$(zk_vol_log "$id")"
  podman rm -f "$name" >/dev/null 2>&1 || true

  # porta exposta no host apenas para zk1
  local port_flag=""
  [[ "$id" == "1" ]] && port_flag="-p 2181:2181"

  title "🦓 Iniciando ZooKeeper ${name}"
  # shellcheck disable=SC2086
  podman run -d --name "$name" --hostname "$name" --network "$NET" \
    $port_flag \
    -e LOG_DIR="$LOG_DIR_ENV" -e KAFKA_GC_LOG_OPTS="$KAFKA_GC_LOG_OPTS" \
    -v "$vdata:/var/lib/zookeeper/data" \
    -v "$vlogs:/var/lib/zookeeper/log" \
    -v "$CONF_ZK:/opt/kafka/config/zookeeper.properties:Z" \
    "$ZK_IMG" \
    bash -lc "echo $id > /var/lib/zookeeper/data/myid && /opt/kafka/bin/zookeeper-server-start.sh /opt/kafka/config/zookeeper.properties" >/dev/null
  ok "$name iniciado."
}
zk_stop()   { local n; n="$(zk_name "$1")"; title "Parando ${n} 🛑"; podman rm -f "$n" >/dev/null 2>&1 || true; ok "$n parado."; }
zk_logs()   { title "Logs $(zk_name "$1") 📜"; podman logs -f "$(zk_name "$1")"; }
zk_status() { title "Status $(zk_name "$1") 🔎"; podman ps -a --filter "name=$(zk_name "$1")"; }

# ======= BROKER =======
broker_start() {
  local id="$1"; ensure_network
  local name conf port vdata; name="$(broker_name "$id")"; conf="$CONF_DIR_KAFKA/server-$id.properties"; port="$(broker_port "$id")"; vdata="$(broker_vol_data "$id")"
  require_file "$conf"; podman rm -f "$name" >/dev/null 2>&1 || true

  title "🟢 Iniciando Broker ${name} (porta $port)"
  podman run -d --name "$name" --hostname "$name" --network "$NET" \
    -p "$port:9092" \
    -e LOG_DIR="$LOG_DIR_ENV" \
    -e KAFKA_HEAP_OPTS="$KAFKA_HEAP_OPTS" \
    -e KAFKA_GC_LOG_OPTS="$KAFKA_GC_LOG_OPTS" \
    -v "$vdata:/var/lib/kafka/data" \
    -v "$conf:/opt/kafka/config/server.properties:Z" \
    "$KAFKA_IMG" \
    bash -lc '/opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/server.properties' >/dev/null
  ok "$name iniciado."
}
broker_stop()   { local n; n="$(broker_name "$1")"; title "Parando ${n} 🛑"; podman rm -f "$n" >/dev/null 2>&1 || true; ok "$n parado."; }
broker_logs()   { title "Logs $(broker_name "$1") 📜"; podman logs -f "$(broker_name "$1")"; }
broker_status() { title "Status $(broker_name "$1") 🔎"; podman ps -a --filter "name=$(broker_name "$1")"; }

# ======= HEALTHCHECK =======
zk_healthcheck() {
  local name; name="$(zk_name "$1")"

  # Existe?
  if ! podman container exists "$name" 2>/dev/null; then
    bad "$name NÃO existe (use: $0 start zks)"; return
  fi

  local running; running=$(podman inspect -f '{{.State.Running}}' "$name" 2>/dev/null || echo false)
  if [[ "$running" != "true" ]]; then
    bad "$name existe mas está PARADO ($0 zk$1 start)"; return
  fi

  local ip; ip=$(podman inspect -f '{{with index .NetworkSettings.Networks "'"$NET"'"}}{{.IPAddress}}{{end}}' "$name" 2>/dev/null || true)
  if [[ -z "$ip" ]]; then
    bad "$name rodando mas SEM IP na $NET (connect + restart)"; return
  fi

  # Tenta 'ruok' com nc
  if podman exec "$name" bash -lc 'command -v nc >/dev/null 2>&1'; then
    local out
    out=$(podman exec "$name" bash -lc 'echo ruok | nc -w 1 127.0.0.1 2181' 2>/dev/null || true)
    if [[ "$out" == imok* ]]; then ok "$name saudável (ruok)"; return; fi
    if [[ "$out" == *"not in the whitelist"* ]]; then ok "$name saudável (4lw bloqueado: whitelist)"; return; fi
  fi

  # Fallback: /dev/tcp
  if podman exec "$name" bash -lc '
      exec 3<>/dev/tcp/127.0.0.1/2181 || exit 2
      printf "ruok" >&3
      read -t 1 resp <&3 || true
      [[ "${resp:-}" == imok* ]] && exit 0
      [[ "${resp:-}" == *"not in the whitelist"* ]] && exit 0
      exit 3
    ' >/dev/null 2>&1; then
    ok "$name saudável (TCP ok; 4lw possivelmente bloqueado)"
  else
    if podman exec "$name" bash -lc 'exec 3<>/dev/tcp/127.0.0.1/2181' >/dev/null 2>&1; then
      ok "$name porta 2181 aberta (sem 4lw)"
    else
      bad "$name não respondeu na porta 2181"
    fi
  fi
}

broker_healthcheck() {
  local name; name="$(broker_name "$1")"
  if podman exec "$name" bash -lc "/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092" >/dev/null 2>&1; then
    ok "$name responde API versions"
  else
    bad "$name não respondeu à API"
  fi
}

status_everything() {
  title "📊 Status geral"
  echo "\n${BOLD}ZooKeeper:${RESET}"
  for i in 1 2 3; do zk_healthcheck "$i"; done
  echo "\n${BOLD}Brokers:${RESET}"
  for i in 1 2 3; do broker_healthcheck "$i"; done
}

logs_everything() {
  title "📚 Últimos logs"
  echo "${BOLD}ZooKeepers:${RESET}"; for i in 1 2 3; do echo "—— zk$i ——"; podman logs --tail=20 "$(zk_name "$i")" || true; done
  echo "${BOLD}Brokers:${RESET}";    for i in 1 2 3; do echo "—— kafka$i ——"; podman logs --tail=20 "$(broker_name "$i")" || true; done
}

# ======= TOPICS & CLIENTS =======
topic_create() {
  local name="${1:-}"; local parts="${2:-6}"; local rf="${3:-3}"
  if [[ -z "$name" ]]; then err "uso: $0 topic_create <nome> [partitions=6] [replication=3]"; exit 1; fi
  title "📦 Criando tópico '${name}' (partitions=${parts}, rf=${rf})"
  podman exec -it kafka1 bash -lc "/opt/kafka/bin/kafka-topics.sh \
    --create --if-not-exists \
    --bootstrap-server kafka1:9092,kafka2:9092,kafka3:9092 \
    --topic '${name}' --partitions ${parts} --replication-factor ${rf}"
  ok "Tópico '${name}' criado (ou já existia)."
}

produce() {
  local topic="${1:-}"; [[ -z "$topic" ]] && { err "uso: $0 produce <topic>"; exit 1; }
  info "✍️  Producer simples para '${topic}' (Ctrl+C para sair)."
  podman exec -it kafka1 bash -lc "/opt/kafka/bin/kafka-console-producer.sh \
    --bootstrap-server kafka1:9092 --topic '${topic}'"
}

produce_keyed() {
  local topic="${1:-}"; [[ -z "$topic" ]] && { err "uso: $0 produce_keyed <topic>"; exit 1; }
  info "🔑 Producer com chave para '${topic}' (formato 'key:value')."
  podman exec -it kafka2 bash -lc "/opt/kafka/bin/kafka-console-producer.sh \
    --bootstrap-server kafka2:9092 --topic '${topic}' \
    --property parse.key=true --property key.separator=:"
}

consume() {
  local topic="${1:-}"; local group="${2:-}"
  [[ -z "$topic" ]] && { err "uso: $0 consume <topic> [group]"; exit 1; }
  local grp_flag=""; [[ -n "$group" ]] && grp_flag="--group '${group}'"
  info "👂 Consumer para '${topic}' ${group:+(group='${group}')} — lendo desde o início."
  podman exec -it kafka3 bash -lc "/opt/kafka/bin/kafka-console-consumer.sh \
    --bootstrap-server kafka3:9092 --topic '${topic}' ${grp_flag} --from-beginning"
}

topic_list() {
  title "📜 Listando tópicos existentes"
  podman exec -it kafka1 bash -lc "/opt/kafka/bin/kafka-topics.sh \
    --list --bootstrap-server kafka1:9092,kafka2:9092,kafka3:9092"
}

# ======= CLI =======
arg1="${1:-}"; arg2="${2:-}"; arg3="${3:-}"; arg4="${4:-}"
[[ -z "$arg1" ]] && usage

case "$arg1" in
  container-setup)   container_setup ;;
  container-destroy) container_destroy ;;
  start)
    case "${arg2:-}" in
      "")        for i in 1 2 3; do zk_start "$i"; done; for i in 1 2 3; do broker_start "$i"; done ;;
      zks)       for i in 1 2 3; do zk_start "$i"; done ;;
      brokers)   for i in 1 2 3; do broker_start "$i"; done ;;
      *) usage ;;
    esac ;;
  stop)
    case "${arg2:-}" in
      "")        for i in 1 2 3; do broker_stop "$i"; done; for i in 1 2 3; do zk_stop "$i"; done ;;
      zks)       for i in 1 2 3; do zk_stop "$i"; done ;;
      brokers)   for i in 1 2 3; do broker_stop "$i"; done ;;
      *) usage ;;
    esac ;;
  status) status_everything ;;
  logs)   logs_everything  ;;

  topic_create)   topic_create "$arg2" "${arg3:-6}" "${arg4:-3}" ;;
  topic_list)     topic_list ;;
  produce)        produce "$arg2" ;;
  produce_keyed)  produce_keyed "$arg2" ;;
  consume)        consume "$arg2" "${arg3:-}" ;;

  zk1|zk2|zk3)
    node="${arg1#zk}"; action="${arg2:-}"; [[ -z "$action" ]] && usage
    case "$action" in
      start)   zk_start "$node" ;;
      stop)    zk_stop "$node" ;;
      restart) zk_stop "$node"; zk_start "$node" ;;
      status)  zk_status "$node" ;;
      logs)    zk_logs "$node" ;;
      *) usage ;;
    esac ;;
  broker1|broker2|broker3)
    node="${arg1#broker}"; action="${arg2:-}"; [[ -z "$action" ]] && usage
    case "$action" in
      start)   broker_start "$node" ;;
      stop)    broker_stop "$node" ;;
      restart) broker_stop "$node"; broker_start "$node" ;;
      status)  broker_status "$node" ;;
      logs)    broker_logs "$node" ;;
      *) usage ;;
    esac ;;
  *) usage ;;
esac