118 lines
4.6 KiB
Bash
118 lines
4.6 KiB
Bash
|
#!/usr/bin/env bash
|
||
|
|
set -euo pipefail
|
||
|
|
|
||
|
|
# Docker 单文件版：直接在 Kafka 容器内创建并校验大日志相关 Topic。
|
||
|
|
# 默认值适配测试环境；如容器名、Kafka 地址或安装目录不同，可通过环境变量覆盖。
|
||
|
|
|
||
|
|
DOCKER_BIN="${DOCKER_BIN:-docker}"
|
||
|
|
KAFKA_CONTAINER="${KAFKA_CONTAINER:-kafka_ycgy-bt-kafka-1}"
|
||
|
|
BOOTSTRAP="${BOOTSTRAP:-localhost:9092}"
|
||
|
|
KAFKA_BIN="${KAFKA_BIN:-/opt/kafka/bin}"
|
||
|
|
RAW_TOPIC="${RAW_TOPIC:-agent-01-syslog-topic}"
|
||
|
|
LARGE_TOPIC="${LARGE_TOPIC:-${RAW_TOPIC}.large}"
|
||
|
|
MAX_MESSAGE_BYTES="${MAX_MESSAGE_BYTES:-33554432}"
|
||
|
|
RETENTION_MS="${RETENTION_MS:-604800000}"
|
||
|
|
DLT_RETENTION_MS="${DLT_RETENTION_MS:-2592000000}"
|
||
|
|
COMMAND_CONFIG="${COMMAND_CONFIG:-}"
|
||
|
|
|
||
|
|
die() {
|
||
|
|
echo "ERROR: $*" >&2
|
||
|
|
exit 1
|
||
|
|
}
|
||
|
|
|
||
|
|
command -v "${DOCKER_BIN}" >/dev/null 2>&1 || die "未找到 Docker CLI: ${DOCKER_BIN}"
|
||
|
|
|
||
|
|
RUNNING="$("${DOCKER_BIN}" inspect -f '{{.State.Running}}' "${KAFKA_CONTAINER}" 2>/dev/null || true)"
|
||
|
|
[[ "${RUNNING}" == "true" ]] || die "Kafka 容器不存在或未运行: ${KAFKA_CONTAINER}"
|
||
|
|
|
||
|
|
"${DOCKER_BIN}" exec "${KAFKA_CONTAINER}" test -x "${KAFKA_BIN}/kafka-topics.sh" \
|
||
|
|
|| die "容器内找不到可执行文件: ${KAFKA_BIN}/kafka-topics.sh"
|
||
|
|
"${DOCKER_BIN}" exec "${KAFKA_CONTAINER}" test -x "${KAFKA_BIN}/kafka-configs.sh" \
|
||
|
|
|| die "容器内找不到可执行文件: ${KAFKA_BIN}/kafka-configs.sh"
|
||
|
|
"${DOCKER_BIN}" exec "${KAFKA_CONTAINER}" bash -c 'exit 0' \
|
||
|
|
|| die "容器内缺少 bash，无法执行原始升级脚本"
|
||
|
|
|
||
|
|
DOCKER_ENV=(
|
||
|
|
-e "BOOTSTRAP=${BOOTSTRAP}"
|
||
|
|
-e "KAFKA_BIN=${KAFKA_BIN}"
|
||
|
|
-e "RAW_TOPIC=${RAW_TOPIC}"
|
||
|
|
-e "LARGE_TOPIC=${LARGE_TOPIC}"
|
||
|
|
-e "MAX_MESSAGE_BYTES=${MAX_MESSAGE_BYTES}"
|
||
|
|
-e "RETENTION_MS=${RETENTION_MS}"
|
||
|
|
-e "DLT_RETENTION_MS=${DLT_RETENTION_MS}"
|
||
|
|
)
|
||
|
|
KAFKA_COMMON=(--bootstrap-server "${BOOTSTRAP}")
|
||
|
|
|
||
|
|
if [[ -n "${COMMAND_CONFIG}" ]]; then
|
||
|
|
"${DOCKER_BIN}" exec "${KAFKA_CONTAINER}" test -r "${COMMAND_CONFIG}" \
|
||
|
|
|| die "容器内无法读取 COMMAND_CONFIG: ${COMMAND_CONFIG}"
|
||
|
|
DOCKER_ENV+=(-e "COMMAND_CONFIG=${COMMAND_CONFIG}")
|
||
|
|
KAFKA_COMMON+=(--command-config "${COMMAND_CONFIG}")
|
||
|
|
fi
|
||
|
|
|
||
|
|
echo "检查 Kafka 连接及原始 topic ..."
|
||
|
|
"${DOCKER_BIN}" exec "${KAFKA_CONTAINER}" \
|
||
|
|
"${KAFKA_BIN}/kafka-topics.sh" \
|
||
|
|
"${KAFKA_COMMON[@]}" \
|
||
|
|
--describe \
|
||
|
|
--topic "${RAW_TOPIC}" >/dev/null \
|
||
|
|
|| die "无法连接 Kafka 或原始 topic 不存在: ${RAW_TOPIC}"
|
||
|
|
|
||
|
|
echo "Kafka 容器: ${KAFKA_CONTAINER}"
|
||
|
|
echo "Bootstrap: ${BOOTSTRAP}"
|
||
|
|
echo "原始 topic: ${RAW_TOPIC}"
|
||
|
|
echo "大日志 topic: ${LARGE_TOPIC}"
|
||
|
|
echo "开始执行 Docker 版升级 ..."
|
||
|
|
|
||
|
|
exec "${DOCKER_BIN}" exec -i "${DOCKER_ENV[@]}" "${KAFKA_CONTAINER}" \
|
||
|
|
bash -s <<'KAFKA_UPGRADE_SCRIPT'
|
||
|
|
set -euo pipefail
|
||
|
|
|
||
|
|
TOPICS="${KAFKA_BIN}/kafka-topics.sh"
|
||
|
|
CONFIGS="${KAFKA_BIN}/kafka-configs.sh"
|
||
|
|
COMMON=(--bootstrap-server "${BOOTSTRAP}")
|
||
|
|
if [[ -n "${COMMAND_CONFIG:-}" ]]; then
|
||
|
|
[[ -r "${COMMAND_CONFIG}" ]] || { echo "ERROR: 无法读取 ${COMMAND_CONFIG}" >&2; exit 1; }
|
||
|
|
COMMON+=(--command-config "${COMMAND_CONFIG}")
|
||
|
|
fi
|
||
|
|
|
||
|
|
RAW_INFO=$("${TOPICS}" "${COMMON[@]}" --describe --topic "${RAW_TOPIC}")
|
||
|
|
PARTITIONS=$(printf '%s\n' "${RAW_INFO}" | awk '
|
||
|
|
/PartitionCount:/ {for(i=1;i<=NF;i++) if($i=="PartitionCount:"){print $(i+1);exit}}
|
||
|
|
')
|
||
|
|
REPLICAS=$(printf '%s\n' "${RAW_INFO}" | awk '
|
||
|
|
/ReplicationFactor:/ {for(i=1;i<=NF;i++) if($i=="ReplicationFactor:"){print $(i+1);exit}}
|
||
|
|
')
|
||
|
|
if [[ ! "${PARTITIONS}" =~ ^[1-9][0-9]*$ ]] || [[ ! "${REPLICAS}" =~ ^[1-9][0-9]*$ ]]; then
|
||
|
|
echo "ERROR: 无法读取 ${RAW_TOPIC} 的分区数或副本数" >&2
|
||
|
|
exit 1
|
||
|
|
fi
|
||
|
|
|
||
|
|
declare -a TOPIC_NAMES=(
|
||
|
|
"${LARGE_TOPIC}"
|
||
|
|
"${LARGE_TOPIC}.RETRY.1M"
|
||
|
|
"${LARGE_TOPIC}.RETRY.10M"
|
||
|
|
"${LARGE_TOPIC}.RETRY.1H"
|
||
|
|
"${LARGE_TOPIC}.DLT"
|
||
|
|
"${LARGE_TOPIC}.REPLAY"
|
||
|
|
)
|
||
|
|
|
||
|
|
echo "Kafka=${BOOTSTRAP}，分区=${PARTITIONS}，副本=${REPLICAS}，单条上限=${MAX_MESSAGE_BYTES}"
|
||
|
|
for topic in "${TOPIC_NAMES[@]}"; do
|
||
|
|
topic_retention="${RETENTION_MS}"
|
||
|
|
[[ "${topic}" == "${LARGE_TOPIC}.DLT" ]] && topic_retention="${DLT_RETENTION_MS}"
|
||
|
|
"${TOPICS}" "${COMMON[@]}" --create --if-not-exists \
|
||
|
|
--topic "${topic}" --partitions "${PARTITIONS}" --replication-factor "${REPLICAS}" \
|
||
|
|
--config cleanup.policy=delete \
|
||
|
|
--config "retention.ms=${topic_retention}" \
|
||
|
|
--config "max.message.bytes=${MAX_MESSAGE_BYTES}"
|
||
|
|
"${CONFIGS}" "${COMMON[@]}" --alter --entity-type topics --entity-name "${topic}" \
|
||
|
|
--add-config "cleanup.policy=delete,retention.ms=${topic_retention},max.message.bytes=${MAX_MESSAGE_BYTES}"
|
||
|
|
"${TOPICS}" "${COMMON[@]}" --describe --topic "${topic}"
|
||
|
|
done
|
||
|
|
|
||
|
|
echo "PASS: 大日志 Topic 已创建并统一配置。"
|
||
|
|
echo "注意：Broker 的 replica.fetch.max.bytes 也必须不小于 ${MAX_MESSAGE_BYTES}。"
|
||
|
|
echo "脚本未删除消息，也未修改任何消费组 offset。"
|
||
|
|
KAFKA_UPGRADE_SCRIPT
|