fix(syslog-consumer-rule): 修复现场日志解析并补充升级工具

- 支持设备头字段值包含空格，避免 WAF 等名称被误拆为无效键值对 - 将空字符串 cropper offset 按未配置处理，兼容历史标准化规则 - 增加设备头解析和空 offset 回归测试 - 增加统计物化 Topic 创建脚本 - 增加容器内 JAR 热替换脚本及操作说明
2026-09-12 11:40:57 +08:00
parent c1472b9a95
commit 3b6c9e4714
10 changed files with 321 additions and 66 deletions
@@ -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());
@@ -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()
@@ -71,46 +71,7 @@ public class KeyValueParser {
     * @throws IllegalArgumentException 如果格式不正确
     */
    public static Map<String, String> 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<String, String> 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"));
    }
-}
+}
@@ -42,6 +42,13 @@ public class SyslogParser {
                    "(?<content>.+)$"
    );

+    /**
+     * syslog-serve 的设备头使用空格分隔 key=value，但历史格式没有对 value 中的空格做转义。
+     * 因此不能直接按空白切分，只能把“空白 + 合法 key=”识别为下一个字段的开始。
+     */
+    private static final Pattern DEVICE_HEADER_KEY_PATTERN =
+            Pattern.compile("(?<!\\S)([A-Za-z_][A-Za-z0-9_.-]*)=");
+
    // RFC 3164 时间戳格式
    private static final DateTimeFormatter RFC3164_FORMATTER =
            DateTimeFormatter.ofPattern("MMM dd HH:mm:ss");
@@ -328,28 +335,31 @@ public class SyslogParser {
            return new HashMap<>(); // 返回空Map
        }

-        // 分割键值对
-        String[] pairs = content.split("\\s+");
-        Map<String, String> result = new HashMap<>();
+        Map<String, String> 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;
    }
 }
-
@@ -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;
@@ -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<String, String> 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<String, String> 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<String, String> 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]"));
+    }
+}
@@ -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
@@ -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'
@@ -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。"
@@ -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