121 lines
4.2 KiB
Bash
121 lines
4.2 KiB
Bash
|
#!/usr/bin/env bash
|
|||
|
|
|
|||
|
|
set -euo pipefail
|
|||
|
|
|
|||
|
|
# 创建 syslog-consumer-rule 统计物化 Topic。
|
|||
|
|
# 默认值对应当前客户现场，直接执行即可：
|
|||
|
|
# bash 03-create-stats-materialize-topic.sh
|
|||
|
|
#
|
|||
|
|
# 其他环境可在执行前覆盖：BOOTSTRAP、RAW_TOPIC、STATS_TOPIC、KAFKA_BIN、COMMAND_CONFIG。
|
|||
|
|
|
|||
|
|
SCRIPT_DIR=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)
|
|||
|
|
BOOTSTRAP="${BOOTSTRAP:-10.141.121.35:9092,10.141.121.50:9092,10.141.121.59:9092}"
|
|||
|
|
RAW_TOPIC="${RAW_TOPIC:-agent-01-syslog-topic}"
|
|||
|
|
STATS_TOPIC="${STATS_TOPIC:-${RAW_TOPIC}.STATS.MATERIALIZE}"
|
|||
|
|
RETENTION_MS="${RETENTION_MS:-604800000}"
|
|||
|
|
|
|||
|
|
if [[ -z "${KAFKA_BIN:-}" ]]; then
|
|||
|
|
for CANDIDATE_BIN in \
|
|||
|
|
"${SCRIPT_DIR}" \
|
|||
|
|
"/opt/kafka/kafka_2.13-3.6.1/bin" \
|
|||
|
|
"/opt/kafka/bin"; do
|
|||
|
|
if [[ -x "${CANDIDATE_BIN}/kafka-topics.sh" ]] \
|
|||
|
|
&& [[ -x "${CANDIDATE_BIN}/kafka-configs.sh" ]]; then
|
|||
|
|
KAFKA_BIN="${CANDIDATE_BIN}"
|
|||
|
|
break
|
|||
|
|
fi
|
|||
|
|
done
|
|||
|
|
fi
|
|||
|
|
|
|||
|
|
if [[ -z "${KAFKA_BIN:-}" ]] && command -v kafka-topics.sh >/dev/null 2>&1; then
|
|||
|
|
KAFKA_BIN=$(dirname -- "$(command -v kafka-topics.sh)")
|
|||
|
|
fi
|
|||
|
|
|
|||
|
|
if [[ -z "${KAFKA_BIN:-}" ]]; then
|
|||
|
|
echo "ERROR: 未找到 Kafka CLI；现场预期路径为 /opt/kafka/kafka_2.13-3.6.1/bin" >&2
|
|||
|
|
exit 1
|
|||
|
|
fi
|
|||
|
|
|
|||
|
|
TOPICS_CMD="${KAFKA_BIN}/kafka-topics.sh"
|
|||
|
|
CONFIGS_CMD="${KAFKA_BIN}/kafka-configs.sh"
|
|||
|
|
COMMON_ARGS=(--bootstrap-server "${BOOTSTRAP}")
|
|||
|
|
|
|||
|
|
if [[ -n "${COMMAND_CONFIG:-}" ]]; then
|
|||
|
|
if [[ ! -r "${COMMAND_CONFIG}" ]]; then
|
|||
|
|
echo "ERROR: 无法读取认证配置 ${COMMAND_CONFIG}" >&2
|
|||
|
|
exit 1
|
|||
|
|
fi
|
|||
|
|
COMMON_ARGS+=(--command-config "${COMMAND_CONFIG}")
|
|||
|
|
fi
|
|||
|
|
|
|||
|
|
echo "Kafka：${BOOTSTRAP}"
|
|||
|
|
echo "原始 Topic：${RAW_TOPIC}"
|
|||
|
|
echo "统计 Topic：${STATS_TOPIC}"
|
|||
|
|
|
|||
|
|
echo "[1/4] 读取原始 Topic 的分区数和副本数"
|
|||
|
|
RAW_INFO=$("${TOPICS_CMD}" "${COMMON_ARGS[@]}" --describe --topic "${RAW_TOPIC}")
|
|||
|
|
printf '%s\n' "${RAW_INFO}"
|
|||
|
|
|
|||
|
|
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
|
|||
|
|
echo "将沿用：分区数=${PARTITIONS}，副本数=${REPLICAS}"
|
|||
|
|
|
|||
|
|
echo "[2/4] 创建统计 Topic（已存在时不重复创建）"
|
|||
|
|
"${TOPICS_CMD}" "${COMMON_ARGS[@]}" \
|
|||
|
|
--create --if-not-exists \
|
|||
|
|
--topic "${STATS_TOPIC}" \
|
|||
|
|
--partitions "${PARTITIONS}" \
|
|||
|
|
--replication-factor "${REPLICAS}" \
|
|||
|
|
--config cleanup.policy=delete \
|
|||
|
|
--config "retention.ms=${RETENTION_MS}"
|
|||
|
|
|
|||
|
|
echo "[3/4] 统一保留策略"
|
|||
|
|
if ! "${CONFIGS_CMD}" "${COMMON_ARGS[@]}" \
|
|||
|
|
--alter --entity-type topics --entity-name "${STATS_TOPIC}" \
|
|||
|
|
--add-config "cleanup.policy=delete,retention.ms=${RETENTION_MS}"; then
|
|||
|
|
echo "WARN: Kafka 配置更新请求超时或连接中断，将继续验证 Topic。" >&2
|
|||
|
|
echo "WARN: 新建命令已经携带相同配置，不影响 syslog-consumer-rule 使用该 Topic。" >&2
|
|||
|
|
fi
|
|||
|
|
|
|||
|
|
echo "[4/4] 验证统计 Topic"
|
|||
|
|
STATS_INFO=$("${TOPICS_CMD}" "${COMMON_ARGS[@]}" --describe --topic "${STATS_TOPIC}")
|
|||
|
|
printf '%s\n' "${STATS_INFO}"
|
|||
|
|
|
|||
|
|
ACTUAL_PARTITIONS=$(printf '%s\n' "${STATS_INFO}" | awk '
|
|||
|
|
/PartitionCount:/ {
|
|||
|
|
for (i=1; i<=NF; i++) if ($i=="PartitionCount:") { print $(i+1); exit }
|
|||
|
|
}
|
|||
|
|
')
|
|||
|
|
ACTUAL_REPLICAS=$(printf '%s\n' "${STATS_INFO}" | awk '
|
|||
|
|
/ReplicationFactor:/ {
|
|||
|
|
for (i=1; i<=NF; i++) if ($i=="ReplicationFactor:") { print $(i+1); exit }
|
|||
|
|
}
|
|||
|
|
')
|
|||
|
|
|
|||
|
|
if [[ "${ACTUAL_PARTITIONS}" != "${PARTITIONS}" ]]; then
|
|||
|
|
echo "ERROR: ${STATS_TOPIC} 已存在，但分区数=${ACTUAL_PARTITIONS}，与原始 Topic 的 ${PARTITIONS} 不一致" >&2
|
|||
|
|
echo "为避免同一 key 重新映射，脚本不会自动增加分区，请先联系研发确认。" >&2
|
|||
|
|
exit 1
|
|||
|
|
fi
|
|||
|
|
if [[ "${ACTUAL_REPLICAS}" != "${REPLICAS}" ]]; then
|
|||
|
|
echo "ERROR: ${STATS_TOPIC} 副本数=${ACTUAL_REPLICAS}，预期=${REPLICAS}" >&2
|
|||
|
|
exit 1
|
|||
|
|
fi
|
|||
|
|
|
|||
|
|
echo "PASS: ${STATS_TOPIC} 已创建并验证成功。"
|
|||
|
|
echo "分区数=${ACTUAL_PARTITIONS}，副本数=${ACTUAL_REPLICAS}，保留时间=${RETENTION_MS}ms。"
|
|||
|
|
echo "脚本未删除消息，未修改任何消费组 offset。"
|