From 3b6c9e4714db014a2b646a8116d5f8d3bf28c7d4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=8B=E7=BB=8D=E5=AE=8F?= <994233303@qq.com> Date: Sat, 12 Sep 2026 11:40:57 +0800 Subject: [PATCH] =?UTF-8?q?fix(syslog-consumer-rule):=20=E4=BF=AE=E5=A4=8D?= =?UTF-8?q?=E7=8E=B0=E5=9C=BA=E6=97=A5=E5=BF=97=E8=A7=A3=E6=9E=90=E5=B9=B6?= =?UTF-8?q?=E8=A1=A5=E5=85=85=E5=8D=87=E7=BA=A7=E5=B7=A5=E5=85=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 支持设备头字段值包含空格，避免 WAF 等名称被误拆为无效键值对 - 将空字符串 cropper offset 按未配置处理，兼容历史标准化规则 - 增加设备头解析和空 offset 回归测试 - 增加统计物化 Topic 创建脚本 - 增加容器内 JAR 热替换脚本及操作说明 --- .../NormalData/LogNormalProcessor.java | 15 ++- .../compiled/CompiledRuleCompiler.java | 6 + .../java/com/common/util/KeyValueParser.java | 43 +------ .../java/com/common/util/SyslogParser.java | 46 ++++--- .../CompiledRuleExecutionEquivalenceTest.java | 16 +++ .../common/util/SyslogParserHeaderTest.java | 54 ++++++++ .../replace-syslog-consumer-rule-jar.sh | 75 +++++++++++ .../replace-syslog-consumer-rule.md | 10 ++ .../03-create-stats-materialize-topic.sh | 120 ++++++++++++++++++ haobang-security-dm/upgrade-doc/v2/README.md | 2 +- 10 files changed, 321 insertions(+), 66 deletions(-) create mode 100644 haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/util/SyslogParserHeaderTest.java create mode 100755 haobang-security-dm/upgrade-doc/replace-syslog-consumer-rule-jar.sh create mode 100644 haobang-security-dm/upgrade-doc/replace-syslog-consumer-rule.md create mode 100755 haobang-security-dm/upgrade-doc/v1/kafka/03-create-stats-materialize-topic.sh diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/LogNormalProcessor.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/LogNormalProcessor.java index 22cd616..f73e8b0 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/LogNormalProcessor.java +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/LogNormalProcessor.java @@ -920,12 +920,15 @@ public class LogNormalProcessor { try { JSONObject jsonObject = new JSONObject(dmNormalizeRule.get("rule_content").toString()); if (jsonObject.isEmpty()) return null; - cropperParams.sethead_key(jsonObject.getJSONObject("cropper_params").get("head_key").toString()); - cropperParams.sethead_offset(Integer.parseInt( - jsonObject.getJSONObject("cropper_params").get("head_offset").toString())); - cropperParams.settail_key(jsonObject.getJSONObject("cropper_params").get("tail_key").toString()); - cropperParams.settail_offset(Integer.parseInt( - jsonObject.getJSONObject("cropper_params").get("tail_offset").toString())); + JSONObject cropper = jsonObject.getJSONObject("cropper_params"); + String headOffset = cropper.get("head_offset").toString().trim(); + String tailOffset = cropper.get("tail_offset").toString().trim(); + // 历史规则可能把 offset 保存为空字符串；此时等同于未配置截取规则，直接使用默认解析。 + if (headOffset.isEmpty() || tailOffset.isEmpty()) return null; + cropperParams.sethead_key(cropper.get("head_key").toString()); + cropperParams.sethead_offset(Integer.parseInt(headOffset)); + cropperParams.settail_key(cropper.get("tail_key").toString()); + cropperParams.settail_offset(Integer.parseInt(tailOffset)); return cropperParams; } catch (Exception ex) { logger.error("getCropperParams：" + ex.getMessage()); diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledRuleCompiler.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledRuleCompiler.java index 2acb088..55d61f1 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledRuleCompiler.java +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledRuleCompiler.java @@ -98,10 +98,16 @@ public class CompiledRuleCompiler { private CompiledCropper compileCropper(JsonNode node, RuleSource source) { if (node == null || node.isNull()) return null; if (!node.isObject()) throw invalid(source, "cropper_params 必须是对象"); + // 达梦历史规则可能把 offset 保存为空字符串；与旧执行器保持一致，忽略整段截取配置。 + if (emptyText(node.get("head_offset")) || emptyText(node.get("tail_offset"))) return null; return new CompiledCropper(textOrEmpty(node.get("head_key")), integer(node, "head_offset", source), textOrEmpty(node.get("tail_key")), integer(node, "tail_offset", source)); } + private static boolean emptyText(JsonNode node) { + return node != null && node.isTextual() && node.textValue().trim().isEmpty(); + } + private KvConfig compileKv(JsonNode node, String dataType, RuleSource source) { if (!"kv".equals(dataType)) { return node != null && node.isObject() diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/util/KeyValueParser.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/util/KeyValueParser.java index fb6f5cb..4b62e12 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/util/KeyValueParser.java +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/util/KeyValueParser.java @@ -71,46 +71,7 @@ public class KeyValueParser { * @throws IllegalArgumentException 如果格式不正确 */ public static Map parseKeyValuePairs(String text) { - if (text == null || text.trim().isEmpty()) { - throw new IllegalArgumentException("Input text cannot be null or empty"); - } - - // 查找第一个 [ 和对应的 ] - int start = text.indexOf('['); - int end = text.indexOf(']', start + 1); - - if (start == -1 || end == -1) { - throw new IllegalArgumentException("No valid [] pair found in text"); - } - - // 提取 [] 内的内容 - String content = text.substring(start + 1, end).trim(); - if (content.isEmpty()) { - return new HashMap<>(); // 返回空Map - } - - // 分割键值对 - String[] pairs = content.split("\\s+"); - Map result = new HashMap<>(); - - for (String pair : pairs) { - // 分割键和值 - int equalSign = pair.indexOf('='); - if (equalSign == -1) { - throw new IllegalArgumentException("Invalid key-value pair format: " + pair); - } - - String key = pair.substring(0, equalSign).trim(); - String value = pair.substring(equalSign + 1).trim(); - - if (key.isEmpty()) { - throw new IllegalArgumentException("Key cannot be empty in pair: " + pair); - } - - result.put(key, value); - } - - return result; + return SyslogParser.parseKeyValuePairs(text); } public static void main(String[] args) { String input = "[name=非工作时间访问 riskLevel=中 riskMain=账号 appName=蜜罐系统-企信外网 riskDesc=账号在2025-04-27，通过47.94.211.49客户端IP在非常用访问时间2025-04-27 15:38:54，进行了业务访问]"; @@ -132,4 +93,4 @@ public class KeyValueParser { System.out.println("风险等级: " + result1.get("riskLevel")); System.out.println("风险描述: " + result1.get("riskDesc")); } -} \ No newline at end of file +} diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/util/SyslogParser.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/util/SyslogParser.java index b27c9cf..ff4e4c2 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/util/SyslogParser.java +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/util/SyslogParser.java @@ -42,6 +42,13 @@ public class SyslogParser { "(?.+)$" ); + /** + * syslog-serve 的设备头使用空格分隔 key=value，但历史格式没有对 value 中的空格做转义。 + * 因此不能直接按空白切分，只能把“空白 + 合法 key=”识别为下一个字段的开始。 + */ + private static final Pattern DEVICE_HEADER_KEY_PATTERN = + Pattern.compile("(?(); // 返回空Map } - // 分割键值对 - String[] pairs = content.split("\\s+"); - Map result = new HashMap<>(); + Map result = new LinkedHashMap<>(); + Matcher matcher = DEVICE_HEADER_KEY_PATTERN.matcher(content); + String previousKey = null; + int previousValueStart = -1; - for (String pair : pairs) { - // 分割键和值 - int equalSign = pair.indexOf('='); - if (equalSign == -1) { - throw new IllegalArgumentException("Invalid key-value pair format: " + pair); + while (matcher.find()) { + if (previousKey == null) { + String invalidPrefix = content.substring(0, matcher.start()).trim(); + if (!invalidPrefix.isEmpty()) { + throw new IllegalArgumentException("Invalid key-value pair format: " + invalidPrefix); + } + } else { + String value = content.substring(previousValueStart, matcher.start()).trim(); + result.put(previousKey, value); } - - String key = pair.substring(0, equalSign).trim(); - String value = pair.substring(equalSign + 1).trim(); - - if (key.isEmpty()) { - throw new IllegalArgumentException("Key cannot be empty in pair: " + pair); - } - - result.put(key, value); + previousKey = matcher.group(1); + previousValueStart = matcher.end(); } + if (previousKey == null) { + throw new IllegalArgumentException("Invalid key-value pair format: " + content); + } + + result.put(previousKey, content.substring(previousValueStart).trim()); + return result; } } - diff --git a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/CompiledRuleExecutionEquivalenceTest.java b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/CompiledRuleExecutionEquivalenceTest.java index daf4fee..8a7aec4 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/CompiledRuleExecutionEquivalenceTest.java +++ b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/CompiledRuleExecutionEquivalenceTest.java @@ -23,6 +23,7 @@ import java.util.LinkedHashMap; import java.util.List; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.mockito.Mockito.mock; class CompiledRuleExecutionEquivalenceTest { @@ -61,6 +62,21 @@ class CompiledRuleExecutionEquivalenceTest { } } + @Test + void blankCropperOffsetsAreTreatedAsNoCropperConfiguration() { + RuleSource source = new RuleSource(9L, "blank-offset", "blank-offset", "json", "{" + + "\"cropper_params\":{\"head_key\":\"{\",\"head_offset\":\"\"," + + "\"tail_key\":\"\",\"tail_offset\":\"\"},\"mappers\":[]}", true, "0"); + String message = "<14> host {\"status\":\"ok\"}"; + + assertNull(legacy.getCropperParams(source.asLegacyMap())); + assertEquals("ok", legacy.getMessageToMap(source.asLegacyMap(), message, "json").get("status")); + + CompiledRule compiled = compiler.compile(source); + assertNull(compiled.getCropper()); + assertEquals("ok", compiled.parse(message).getFields().get("status")); + } + private static final class Fixture { private final String type; private final String message; diff --git a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/util/SyslogParserHeaderTest.java b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/util/SyslogParserHeaderTest.java new file mode 100644 index 0000000..4d82840 --- /dev/null +++ b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/util/SyslogParserHeaderTest.java @@ -0,0 +1,54 @@ +package com.common.util; + +import org.junit.jupiter.api.Test; + +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +class SyslogParserHeaderTest { + + @Test + void shouldKeepSpacesInsideDeviceHeaderValues() { + String message = "[receive_time=20260911221200000 event_time=20260911221159000 " + + "device_id=12 device_name=边界防护 WAF vendor=某某 安全厂商 " + + "data_type=json device_collect_id=2]{\"message\":\"ok\"}"; + + Map result = SyslogParser.parseKeyValuePairs(message); + + assertEquals("20260911221200000", result.get("receive_time")); + assertEquals("20260911221159000", result.get("event_time")); + assertEquals("12", result.get("device_id")); + assertEquals("边界防护 WAF", result.get("device_name")); + assertEquals("某某 安全厂商", result.get("vendor")); + assertEquals("json", result.get("data_type")); + assertEquals("2", result.get("device_collect_id")); + } + + @Test + void shouldKeepEmptyAndEqualsContainingValues() { + Map result = SyslogParser.parseKeyValuePairs( + "[device_id=12 device_name= vendor=WAF=主节点 data_type=json]"); + + assertEquals("", result.get("device_name")); + assertEquals("WAF=主节点", result.get("vendor")); + } + + @Test + void shouldStillParseLegacySingleWordValues() { + Map result = SyslogParser.parseKeyValuePairs( + "[device_id=1 device_name=honeypot vendor=changting data_type=json]"); + + assertEquals("honeypot", result.get("device_name")); + assertEquals("changting", result.get("vendor")); + } + + @Test + void shouldRejectHeaderWithoutKeyValuePair() { + assertThrows(IllegalArgumentException.class, + () -> SyslogParser.parseKeyValuePairs("[WAF]")); + assertThrows(IllegalArgumentException.class, + () -> SyslogParser.parseKeyValuePairs("[WAF device_id=12]")); + } +} diff --git a/haobang-security-dm/upgrade-doc/replace-syslog-consumer-rule-jar.sh b/haobang-security-dm/upgrade-doc/replace-syslog-consumer-rule-jar.sh new file mode 100755 index 0000000..bf85f09 --- /dev/null +++ b/haobang-security-dm/upgrade-doc/replace-syslog-consumer-rule-jar.sh @@ -0,0 +1,75 @@ +#!/usr/bin/env bash +set -Eeuo pipefail + +# 现场临时热替换 syslog-consumer-rule 容器内的 JAR。 +# 默认容器名和容器内路径来自当前生产镜像： +# syslog-consumer-rule-dm:/app/syslog-consumer-rule.jar +# +# 用法： +# ./upgrade-doc/replace-syslog-consumer-rule-jar.sh +# ./upgrade-doc/replace-syslog-consumer-rule-jar.sh /path/to/syslog-consumer-rule-1.0.0.jar + +CONTAINER_NAME="syslog-consumer-rule-dm" +CONTAINER_JAR="/app/syslog-consumer-rule.jar" + +SCRIPT_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) +PROJECT_DIR=$(cd "${SCRIPT_DIR}/.." && pwd) +DEFAULT_JAR="${PROJECT_DIR}/syslog-consumer-rule/target/syslog-consumer-rule-1.0.0.jar" +LOCAL_JAR="${1:-${DEFAULT_JAR}}" +BACKUP_DIR="${SCRIPT_DIR}/backups" +BACKUP_JAR="${BACKUP_DIR}/syslog-consumer-rule-$(date '+%Y%m%d-%H%M%S').jar" +STARTED_AT="" + +fail() { + echo "ERROR: $*" >&2 + exit 1 +} + +command -v docker >/dev/null 2>&1 || fail "未找到 docker 命令" +[[ -f "${LOCAL_JAR}" && -s "${LOCAL_JAR}" ]] || fail "新 JAR 不存在或为空: ${LOCAL_JAR}" +docker inspect "${CONTAINER_NAME}" >/dev/null 2>&1 || fail "容器不存在: ${CONTAINER_NAME}" + +mkdir -p "${BACKUP_DIR}" + +echo "[1/5] 备份容器内旧 JAR" +docker cp "${CONTAINER_NAME}:${CONTAINER_JAR}" "${BACKUP_JAR}" +echo "旧 JAR 已备份到: ${BACKUP_JAR}" + +echo "[2/5] 停止容器 ${CONTAINER_NAME}" +docker stop "${CONTAINER_NAME}" >/dev/null + +restart_after_failure() { + local exit_code=$? + if [[ ${exit_code} -ne 0 ]]; then + echo "替换过程失败，尝试重新启动容器。旧 JAR 备份位于: ${BACKUP_JAR}" >&2 + docker start "${CONTAINER_NAME}" >/dev/null 2>&1 || true + fi + exit "${exit_code}" +} +trap restart_after_failure ERR + +echo "[3/5] 替换容器内 JAR" +docker cp "${LOCAL_JAR}" "${CONTAINER_NAME}:${CONTAINER_JAR}" + +echo "[4/5] 启动容器" +STARTED_AT=$(date -u '+%Y-%m-%dT%H:%M:%SZ') +docker start "${CONTAINER_NAME}" >/dev/null +sleep 8 + +RUNNING=$(docker inspect --format '{{.State.Running}}' "${CONTAINER_NAME}") +if [[ "${RUNNING}" != "true" ]]; then + trap - ERR + echo "ERROR: 容器启动后未保持运行，启动日志如下：" >&2 + docker logs --since "${STARTED_AT}" "${CONTAINER_NAME}" 2>&1 | tail -80 >&2 || true + echo "可使用以下命令回滚：" >&2 + echo "docker cp '${BACKUP_JAR}' '${CONTAINER_NAME}:${CONTAINER_JAR}' && docker start '${CONTAINER_NAME}'" >&2 + exit 1 +fi + +trap - ERR + +echo "[5/5] 替换完成，显示本次启动日志" +echo "新 JAR: ${LOCAL_JAR}" +echo "旧 JAR 备份: ${BACKUP_JAR}" +echo "注意：以后若删除或重新创建容器，容器内替换会丢失，仍需更新镜像。" +docker logs --since "${STARTED_AT}" "${CONTAINER_NAME}" 2>&1 | tail -80 diff --git a/haobang-security-dm/upgrade-doc/replace-syslog-consumer-rule.md b/haobang-security-dm/upgrade-doc/replace-syslog-consumer-rule.md new file mode 100644 index 0000000..4e9b6bc --- /dev/null +++ b/haobang-security-dm/upgrade-doc/replace-syslog-consumer-rule.md @@ -0,0 +1,10 @@ + + +docker inspect syslog-consumer-rule-dm \ + --format 'WorkingDir={{.Config.WorkingDir}}{{println}}Entrypoint={{json .Config.Entrypoint}}{{println}}Cmd={{json .Config.Cmd}}' + + +再确认文件确实存在： + + docker exec syslog-consumer-rule-dm \ + sh -c 'ls -lh /app/*.jar 2>/dev/null' \ No newline at end of file diff --git a/haobang-security-dm/upgrade-doc/v1/kafka/03-create-stats-materialize-topic.sh b/haobang-security-dm/upgrade-doc/v1/kafka/03-create-stats-materialize-topic.sh new file mode 100755 index 0000000..9011195 --- /dev/null +++ b/haobang-security-dm/upgrade-doc/v1/kafka/03-create-stats-materialize-topic.sh @@ -0,0 +1,120 @@ +#!/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。" diff --git a/haobang-security-dm/upgrade-doc/v2/README.md b/haobang-security-dm/upgrade-doc/v2/README.md index 339f1f2..c113df2 100644 --- a/haobang-security-dm/upgrade-doc/v2/README.md +++ b/haobang-security-dm/upgrade-doc/v2/README.md @@ -163,7 +163,7 @@ services: ## 发布顺序 升级 `syslog-consumer-rule` 前，先在 Kafka 集群执行一次 -`upgrade-doc/v1/kafka/kafka-cluster-upgrade.sh`，确认已经创建： +`upgrade-doc/v1/kafka/03-create-stats-materialize-topic.sh`，确认已经创建： ```text agent-01-syslog-topic.STATS.MATERIALIZE