From 1c07acc01d0a7649bf4f54ecf0de1e67287b2e74 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8E=8B=E7=BB=8D=E5=AE=8F?= <994233303@qq.com> Date: Mon, 14 Sep 2026 23:11:45 +0800 Subject: [PATCH] =?UTF-8?q?feat(syslog-consumer-rule):=20=E5=A2=9E?= =?UTF-8?q?=E5=8A=A0=E5=A4=9A=E8=8A=82=E7=82=B9=E5=8E=9F=E5=A7=8B=E6=97=A5?= =?UTF-8?q?=E5=BF=97=E9=87=87=E9=9B=86=E5=B9=B6=E4=BF=AE=E5=A4=8D=E5=A4=A9?= =?UTF-8?q?=E7=9C=BC/=E5=BE=AE=E6=AD=A5JSON=E8=A7=84=E5=88=99=E5=8C=B9?= =?UTF-8?q?=E9=85=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 一、增加现场原始日志采集功能 1. 新增按设备采集原始日志的API： - 指定设备ID； - 采集时长最长10分钟； - 单次最多返回50条样本。 2. 样本选择策略： - 优先保留时间窗口内最早到达的N条日志； - 后续出现JSON解析异常、非标、规则执行异常、InfluxDB/达梦写入异常或Retry/DLT时， 使用异常日志逐条替换最早的正常样本； - 异常样本同样遵循先到先得。 3. 支持三个syslog-consumer-rule实例联合采集： - 采集请求只需发送到任意一个实例； - 三个实例通过公共Redis同步采集任务并汇总候选样本； - 不需要配置其他实例IP； - Redis只保存有界候选集，不保存时间窗口内的全部日志。 4. 生命周期和资源控制： - 采集到期后自动停止接收新日志； - 默认进入30秒DRAINING阶段，等待异步处理异常完成样本替换； - 最终结果默认保留30分钟，之后自动删除； - 单实例、单会话最多保留“前N条+前N条异常”； - Redis故障不阻断Kafka主消费链路。 5. 返回规则解析和非标表截断之前的完整原始报文，不受 HB_XDR_TRUNCATED字段截断影响。 6. 原始日志采集和多实例汇聚默认启用： SYSLOG_RAW_CAPTURE_ENABLED=true SYSLOG_RAW_CAPTURE_CLUSTER_ENABLED=true 二、修复天眼/微步JSON规则匹配问题 1. JSON规则配置了cropper时，先判断原始报文是否包含对应head_key； 不匹配的规则不再将当前报文误判为非法JSON。 2. 同一设备绑定多条规则时，单条不适用或解析失败的JSON规则不再立即终止整个规则链， 程序会继续匹配后续适用规则。 3. 只有全部适用JSON规则均解析失败、且没有其他规则成功处理时，才写入 “JSON解析异常”非标记录。 4. 对cropper截取后仍包含Syslog或设备前缀的内容，在JSON专用路径中重新收敛到 第一个“{”至最后一个“}”，确保交给Jackson的是完整外层JSON。 5. 修复以下现场场景： - 天眼规则129截取结果仍带有alarm前缀； - 微步设备同时绑定固定偏移规则和Threatbook/TDP JSON规则； - 不适用的固定偏移规则抢先产生JSON解析异常； - Syslog头加JSON正文不能继续匹配后续适用规则。 三、增加现场采集工具及运维说明 新增upgrade-doc/v5： - README.md：升级、配置、生命周期和故障排查说明； - capture-device-raw-logs.sh：一键创建采集会话、等待完成并自动落盘。 脚本输出： - result.json：完整采集结果； - samples.jsonl：每行一条样本及处理状态； - raw/*.log：每条未截断原始报文； - summary.txt：采集数量、异常数量和替换数量； - latest.json：中断或超时时的最后有效快照。 四、兼容性与升级范围 - 兼容JDK 8； - 无需新增或修改达梦数据库表； - 无需创建新的Kafka Topic； - 需要滚动升级全部syslog-consumer-rule实例； - 三个实例必须连接相同的Redis； - 本次提交不包含syslog-serve的高内存、WAL或JVM参数调整。 --- .../scripts/dm_seed_syslog_normal_data.py | 1016 +++++++++++++++++ .../NormalData/LogNormalProcessor.java | 64 +- .../Modules/NormalData/SysLogProcessor.java | 57 +- .../NormalData/compiled/CompiledRule.java | 27 + .../service/RawLogCaptureClusterStore.java | 304 +++++ .../common/service/RawLogCaptureService.java | 798 +++++++++++++ .../controllers/RawLogCaptureController.java | 104 ++ .../src/main/resources/application.properties | 13 + .../LogNormalProcessorRuleSelectionTest.java | 122 +- .../service/RawLogCaptureServiceTest.java | 154 +++ haobang-security-dm/upgrade-doc/README.md | 7 +- .../config/syslog-consumer-rule.env.example | 10 + .../v3/config/01-new-environment-variables.md | 24 +- .../v3/kafka/03-query-kafka-consumer-lag.md | 45 + .../04-query-backlog-sampling-dropped.md | 61 + haobang-security-dm/upgrade-doc/v5/README.md | 139 +++ .../upgrade-doc/v5/capture-device-raw-logs.sh | 229 ++++ 17 files changed, 3131 insertions(+), 43 deletions(-) create mode 100755 haobang-security-dm/scripts/dm_seed_syslog_normal_data.py create mode 100644 haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/RawLogCaptureClusterStore.java create mode 100644 haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/RawLogCaptureService.java create mode 100644 haobang-security-dm/syslog-consumer-rule/src/main/java/com/controllers/RawLogCaptureController.java create mode 100644 haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/RawLogCaptureServiceTest.java create mode 100644 haobang-security-dm/upgrade-doc/v3/kafka/03-query-kafka-consumer-lag.md create mode 100644 haobang-security-dm/upgrade-doc/v3/kafka/04-query-backlog-sampling-dropped.md create mode 100644 haobang-security-dm/upgrade-doc/v5/README.md create mode 100755 haobang-security-dm/upgrade-doc/v5/capture-device-raw-logs.sh diff --git a/haobang-security-dm/scripts/dm_seed_syslog_normal_data.py b/haobang-security-dm/scripts/dm_seed_syslog_normal_data.py new file mode 100755 index 0000000..e09d9a4 --- /dev/null +++ b/haobang-security-dm/scripts/dm_seed_syslog_normal_data.py @@ -0,0 +1,1016 @@ +#!/usr/bin/env python3 +"""向达梦 syslog_normal_data 写入 2026-09-20 的大记录测试数据。 + +单条记录的目标大小可由用户设置，默认每一整行的所有字符字段内容合计 +512 KiB，而不是每个字段 512 KiB。脚本从达梦系统视图读取实际列结构， +为除 created_at（使用数据库默认值）外的字段生成非空值。为避免一次占用约 5 GiB +内存及产生超大事务，10,000 条记录按小批次生成、写入和提交。 + +脚本会自动识别本机达梦安装目录、设置运行库路径并加载 dmPython；如果只有 +达梦自带的驱动源码，则在 /tmp 中离线编译后直接加载，不执行联网安装。 +""" + +from __future__ import annotations + +import concurrent.futures +import getpass +import importlib +import os +import queue +import random +import re +import subprocess +import sys +import tempfile +import threading +import time +import uuid +from dataclasses import dataclass +from datetime import datetime, timedelta +from pathlib import Path +from typing import Dict, List, Optional, Sequence, Tuple +from urllib.parse import urlparse + + +TARGET_OWNER = "ECOSYS" +TARGET_TABLE_NAME = "syslog_normal_data" +TARGET_TABLE = '"ECOSYS"."syslog_normal_data"' +TARGET_PARTITION = '"ECOSYS"."syslog_normal_data_p_20260920"' +TARGET_DAY_START = datetime(2026, 9, 20, 0, 0, 0) +TARGET_DAY_END = datetime(2026, 9, 21, 0, 0, 0) +DEFAULT_RECORD_SIZE_KIB = 512 +MIN_RECORD_SIZE_KIB = 1 +MAX_RECORD_SIZE_KIB = 10 * 1024 +BATCH_RECORD_COUNT = 10_000 +DEFAULT_BATCH_SIZE = 200 +DEFAULT_CONCURRENCY = 1 +MAX_CONCURRENCY = 32 +PROGRESS_EVERY = 100 +DM_RUNTIME_MARKER = "DM_SEED_RUNTIME_READY" +DM_HOME_CANDIDATES = ( + Path("/dmdba/dmdbms"), + Path("/home/dmdba/dmdbms"), + Path("/opt/dmdbms"), +) + + +@dataclass(frozen=True) +class ConnectionTarget: + host: str + port: int + schema: Optional[str] = None + + +@dataclass(frozen=True) +class ColumnInfo: + name: str + data_type: str + data_length: Optional[int] + + +@dataclass(frozen=True) +class WorkerStats: + generation_elapsed: float + insert_elapsed: float + commit_elapsed: float + committed_count: int + + +def find_dm_home() -> Optional[Path]: + """查找本机达梦安装目录，优先采用已有 DM_HOME。""" + candidates = [] + configured_home = os.environ.get("DM_HOME") + if configured_home: + candidates.append(Path(configured_home)) + candidates.extend(DM_HOME_CANDIDATES) + + for candidate in candidates: + resolved = candidate.expanduser().resolve() + if (resolved / "bin" / "disql").exists() or ( + resolved / "drivers" / "python" / "dmPython" + ).exists(): + return resolved + return None + + +def prepare_dm_runtime_environment() -> Optional[Path]: + """自动设置达梦运行库路径，并通过重启 Python 让动态加载器生效。""" + dm_home = find_dm_home() + if dm_home is None or os.environ.get(DM_RUNTIME_MARKER) == "1": + return dm_home + + library_directories = [ + path + for path in (dm_home / "bin", dm_home / "drivers" / "dpi") + if path.is_dir() + ] + child_environment = os.environ.copy() + child_environment["DM_HOME"] = str(dm_home) + child_environment[DM_RUNTIME_MARKER] = "1" + + current_library_path = child_environment.get("LD_LIBRARY_PATH", "") + library_path_parts = [str(path) for path in library_directories] + if current_library_path: + library_path_parts.append(current_library_path) + child_environment["LD_LIBRARY_PATH"] = os.pathsep.join(library_path_parts) + + script_path = str(Path(__file__).resolve()) + print( + f"[准备] 已识别达梦目录 {dm_home}，正在自动加载达梦运行库环境……", + flush=True, + ) + os.execve( + sys.executable, + [sys.executable, script_path, *sys.argv[1:]], + child_environment, + ) + return dm_home # pragma: no cover - os.execve 成功后不会返回 + + +def try_import_dm_python(search_paths: Sequence[Path] = ()): + for path in search_paths: + path_text = str(path) + if path.is_dir() and path_text not in sys.path: + sys.path.insert(0, path_text) + importlib.invalidate_caches() + sys.modules.pop("dmPython", None) + sys.modules.pop("dpi", None) + return importlib.import_module("dmPython") + + +def existing_dm_python_paths(dm_home: Path) -> List[Path]: + source_directory = dm_home / "drivers" / "python" / "dmPython" + paths = [source_directory] + if source_directory.is_dir(): + paths.extend( + path + for path in source_directory.glob("build/lib*") + if path.is_dir() + ) + paths.extend( + module.parent + for module in source_directory.rglob("dmPython*.so") + ) + # 保持顺序并去重。 + return list(dict.fromkeys(paths)) + + +def build_dm_python_offline(dm_home: Path) -> Path: + """使用达梦安装介质中的源码在 /tmp 离线编译扩展，不执行 pip。""" + source_directory = dm_home / "drivers" / "python" / "dmPython" + setup_file = source_directory / "setup.py" + if not setup_file.is_file(): + raise RuntimeError(f"未找到达梦 Python 驱动源码: {setup_file}") + + user_id = os.getuid() if hasattr(os, "getuid") else 0 + python_tag = f"py{sys.version_info.major}{sys.version_info.minor}" + build_root = ( + Path(tempfile.gettempdir()) + / f"dm_seed_dmpython_{user_id}_{python_tag}" + ) + build_library = build_root / "lib" + build_temporary = build_root / "temp" + build_library.mkdir(parents=True, exist_ok=True) + build_temporary.mkdir(parents=True, exist_ok=True) + + command = [ + sys.executable, + str(setup_file), + "build_ext", + "--build-lib", + str(build_library), + "--build-temp", + str(build_temporary), + ] + print( + f"[准备] 未检测到可直接加载的 dmPython，正在使用 {setup_file} " + "离线编译（以下为实时编译输出）……", + flush=True, + ) + try: + result = subprocess.run( + command, + cwd=str(source_directory), + env=os.environ.copy(), + check=False, + timeout=300, + ) + except subprocess.TimeoutExpired as exc: + raise RuntimeError( + "达梦自带 dmPython 离线编译超过 300 秒，已停止等待。" + "请检查编译器或 Python 开发环境是否卡住" + ) from exc + if result.returncode != 0: + raise RuntimeError( + "达梦自带 dmPython 离线编译失败。服务器可能缺少 gcc、" + f"Python 开发头文件或 setuptools，退出码={result.returncode}。" + "具体原因见上方实时编译输出" + ) + print(f"[成功] dmPython 离线编译完成，输出目录: {build_library}", flush=True) + return build_library + + +def load_dm_python(dm_home: Optional[Path]): + """自动加载已安装、达梦自带或现场离线编译的 dmPython。""" + initial_error: Optional[Exception] = None + try: + module = try_import_dm_python() + print(f"[成功] 已加载现有 dmPython: {module.__file__}", flush=True) + return module + except Exception as exc: + initial_error = exc + + if dm_home is None: + raise RuntimeError( + "未找到 dmPython，也未找到达梦安装目录。已自动检查 " + "/dmdba/dmdbms、/home/dmdba/dmdbms 和 /opt/dmdbms" + ) from initial_error + + try: + module = try_import_dm_python(existing_dm_python_paths(dm_home)) + print( + f"[成功] 已从达梦安装目录加载 dmPython: {module.__file__}", + flush=True, + ) + return module + except Exception as exc: + print( + f"[准备] 达梦目录中的驱动尚不能直接加载: " + f"{type(exc).__name__}: {exc}", + flush=True, + ) + + build_library = build_dm_python_offline(dm_home) + try: + module = try_import_dm_python([build_library]) + print( + f"[成功] 已从本机达梦安装介质离线加载 dmPython: {module.__file__}", + flush=True, + ) + return module + except Exception as exc: + raise RuntimeError( + f"dmPython 已离线编译，但加载失败: {exc}" + ) from exc + + +def parse_connection_address(raw_address: str) -> ConnectionTarget: + """解析 host、host:port 或 jdbc:dm://host:port[/schema]。""" + address = raw_address.strip() + if not address: + raise ValueError("连接地址不能为空") + + if address.lower().startswith("jdbc:dm://"): + address = address[len("jdbc:") :] + elif "://" not in address: + address = "dm://" + address + + parsed = urlparse(address) + if parsed.scheme.lower() != "dm" or not parsed.hostname: + raise ValueError( + "连接地址格式不正确，请使用 IP:端口，" + "例如 192.168.4.117:5237" + ) + + try: + port = parsed.port or 5236 + except ValueError as exc: + raise ValueError("端口必须是 1 到 65535 的整数") from exc + + if not 1 <= port <= 65535: + raise ValueError("端口必须在 1 到 65535 之间") + + schema = parsed.path.strip("/") or None + if schema and "/" in schema: + raise ValueError("连接地址中只能包含一个 schema 名称") + + return ConnectionTarget(host=parsed.hostname, port=port, schema=schema) + + +def parse_record_size(raw_value: str) -> int: + """解析 512、512KB、512KiB、1MB 或 1MiB，返回字节数。""" + value = raw_value.strip() + if not value: + return DEFAULT_RECORD_SIZE_KIB * 1024 + + match = re.fullmatch(r"(\d+)\s*(KB|KIB|MB|MIB)?", value, re.IGNORECASE) + if match is None: + raise ValueError("整行大小格式错误，请输入例如 256、512KB 或 1MB") + + number = int(match.group(1)) + unit = (match.group(2) or "KIB").upper() + multiplier = 1024**2 if unit in {"MB", "MIB"} else 1024 + size_bytes = number * multiplier + minimum = MIN_RECORD_SIZE_KIB * 1024 + maximum = MAX_RECORD_SIZE_KIB * 1024 + if not minimum <= size_bytes <= maximum: + raise ValueError( + f"单条整行目标大小必须在 {MIN_RECORD_SIZE_KIB}KB 到 " + f"{MAX_RECORD_SIZE_KIB // 1024}MB 之间" + ) + return size_bytes + + +def format_size(size_bytes: int) -> str: + if size_bytes % (1024**2) == 0: + return f"{size_bytes // (1024**2):,} MiB" + return f"{size_bytes / 1024:,.2f} KiB" + + +def prompt_connection() -> Tuple[ConnectionTarget, str, str, int, int, int]: + print("达梦 syslog_normal_data 大记录测试数据写入工具") + print("连接地址支持：IP:端口 或 jdbc:dm://IP:端口[/schema]") + address = input("达梦连接地址: ").strip() + username = input("用户名: ").strip() + password = getpass.getpass("密码（输入内容不会显示）: ") + raw_record_size = input( + f"单条整行目标大小（所有字符字段合计；直接回车默认 " + f"{DEFAULT_RECORD_SIZE_KIB}KB，" + f"范围 {MIN_RECORD_SIZE_KIB}KB～{MAX_RECORD_SIZE_KIB // 1024}MB，" + "支持如 256KB、1MB）: " + ).strip() + raw_batch_size = input( + f"每批写入条数（直接回车默认 {DEFAULT_BATCH_SIZE}）: " + ).strip() + raw_concurrency = input( + f"并发数（直接回车默认 {DEFAULT_CONCURRENCY}，1 表示不并发，" + f"范围 1～{MAX_CONCURRENCY}）: " + ).strip() + + if not username: + raise ValueError("用户名不能为空") + if not password: + raise ValueError("密码不能为空") + + record_size_bytes = parse_record_size(raw_record_size) + + try: + batch_size = ( + DEFAULT_BATCH_SIZE if not raw_batch_size else int(raw_batch_size) + ) + except ValueError as exc: + raise ValueError("每批写入条数必须是整数") from exc + if not 1 <= batch_size <= BATCH_RECORD_COUNT: + raise ValueError( + f"每批写入条数必须在 1 到 {BATCH_RECORD_COUNT:,} 之间" + ) + + try: + concurrency = ( + DEFAULT_CONCURRENCY if not raw_concurrency else int(raw_concurrency) + ) + except ValueError as exc: + raise ValueError("并发数必须是整数") from exc + if not 1 <= concurrency <= MAX_CONCURRENCY: + raise ValueError(f"并发数必须在 1 到 {MAX_CONCURRENCY} 之间") + + return ( + parse_connection_address(address), + username, + password, + batch_size, + record_size_bytes, + concurrency, + ) + + +def confirm_execution( + target: ConnectionTarget, + username: str, + batch_size: int, + record_size_bytes: int, + concurrency: int, +) -> bool: + schema_info = f"，连接 schema={target.schema}" if target.schema else "" + payload_gib = ( + (BATCH_RECORD_COUNT + 1) * record_size_bytes / (1024**3) + ) + batch_payload_mib = batch_size * record_size_bytes / (1024**2) + batch_count = (BATCH_RECORD_COUNT + batch_size - 1) // batch_size + effective_concurrency = min(concurrency, batch_count) + inflight_payload_mib = batch_payload_mib * effective_concurrency + print("\n即将执行以下操作：") + print(f" 数据库: {target.host}:{target.port}{schema_info}") + print(f" 用户名: {username}") + print(f" 目标表: {TARGET_TABLE}") + print(f" 目标日分区: {TARGET_PARTITION}") + print(" log_time: 全部位于 2026-09-20") + print(" 写入数量: 1 条验证记录 + 10,000 条批量记录") + print( + f" 单条整行目标内容: {format_size(record_size_bytes)} " + "（所有字符字段合计，不是每个字段）" + ) + print( + f" 允许的整行大小范围: {MIN_RECORD_SIZE_KIB} KiB～" + f"{MAX_RECORD_SIZE_KIB // 1024} MiB" + ) + print(" 其他字段: 尽量全部非空") + print( + f" 10,001 行字符内容总量: 约 {payload_gib:.2f} GiB；" + "数据库实际占用还包含数值字段、行和 LOB 开销" + ) + print( + f" 提交策略: 每 {batch_size:,} 条提交一次，" + f"约 {batch_payload_mib:,.2f} MiB 字符内容/批" + ) + print( + f" 写入并发: {effective_concurrency} " + f"（用户设置 {concurrency}，每个并发使用独立连接；" + "另保留 1 个验证连接）" + ) + print( + f" 并发批次字符内容上限: 约 {inflight_payload_mib:,.2f} MiB，" + "不含 Python 和驱动额外内存" + ) + print(" 中断处理: 当前批次回滚，此前已提交批次会保留") + answer = input("确认表空间充足并继续？请输入 yes: ").strip().lower() + return answer in {"yes", "y"} + + +def connect_dm(dm_python, target: ConnectionTarget, username: str, password: str): + properties = { + "user": username, + "password": password, + "server": target.host, + "port": target.port, + "autoCommit": False, + } + if target.schema: + properties["schema"] = target.schema + return dm_python.connect(**properties) + + +def quote_identifier(identifier: str) -> str: + return '"' + identifier.replace('"', '""') + '"' + + +def load_table_columns(cursor) -> List[ColumnInfo]: + sql = ( + "SELECT COLUMN_NAME, DATA_TYPE, DATA_LENGTH " + "FROM ALL_TAB_COLUMNS " + "WHERE OWNER = ? AND TABLE_NAME = ? ORDER BY COLUMN_ID" + ) + candidates = ( + (TARGET_OWNER, TARGET_TABLE_NAME), + (TARGET_OWNER, TARGET_TABLE_NAME.upper()), + ) + for owner, table_name in candidates: + cursor.execute(sql, (owner, table_name)) + rows = cursor.fetchall() + if rows: + return [ + ColumnInfo( + name=str(row[0]), + data_type=str(row[1]).upper(), + data_length=None if row[2] is None else int(row[2]), + ) + for row in rows + ] + + raise RuntimeError( + f"无法从 ALL_TAB_COLUMNS 读取 {TARGET_TABLE} 的列结构；" + "请确认表名、权限和当前连接实例" + ) + + +def test_connection_and_schema(cursor) -> List[ColumnInfo]: + cursor.execute("SELECT 1 FROM DUAL") + row = cursor.fetchone() + if row is None or int(row[0]) != 1: + raise RuntimeError(f"连接测试查询返回异常结果: {row!r}") + + try: + cursor.execute( + f'SELECT "id", "log_time" FROM {TARGET_PARTITION} WHERE 1 = 0' + ) + except Exception as exc: + raise RuntimeError( + f"2026-09-20 日分区 {TARGET_PARTITION} 不存在或当前用户无权访问；" + "脚本不会自动执行 DDL，请先确认分区已创建" + ) from exc + + columns = load_table_columns(cursor) + names = {column.name.lower() for column in columns} + missing = {"id", "created_at", "log_time"} - names + if missing: + raise RuntimeError(f"目标表缺少必要字段: {', '.join(sorted(missing))}") + return columns + + +def is_character(column: ColumnInfo) -> bool: + data_type = column.data_type + return "CHAR" in data_type or "TEXT" in data_type or "CLOB" in data_type + + +def is_large_text(column: ColumnInfo) -> bool: + return "TEXT" in column.data_type or "CLOB" in column.data_type + + +def make_ascii_payload(length: int, label: str) -> str: + if length <= 0: + return "" + seed = f"dm-load-test|{label}|" + repeats = length // len(seed) + 1 + return (seed * repeats)[:length] + + +def make_short_character_value(column: ColumnInfo, run_token: str) -> str: + name = column.name.lower() + max_length = max(1, column.data_length or 64) + if "ip" in name: + value = "192.0.2.10" + elif name == "syslog_uuid": + value = f"{run_token}-0000-4000-8000-000000000000" + elif "port" in name: + value = "5237" + elif "md5" in name: + value = "0123456789abcdef0123456789abcdef" + else: + value = f"load-test-{run_token}-{column.name}" + return value.encode("ascii", errors="ignore").decode("ascii")[:max_length] + + +def build_character_values( + columns: Sequence[ColumnInfo], + run_token: str, + example_id: str, + record_size_bytes: int, +) -> Dict[str, str]: + """生成字符字段模板，使字符内容总量等于用户指定大小。""" + values: Dict[str, str] = {} + large_text_columns: List[ColumnInfo] = [] + used_bytes = len(example_id.encode("utf-8")) + + for column in columns: + name = column.name.lower() + if name in {"id", "created_at"} or not is_character(column): + continue + if is_large_text(column): + large_text_columns.append(column) + continue + value = make_short_character_value(column, run_token) + values[column.name] = value + used_bytes += len(value.encode("utf-8")) + + if not large_text_columns: + raise RuntimeError("目标表没有 TEXT/CLOB 字段，无法生成指定大小的记录") + + remaining = record_size_bytes - used_bytes + if remaining < len(large_text_columns): + raise RuntimeError("字符字段固定内容已经超过目标记录大小") + + base_size, extra_count = divmod(remaining, len(large_text_columns)) + for index, column in enumerate(large_text_columns): + length = base_size + (1 if index < extra_count else 0) + values[column.name] = make_ascii_payload( + length, f"{run_token}|{column.name}" + ) + + return values + + +def build_insert_sql(columns: Sequence[ColumnInfo]) -> str: + names = ", ".join(quote_identifier(column.name) for column in columns) + placeholders = ", ".join("?" for _ in columns) + return f"INSERT INTO {TARGET_TABLE} ({names}) VALUES ({placeholders})" + + +def value_for_column( + column: ColumnInfo, + record_id: str, + log_time: datetime, + index: int, + character_values: Dict[str, str], +): + name = column.name.lower() + data_type = column.data_type + + if name == "id": + return record_id + if is_character(column): + return character_values[column.name] + if "TIMESTAMP" in data_type or "DATETIME" in data_type or data_type == "DATE": + return log_time + if "BIT" in data_type or "BOOL" in data_type: + return index % 2 + if any(kind in data_type for kind in ("INT", "NUMBER", "DECIMAL", "NUMERIC")): + if "port" in name: + return 10_000 + index % 50_000 + return 1_000_000 + index + if any(kind in data_type for kind in ("DOUBLE", "FLOAT", "REAL")): + return float(index) + 0.125 + if "BLOB" in data_type or "BINARY" in data_type: + return b"dm-load-test" + raise RuntimeError( + f"暂不支持字段 {column.name} 的数据类型 {column.data_type}，" + "为避免静默写入 NULL，脚本已停止" + ) + + +def build_row( + columns: Sequence[ColumnInfo], + character_values: Dict[str, str], + record_id: str, + log_time: datetime, + index: int, +) -> Tuple[object, ...]: + return tuple( + value_for_column(column, record_id, log_time, index, character_values) + for column in columns + ) + + +def character_bytes_in_row( + columns: Sequence[ColumnInfo], row: Sequence[object] +) -> int: + return sum( + len(value.encode("utf-8")) + for column, value in zip(columns, row) + if is_character(column) and isinstance(value, str) + ) + + +def insert_and_verify_one( + connection, + cursor, + columns: Sequence[ColumnInfo], + insert_sql: str, + run_token: str, + record_size_bytes: int, +) -> None: + test_id = f"dm_test_{run_token}" + test_log_time = datetime(2026, 9, 20, 12, 34, 56, 123456) + character_values = build_character_values( + columns, run_token, test_id, record_size_bytes + ) + row_values = build_row( + columns, character_values, test_id, test_log_time, -1 + ) + + try: + cursor.execute(insert_sql, row_values) + connection.commit() + except Exception: + connection.rollback() + raise + + cursor.execute( + f'SELECT "id", "log_time", "created_at" ' + f'FROM {TARGET_TABLE} ' + 'WHERE "id" = ? AND "log_time" >= ? AND "log_time" < ?', + (test_id, TARGET_DAY_START, TARGET_DAY_END), + ) + result = cursor.fetchone() + if result is None: + raise RuntimeError("单条记录已提交，但提交后未查询到该记录") + + payload_bytes = character_bytes_in_row(columns, row_values) + print(f"[成功] 单条记录写入并查询成功") + print(f" id={result[0]}, log_time={result[1]}, created_at={result[2]}") + print(f" 该整行所有字符字段合计={payload_bytes:,} 字节") + + +def query_batch_summary(cursor, run_token: str): + id_prefix = f"dm_seed_{run_token}_" + cursor.execute( + f'SELECT COUNT(*), MIN("log_time"), MAX("log_time") ' + f'FROM {TARGET_TABLE} WHERE "id" BETWEEN ? AND ?', + (f"{id_prefix}00000", f"{id_prefix}09999"), + ) + return cursor.fetchone() + + +def insert_and_verify_batch( + verification_connection, + verification_cursor, + dm_python, + target: ConnectionTarget, + username: str, + password: str, + columns: Sequence[ColumnInfo], + insert_sql: str, + run_token: str, + batch_size: int, + record_size_bytes: int, + concurrency: int, +) -> None: + example_id = f"dm_seed_{run_token}_00000" + character_values = build_character_values( + columns, run_token, example_id, record_size_bytes + ) + random_source = random.Random(run_token) + day_microseconds = 24 * 60 * 60 * 1_000_000 + time_offsets = [ + random_source.randrange(day_microseconds) + for _ in range(BATCH_RECORD_COUNT) + ] + batch_starts = list(range(0, BATCH_RECORD_COUNT, batch_size)) + effective_concurrency = min(concurrency, len(batch_starts)) + work_queue = queue.Queue() + for batch_start in batch_starts: + work_queue.put(batch_start) + + stop_event = threading.Event() + progress_lock = threading.Lock() + progress = {"committed": 0, "next": PROGRESS_EVERY} + + def report_progress(batch_count: int) -> None: + with progress_lock: + progress["committed"] += batch_count + committed = progress["committed"] + if committed >= progress["next"] or committed == BATCH_RECORD_COUNT: + print( + f" 已提交 {committed:,}/{BATCH_RECORD_COUNT:,} 条", + flush=True, + ) + while progress["next"] <= committed: + progress["next"] += PROGRESS_EVERY + + def run_worker(worker_number: int) -> WorkerStats: + worker_connection = None + worker_cursor = None + generation_elapsed = 0.0 + insert_elapsed = 0.0 + commit_elapsed = 0.0 + committed_count = 0 + try: + worker_connection = connect_dm( + dm_python, target, username, password + ) + worker_cursor = worker_connection.cursor() + + while not stop_event.is_set(): + try: + batch_start = work_queue.get_nowait() + except queue.Empty: + break + + batch_count = min( + batch_size, BATCH_RECORD_COUNT - batch_start + ) + generation_started_at = time.perf_counter() + rows = [] + for index in range(batch_start, batch_start + batch_count): + log_time = TARGET_DAY_START + timedelta( + microseconds=time_offsets[index] + ) + record_id = f"dm_seed_{run_token}_{index:05d}" + rows.append( + build_row( + columns, + character_values, + record_id, + log_time, + index, + ) + ) + generation_elapsed += ( + time.perf_counter() - generation_started_at + ) + + insert_started_at = time.perf_counter() + worker_cursor.executemany(insert_sql, rows) + insert_elapsed += time.perf_counter() - insert_started_at + + commit_started_at = time.perf_counter() + worker_connection.commit() + commit_elapsed += time.perf_counter() - commit_started_at + committed_count += batch_count + report_progress(batch_count) + + return WorkerStats( + generation_elapsed=generation_elapsed, + insert_elapsed=insert_elapsed, + commit_elapsed=commit_elapsed, + committed_count=committed_count, + ) + except Exception as exc: + stop_event.set() + if worker_connection is not None: + try: + worker_connection.rollback() + except Exception: + pass + raise RuntimeError( + f"并发工作线程 {worker_number} 写入失败: {exc}" + ) from exc + finally: + if worker_cursor is not None: + try: + worker_cursor.close() + except Exception: + pass + if worker_connection is not None: + try: + worker_connection.close() + except Exception: + pass + + total_started_at = time.perf_counter() + print( + f"\n开始流式生成并写入 {BATCH_RECORD_COUNT:,} 条数据，" + f"每批 {batch_size:,} 条，并发数 {effective_concurrency}……" + ) + worker_stats: List[WorkerStats] = [] + first_error: Optional[Exception] = None + write_phase_started_at = time.perf_counter() + with concurrent.futures.ThreadPoolExecutor( + max_workers=effective_concurrency, + thread_name_prefix="dm-seed", + ) as executor: + futures = [ + executor.submit(run_worker, worker_number) + for worker_number in range(1, effective_concurrency + 1) + ] + for future in concurrent.futures.as_completed(futures): + try: + worker_stats.append(future.result()) + except Exception as exc: + stop_event.set() + if first_error is None: + first_error = exc + write_phase_elapsed = time.perf_counter() - write_phase_started_at + + if first_error is not None: + print( + f"[提示] 失败线程的当前批次已回滚，共已提交 " + f"{progress['committed']:,} 条；" + f"ID 前缀为 dm_seed_{run_token}_", + file=sys.stderr, + ) + raise first_error + + # 结束验证连接上此前的只读事务，确保看到并发连接刚提交的数据。 + verification_connection.commit() + summary = query_batch_summary(verification_cursor, run_token) + if summary is None or int(summary[0]) != BATCH_RECORD_COUNT: + actual = 0 if summary is None else int(summary[0]) + raise RuntimeError( + f"提交后校验失败：期望 {BATCH_RECORD_COUNT:,} 条，实际 {actual:,} 条" + ) + + total_elapsed = time.perf_counter() - total_started_at + generation_elapsed = sum(item.generation_elapsed for item in worker_stats) + insert_elapsed = sum(item.insert_elapsed for item in worker_stats) + commit_elapsed = sum(item.commit_elapsed for item in worker_stats) + rows_per_second = BATCH_RECORD_COUNT / write_phase_elapsed + milliseconds_per_row = write_phase_elapsed * 1_000 / BATCH_RECORD_COUNT + content_mib_per_second = ( + BATCH_RECORD_COUNT * record_size_bytes + / (1024**2) + / write_phase_elapsed + ) + + sample_row = build_row( + columns, + character_values, + example_id, + TARGET_DAY_START, + 0, + ) + sample_payload_bytes = character_bytes_in_row(columns, sample_row) + + print(f"[成功] {BATCH_RECORD_COUNT:,} 条大记录写入并查询成功") + print(f" 数量={int(summary[0]):,}") + print(f" 每一整行所有字符字段合计={sample_payload_bytes:,} 字节") + print(f" 最早 log_time={summary[1]}") + print(f" 最晚 log_time={summary[2]}") + print(f" 实际写入并发数={effective_concurrency}") + print(f" 各线程累计数据生成耗时={generation_elapsed:.3f} 秒") + print(f" 各线程累计批量插入耗时={insert_elapsed:.3f} 秒") + print(f" 各线程累计事务提交耗时={commit_elapsed:.3f} 秒") + print( + f" 并发写入墙钟耗时（连接、生成、插入、提交）=" + f"{write_phase_elapsed:.3f} 秒" + ) + print(f" 平均每条墙钟耗时={milliseconds_per_row:.3f} 毫秒") + print(f" 平均写入速度={rows_per_second:,.2f} 条/秒") + print(f" 字段内容吞吐={content_mib_per_second:,.2f} MiB/秒") + print(f" 生成、写入、提交及校验总耗时={total_elapsed:.3f} 秒") + print(f" 本次批量 ID 前缀=dm_seed_{run_token}_") + + +def main() -> int: + dm_home = prepare_dm_runtime_environment() + if dm_home is not None and os.environ.get(DM_RUNTIME_MARKER) == "1": + print( + f"[成功] 达梦运行库环境已自动就绪: {dm_home}", + flush=True, + ) + + try: + ( + target, + username, + password, + batch_size, + record_size_bytes, + concurrency, + ) = prompt_connection() + except (EOFError, KeyboardInterrupt): + print("\n操作已取消") + return 130 + except ValueError as exc: + print(f"[错误] {exc}", file=sys.stderr) + return 2 + + try: + if not confirm_execution( + target, + username, + batch_size, + record_size_bytes, + concurrency, + ): + print("操作已取消，未连接数据库，也未写入任何数据") + return 0 + except (EOFError, KeyboardInterrupt): + print("\n操作已取消，未连接数据库，也未写入任何数据") + return 130 + + try: + dm_python = load_dm_python(dm_home) + except Exception as exc: + print(f"[错误] 无法自动加载达梦 Python 驱动: {exc}", file=sys.stderr) + return 3 + + connection = None + cursor = None + run_token = uuid.uuid4().hex[:12] + + try: + print("\n[下一步] 正在连接达梦数据库并读取实际表结构……", flush=True) + connection = connect_dm(dm_python, target, username, password) + cursor = connection.cursor() + all_columns = test_connection_and_schema(cursor) + insert_columns = [ + column + for column in all_columns + if column.name.lower() != "created_at" + ] + insert_sql = build_insert_sql(insert_columns) + print( + f"[成功] 连接、日分区及表结构检查通过，共读取 " + f"{len(all_columns)} 个字段，本次填写 {len(insert_columns)} 个字段；" + "created_at 使用数据库默认值" + ) + + insert_and_verify_one( + connection, + cursor, + insert_columns, + insert_sql, + run_token, + record_size_bytes, + ) + insert_and_verify_batch( + connection, + cursor, + dm_python, + target, + username, + password, + insert_columns, + insert_sql, + run_token, + batch_size, + record_size_bytes, + concurrency, + ) + print(f"\n全部完成：本次共写入 {BATCH_RECORD_COUNT + 1:,} 条大记录。") + return 0 + except KeyboardInterrupt: + if connection is not None: + connection.rollback() + print( + "\n操作被中断，仅当前未提交批次已回滚；此前已提交记录会保留。", + file=sys.stderr, + ) + return 130 + except Exception as exc: + if connection is not None: + try: + connection.rollback() + except Exception: + pass + print(f"\n[失败] {type(exc).__name__}: {exc}", file=sys.stderr) + print( + "仅当前未提交事务已回滚；单条验证记录和此前成功提交的批次会保留。", + file=sys.stderr, + ) + return 1 + finally: + if cursor is not None: + try: + cursor.close() + except Exception: + pass + if connection is not None: + try: + connection.close() + except Exception: + pass + + +if __name__ == "__main__": + raise SystemExit(main()) 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 c640e71..1e88a3b 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 @@ -411,9 +411,6 @@ public class LogNormalProcessor { } if (ctx.dmNormalizeRuleList.size() > 1 && selectionConfig.isShadowEnabled()) { shadowSelect(ctx); - if (ctx.invalidJson) { - return invalidJsonResult(ctx, deviceID); - } } if (optimizationProperties.getCompiledRule().isEnabled()) { return executeCompiled(ctx, deviceID); @@ -433,15 +430,15 @@ public class LogNormalProcessor { for (RuleSource source : ctx.dmNormalizeRuleList) { try { executeLegacyRule(ctx, deviceId, source); - if (ctx.invalidJson) { - return invalidJsonResult(ctx, deviceId); - } } catch (Exception failure) { logger.error("泛化规则处理失败, eventId={}, ruleId={}", ctx.strSyslogUUID, source.getId(), failure); throw new IllegalStateException("标准化规则处理失败", failure); } } + if (ctx.savedCount == 0 && ctx.filteredCount == 0 && ctx.invalidJson) { + return invalidJsonResult(ctx, deviceId); + } return finish(ctx); } @@ -450,7 +447,7 @@ public class LogNormalProcessor { */ private NormalizationResult executeSelected(ProcessContext ctx, long deviceId) { List> candidates = collectCandidates(ctx); - if (ctx.invalidJson) { + if (candidates.isEmpty() && ctx.invalidJson) { return invalidJsonResult(ctx, deviceId); } RuleSelectionEngine.Decision decision = @@ -505,9 +502,9 @@ public class LogNormalProcessor { /** 关闭切换开关时仅做归属影子计算，不改变旧的多结果行为。 */ private void shadowSelect(ProcessContext ctx) { try { - List> candidates = collectCandidates(ctx); - if (ctx.invalidJson) return; - ruleSelectionEngine.select(candidates, ctx.strLogMsg, true); + ProcessContext shadow = shadowContext(ctx); + List> candidates = collectCandidates(shadow); + ruleSelectionEngine.select(candidates, shadow.strLogMsg, true); } catch (Exception failure) { // 影子链路不能影响生产结果。 logger.debug("规则归属影子计算失败: eventHash={}, error={}", @@ -525,12 +522,14 @@ public class LogNormalProcessor { if (compiledEnabled) { try { CompiledRule compiled = compiledRuleCache.get(source, ctx.executionMetrics); + if (!compiled.isApplicableTo(ctx.strLogMsg)) { + continue; + } ParsedMessage parsed = ctx.messageParseContext.parse( compiled, ctx.strLogMsg, ctx.executionMetrics); if (parsed == null) { if (isJsonRule(source)) { markInvalidJson(ctx, source); - return candidates; } continue; } @@ -557,6 +556,9 @@ public class LogNormalProcessor { } private ExecutionPlan legacyCandidate(ProcessContext ctx, RuleSource source) { + if (isJsonRule(source) && !isJsonRuleApplicable(source, ctx.strLogMsg)) { + return null; + } long parseStarted = System.nanoTime(); globalMetrics.ruleJsonParse(); LinkedHashMap parsed = getMessageToMap( @@ -678,9 +680,6 @@ public class LogNormalProcessor { // 首次失败原因由 CompiledRuleCache 按 fingerprint 限频记录；此处只回退该规则。 try { executeLegacyRule(ctx, deviceId, source); - if (ctx.invalidJson) { - return invalidJsonResult(ctx, deviceId); - } } catch (Exception legacyFailure) { logger.error("编译失败规则回退旧执行器后仍失败, eventId={}, ruleId={}", ctx.strSyslogUUID, source.getId(), legacyFailure); @@ -689,10 +688,13 @@ public class LogNormalProcessor { continue; } try { + if (!rule.isApplicableTo(ctx.strLogMsg)) { + continue; + } ParsedMessage parsed = ctx.messageParseContext.parse(rule, ctx.strLogMsg, ctx.executionMetrics); if (parsed == null && isJsonRule(source)) { markInvalidJson(ctx, source); - return invalidJsonResult(ctx, deviceId); + continue; } if (parsed == null || parsed.getFields().isEmpty()) { rejectParsedMessage(ctx, deviceId, source); @@ -718,11 +720,17 @@ public class LogNormalProcessor { throw new IllegalStateException("编译标准化规则处理失败", failure); } } + if (ctx.savedCount == 0 && ctx.filteredCount == 0 && ctx.invalidJson) { + return invalidJsonResult(ctx, deviceId); + } return finish(ctx); } /** 执行单条旧格式规则，供旧主路径和编译不兼容规则的精确回退共同使用。 */ private void executeLegacyRule(ProcessContext ctx, long deviceId, RuleSource source) { + if (isJsonRule(source) && !isJsonRuleApplicable(source, ctx.strLogMsg)) { + return; + } Map rule = source.asLegacyMap(); long parseStarted = System.nanoTime(); globalMetrics.ruleJsonParse(); @@ -773,14 +781,11 @@ public class LogNormalProcessor { ctx.rejectedCount++; } - /** - * JSON 语法失败是当前消息可确定的业务终态：只产生一条非标记录， - * 不再尝试后续规则，也不抛异常进入 Retry/DLT。 - */ + /** 所有适用规则均未形成有效候选后，只产生一条 JSON 非标记录。 */ private NormalizationResult invalidJsonResult(ProcessContext ctx, long deviceId) { RuleSource source = ctx.invalidJsonRule; ctx.syslogNonNormalMessage.setReason("JSON解析异常"); - ctx.syslogNonNormalMessage.setReasonDetail("发现非法JSON，已停止后续标准化规则处理, ruleId=" + ctx.syslogNonNormalMessage.setReasonDetail("所有适用JSON规则均解析失败，未形成可用候选, ruleId=" + (source == null ? "unknown" : source.getId()) + ", ruleName=" + (source == null ? "unknown" : source.getDisplayName())); if (!ctx.shadowMode) { @@ -803,6 +808,15 @@ public class LogNormalProcessor { return source != null && "json".equalsIgnoreCase(source.getDataType()); } + private boolean isJsonRuleApplicable(RuleSource source, String message) { + Cropper_paramsType cropper = getCropperParams(source.asLegacyMap()); + if (cropper == null || cropper.gethead_key() == null + || cropper.gethead_key().trim().isEmpty()) { + return true; + } + return message != null && message.contains(cropper.gethead_key()); + } + /** 现场坏 JSON 可能高频出现，WARN 每分钟最多一条，避免日志 I/O 反向拖慢消费。 */ private void logInvalidJsonRateLimited(ProcessContext ctx, long deviceId, RuleSource source) { invalidJsonSinceLastWarn.incrementAndGet(); @@ -813,7 +827,7 @@ public class LogNormalProcessor { return; } long count = invalidJsonSinceLastWarn.getAndSet(0L); - logger.warn("JSON解析失败已直接转非标: countSinceLastLog={}, eventId={}, deviceId={}, " + logger.warn("全部适用JSON候选解析失败已转非标: countSinceLastLog={}, eventId={}, deviceId={}, " + "ruleId={}, messageLength={}", count, ctx.strSyslogUUID, deviceId, source == null ? null : source.getId(), @@ -998,6 +1012,9 @@ public class LogNormalProcessor { Cropper_paramsType cropperParams = getCropperParams(dmNormalizeRule); if (dataType.equals("json")) { String strJson = logNormalData.ParserMessageJsonType(MsgContent, cropperParams); + if (cropperParams != null && !hasCompleteJsonObjectBoundary(strJson)) { + strJson = JsonParser.extractJsonFromLogMessage(strJson); + } LinkedHashMap result1 = NestedJsonParserUtil.safeParseJson(strJson); if (result1 == null) return null; LinkedHashMap flattened = NestedJsonUtils.flattenNestedJson(result1); @@ -1028,6 +1045,11 @@ public class LogNormalProcessor { return linkMap; } + private static boolean hasCompleteJsonObjectBoundary(String value) { + return value != null && value.length() >= 2 + && value.charAt(0) == '{' && value.charAt(value.length() - 1) == '}'; + } + public List> getRuleList(long device_id) { List> rulelst = dmNormalizeRuleService.selectByDeviceId((long) 1); if (rulelst == null) { diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/SysLogProcessor.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/SysLogProcessor.java index 05032be..045c09f 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/SysLogProcessor.java +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/SysLogProcessor.java @@ -9,6 +9,7 @@ import com.config.AppProperties; import com.config.KafkaConsumerProperties; import com.common.service.DmStandardizedBatchWriter; import com.common.service.RawLaneFieldLimiter; +import com.common.service.RawLogCaptureService; import com.common.service.DirectStatsProjectionWriter; import com.common.service.SyslogNonNormalMessageService; import com.common.service.SyslogNormalDataService; @@ -134,6 +135,7 @@ public class SysLogProcessor { @Autowired private BacklogSamplingDecider backlogSamplingDecider; @Autowired private BacklogSamplingAuditService backlogSamplingAuditService; @Autowired private SyslogNormalDataService syslogNormalDataService; + @Autowired(required = false) private RawLogCaptureService rawLogCaptureService; private final AtomicInteger totalProcessed = new AtomicInteger(0); private final AtomicLong nonNormalAsyncCompletedTotal = new AtomicLong(); @@ -620,6 +622,9 @@ public class SysLogProcessor { for (PreparedMessage message : messages) { RuntimeException dltFailure = tryRouteContinuityRecord( message.record, message.context, failure, stage); + captureFailure(message, stage, + dltFailure == null ? SyslogFailureRouter.TerminalState.DLT_PUBLISHED : null, + dltFailure == null ? failure : dltFailure); if (dltFailure != null) { recordContinuityLoss(messages.size(), published, message.record, failure, dltFailure, stage); @@ -884,9 +889,12 @@ public class SysLogProcessor { parsedEventTime.getLocalDateTime(), deviceId, collectId, deviceInfo.get("data_type"), SyslogParser.substringAfterFirstCloseBracket(decryptedMessage), decryptedMessage, deviceHeader, deviceInfo); - return new PreparedMessage(record, messageContext, decryptedMessage, eventId, rawPoint.getPoint(), + PreparedMessage prepared = new PreparedMessage(record, messageContext, decryptedMessage, + eventId, rawPoint.getPoint(), rawPoint.getBucket(), rawPoint.isV2(), parsedReceiveTime.getLocalDateTime(), deviceId, collectId, startTime, normalizationInput); + captureReceived(prepared); + return prepared; } /** @@ -948,7 +956,9 @@ public class SysLogProcessor { persisted.add(message); } catch (Exception singleFailure) { // 原文格式问题进入 DLT，但不能因此丢掉已经识别出的告警/标准化结果。 - routeFailure(message.record, message.context, singleFailure); + SyslogFailureRouter.TerminalState state = + routeFailure(message.record, message.context, singleFailure); + captureFailure(message, "RAW_STORAGE", state, singleFailure); persisted.add(message); } } @@ -1008,6 +1018,7 @@ public class SysLogProcessor { RuleExecutionMetrics executionMetrics) { LogNormalProcessor.NormalizationResult result = logNormalProcessor.prepare( message.normalizationInput, ruleSnapshot, enrichmentContext, executionMetrics); + captureNormalization(message, result); log.debug("消息标准化计算完成, offset: {}, rows={}, 耗时: {}ms", message.record.offset(), result.getRows().size(), System.currentTimeMillis() - message.startTime); return new PreparedNormalization(message, result, null); @@ -1063,8 +1074,10 @@ public class SysLogProcessor { results[index] = prepareNormalization(message, ruleSnapshot, enrichmentContext, executionMetrics); } catch (Exception failure) { - results[index] = new PreparedNormalization(message, null, - routeFailure(message.record, message.context, failure)); + SyslogFailureRouter.TerminalState state = + routeFailure(message.record, message.context, failure); + captureFailure(message, "RULE_PROCESSING", state, failure); + results[index] = new PreparedNormalization(message, null, state); } } } finally { @@ -1185,7 +1198,10 @@ public class SysLogProcessor { enterStage("DM_FAILURE_ROUTE"); for (Map.Entry failure : failures.entrySet()) { PreparedMessage message = failure.getKey(); - metrics.record(routeFailure(message.record, message.context, failure.getValue())); + SyslogFailureRouter.TerminalState state = + routeFailure(message.record, message.context, failure.getValue()); + captureFailure(message, "DM_WRITE", state, failure.getValue()); + metrics.record(state); } for (PreparedMessage message : plan.pendingDatabaseMessages) { if (!failedMessages.contains(message)) { @@ -1363,6 +1379,37 @@ public class SysLogProcessor { return state; } + private void captureReceived(PreparedMessage message) { + if (rawLogCaptureService == null || message == null) return; + rawLogCaptureService.observeReceived(message.startTime, message.deviceId, + message.deviceCollectId, message.eventId, captureTopic(message), + message.receivedAt, message.normalizationInput.getSyslogBody()); + } + + private void captureNormalization(PreparedMessage message, + LogNormalProcessor.NormalizationResult result) { + if (rawLogCaptureService == null || message == null) return; + rawLogCaptureService.observeNormalization(message.startTime, message.deviceId, + message.deviceCollectId, message.eventId, captureTopic(message), + message.receivedAt, message.normalizationInput.getSyslogBody(), result); + } + + private void captureFailure(PreparedMessage message, String stage, + SyslogFailureRouter.TerminalState state, Throwable failure) { + if (rawLogCaptureService == null || message == null) return; + rawLogCaptureService.observeFailure(message.startTime, message.deviceId, + message.deviceCollectId, message.eventId, captureTopic(message), + message.receivedAt, message.normalizationInput.getSyslogBody(), stage, + state == null ? null : state.name(), failure); + } + + private String captureTopic(PreparedMessage message) { + if (message.context != null && message.context.getOriginalTopic() != null) { + return message.context.getOriginalTopic(); + } + return message.record == null ? null : message.record.topic(); + } + /** 已解密、已构造原文点，但尚未提交 Kafka offset 的批内消息。 */ static final class PreparedMessage { private final ConsumerRecord record; diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledRule.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledRule.java index 3c674ea..957c6a5 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledRule.java +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledRule.java @@ -1,6 +1,7 @@ package com.Modules.NormalData.compiled; import com.Modules.NormalData.logNormalData; +import com.common.util.JsonParser; import com.common.util.NestedJsonParserUtil; import com.common.util.NestedJsonUtils; @@ -85,6 +86,16 @@ public final class CompiledRule { LinkedHashMap parsed; switch (dataType) { case "json": + // 历史 JSON 规则的空 tail cropper 使用“head 起点 + offset”语义，部分规则 + // 因而会保留诸如 "7|!alarm|!" 的前缀。只在 JSON 专用路径重新收敛到 + // 完整对象边界，避免改变 KV/SEP/REGEX 及既有 offset 的全局兼容语义。 + if (cropper != null && !hasCompleteObjectBoundary(cropped)) { + try { + cropped = JsonParser.extractJsonFromLogMessage(cropped); + } catch (IllegalArgumentException incompleteJson) { + return null; + } + } LinkedHashMap nested = NestedJsonParserUtil.safeParseJson(cropped); parsed = nested == null ? null : NestedJsonUtils.flattenNestedJson(nested); break; @@ -106,6 +117,22 @@ public final class CompiledRule { return parsed == null ? null : new ParsedMessage(parsed); } + /** + * cropper 的 head_key 是规则格式指纹。消息中不存在该指纹时表示规则不适用， + * 不能把空截取结果误判为原始消息 JSON 非法。 + */ + public boolean isApplicableTo(String message) { + if (!"json".equals(dataType) || cropper == null || empty(cropper.getHeadKey())) { + return true; + } + return message != null && message.contains(cropper.getHeadKey()); + } + + private static boolean hasCompleteObjectBoundary(String value) { + return value != null && value.length() >= 2 + && value.charAt(0) == '{' && value.charAt(value.length() - 1) == '}'; + } + public Map map(ParsedMessage message) { Map destination = new LinkedHashMap<>(); for (Map.Entry source : message.getFields().entrySet()) { diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/RawLogCaptureClusterStore.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/RawLogCaptureClusterStore.java new file mode 100644 index 0000000..744bb1e --- /dev/null +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/RawLogCaptureClusterStore.java @@ -0,0 +1,304 @@ +package com.common.service; + +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.ObjectMapper; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.stereotype.Component; + +import java.time.LocalDateTime; +import java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; + +/** Redis只保存每个消费实例的有界候选集，用于多实例汇聚，不承载逐条日志流。 */ +@Slf4j +@Component +public class RawLogCaptureClusterStore { + private static final String PREFIX = "hb:xdr:raw-capture:"; + private static final String ACTIVE_KEY = PREFIX + "active"; + private static final TypeReference> MAP_TYPE = + new TypeReference>() { }; + + private final StringRedisTemplate redis; + private final ObjectMapper mapper; + private final String nodeId; + private final AtomicLong lastFailureLogAt = new AtomicLong(); + + @Value("${syslog.raw-capture.cluster-activation-delay-ms:1000}") + private long activationDelayMs = 1000L; + + public RawLogCaptureClusterStore(StringRedisTemplate redis, ObjectMapper mapper, + @Value("${syslog.raw-capture.node-id:${HOSTNAME:unknown}}") + String configuredNodeId) { + this.redis = redis; + this.mapper = mapper; + this.nodeId = "unknown".equals(configuredNodeId) + ? UUID.randomUUID().toString() : configuredNodeId; + } + + Definition create(long deviceId, int durationSeconds, int settleSeconds, int messageCount, + int maxActiveSessions, int retentionMinutes) { + try { + long now = System.currentTimeMillis(); + redis.opsForZSet().removeRangeByScore(ACTIVE_KEY, 0D, now - 1D); + Long active = redis.opsForZSet().zCard(ACTIVE_KEY); + if (active != null && active >= maxActiveSessions) { + throw new RawLogCaptureService.CaptureCapacityException(maxActiveSessions); + } + long startedAt = now + Math.max(200L, activationDelayMs); + Definition definition = new Definition(UUID.randomUUID().toString(), deviceId, + startedAt, startedAt + durationSeconds * 1000L, + startedAt + (durationSeconds + settleSeconds) * 1000L, messageCount, + retentionMinutes); + String metaKey = metaKey(definition.id); + redis.opsForHash().putAll(metaKey, definition.toHash()); + redis.expire(metaKey, definition.ttlSeconds(now), TimeUnit.SECONDS); + redis.opsForZSet().add(ACTIVE_KEY, definition.id, definition.settleEndsAt); + return definition; + } catch (RawLogCaptureService.CaptureCapacityException failure) { + throw failure; + } catch (RuntimeException failure) { + throw new RawLogCaptureService.CaptureCoordinationException(failure); + } + } + + List active(long now) { + redis.opsForZSet().removeRangeByScore(ACTIVE_KEY, 0D, now - 1D); + Set ids = redis.opsForZSet().rangeByScore(ACTIVE_KEY, now, Double.MAX_VALUE); + if (ids == null || ids.isEmpty()) return Collections.emptyList(); + List result = new ArrayList<>(ids.size()); + for (String id : ids) { + Definition definition = definition(id); + if (definition != null && definition.stoppedAt == 0L) result.add(definition); + } + return result; + } + + void publish(RawLogCaptureService.CaptureCandidates candidates) { + try { + Definition definition = definition(candidates.getSessionId()); + if (definition == null || definition.stoppedAt > 0L) return; + String snapshotKey = snapshotKey(candidates.getSessionId(), nodeId); + redis.opsForValue().set(snapshotKey, mapper.writeValueAsString(toMap(candidates)), + definition.ttlSeconds(System.currentTimeMillis()), TimeUnit.SECONDS); + String nodesKey = nodesKey(candidates.getSessionId()); + redis.opsForSet().add(nodesKey, nodeId); + redis.expire(nodesKey, definition.ttlSeconds(System.currentTimeMillis()), TimeUnit.SECONDS); + } catch (Exception failure) { + throw new RawLogCaptureService.CaptureCoordinationException(failure); + } + } + + RawLogCaptureService.CaptureSnapshot get(String sessionId) { + try { + Definition definition = definition(sessionId); + if (definition == null) return null; + Set nodes = redis.opsForSet().members(nodesKey(sessionId)); + List first = new ArrayList<>(); + List anomalies = new ArrayList<>(); + long received = 0L, processed = 0L, observedAnomalies = 0L; + if (nodes != null) { + for (String node : nodes) { + String json = redis.opsForValue().get(snapshotKey(sessionId, node)); + if (json == null) continue; + Map candidate = mapper.readValue(json, MAP_TYPE); + received += longValue(candidate.get("receivedCount")); + processed += longValue(candidate.get("processedCount")); + observedAnomalies += longValue(candidate.get("observedAnomalyCount")); + first.addAll(messages(candidate.get("firstMessages"))); + anomalies.addAll(messages(candidate.get("firstAnomalies"))); + } + } + List selected = + RawLogCaptureService.mergeCandidates(first, anomalies, definition.messageCount); + int retainedAnomalies = 0; + for (RawLogCaptureService.CapturedLogSnapshot message : selected) { + if (message.isAnomaly()) retainedAnomalies++; + } + long replacements = replacementCount(first, anomalies, definition.messageCount); + long now = System.currentTimeMillis(); + return new RawLogCaptureService.CaptureSnapshot(definition.id, definition.deviceId, + definition.status(now), RawLogCaptureService.instant(definition.startedAt), + RawLogCaptureService.instant(definition.captureEndsAt), + RawLogCaptureService.instant(definition.settleEndsAt), definition.messageCount, + received, processed, observedAnomalies, retainedAnomalies, replacements, selected); + } catch (RawLogCaptureService.CaptureCoordinationException failure) { + throw failure; + } catch (Exception failure) { + throw new RawLogCaptureService.CaptureCoordinationException(failure); + } + } + + RawLogCaptureService.CaptureSnapshot stop(String sessionId) { + try { + Definition definition = definition(sessionId); + if (definition == null) return null; + redis.opsForHash().put(metaKey(sessionId), "stoppedAt", + String.valueOf(System.currentTimeMillis())); + redis.opsForZSet().remove(ACTIVE_KEY, sessionId); + return get(sessionId); + } catch (RawLogCaptureService.CaptureCoordinationException failure) { + throw failure; + } catch (RuntimeException failure) { + throw new RawLogCaptureService.CaptureCoordinationException(failure); + } + } + + void recordSyncFailure(RuntimeException failure) { + long now = System.currentTimeMillis(), previous = lastFailureLogAt.get(); + if (now - previous >= 30_000L && lastFailureLogAt.compareAndSet(previous, now)) { + log.warn("原始日志采样跨实例同步失败；不影响主消费链路: {}", failure.getMessage()); + } + } + + private Definition definition(String id) { + Map values = redis.opsForHash().entries(metaKey(id)); + return values == null || values.isEmpty() ? null : Definition.from(id, values); + } + + private Map toMap(RawLogCaptureService.CaptureCandidates candidates) { + Map value = new LinkedHashMap<>(); + value.put("receivedCount", candidates.getReceivedCount()); + value.put("processedCount", candidates.getProcessedCount()); + value.put("observedAnomalyCount", candidates.getObservedAnomalyCount()); + value.put("firstMessages", candidates.getFirstMessages()); + value.put("firstAnomalies", candidates.getFirstAnomalies()); + return value; + } + + @SuppressWarnings("unchecked") + private List messages(Object value) { + if (!(value instanceof List)) return Collections.emptyList(); + List result = new ArrayList<>(); + for (Object item : (List) value) { + if (!(item instanceof Map)) continue; + Map map = (Map) item; + Object receivedAt = map.get("receivedAt"); + result.add(new RawLogCaptureService.CapturedLogSnapshot( + longValue(map.get("sequence")), string(map.get("eventId")), + nullableLong(map.get("deviceId")), nullableLong(map.get("deviceCollectId")), + string(map.get("topic")), localDateTime(receivedAt), string(map.get("observedAt")), + string(map.get("processingStatus")), booleanValue(map.get("anomaly")), + string(map.get("reason")), string(map.get("reasonDetail")), + string(map.get("processedAt")), string(map.get("rawMessage")))); + } + return result; + } + + private long replacementCount(List first, + List anomalies, int limit) { + int finalAnomalies = 0, initialAnomalies = 0; + for (RawLogCaptureService.CapturedLogSnapshot value + : RawLogCaptureService.mergeCandidates(first, anomalies, limit)) { + if (value.isAnomaly()) finalAnomalies++; + } + List initial = + RawLogCaptureService.mergeCandidates(first, Collections.emptyList(), limit); + for (RawLogCaptureService.CapturedLogSnapshot value : initial) { + if (value.isAnomaly()) initialAnomalies++; + } + return Math.max(0, finalAnomalies - initialAnomalies); + } + + private static String metaKey(String id) { return PREFIX + "{" + id + "}:meta"; } + private static String nodesKey(String id) { return PREFIX + "{" + id + "}:nodes"; } + private static String snapshotKey(String id, String node) { + return PREFIX + "{" + id + "}:node:" + node; + } + private static String string(Object value) { return value == null ? null : String.valueOf(value); } + private static long longValue(Object value) { + return value == null ? 0L : Long.parseLong(String.valueOf(value)); + } + private static Long nullableLong(Object value) { + return value == null ? null : Long.valueOf(String.valueOf(value)); + } + private static boolean booleanValue(Object value) { + return value != null && Boolean.parseBoolean(String.valueOf(value)); + } + private static LocalDateTime localDateTime(Object value) { + if (value == null) return null; + if (value instanceof List) { + List parts = (List) value; + if (parts.size() >= 6) { + int nano = parts.size() >= 7 ? ((Number) parts.get(6)).intValue() : 0; + return LocalDateTime.of(((Number) parts.get(0)).intValue(), + ((Number) parts.get(1)).intValue(), ((Number) parts.get(2)).intValue(), + ((Number) parts.get(3)).intValue(), ((Number) parts.get(4)).intValue(), + ((Number) parts.get(5)).intValue(), nano); + } + } + return LocalDateTime.parse(String.valueOf(value)); + } + + static final class Definition { + private final String id; + private final long deviceId; + private final long startedAt; + private final long captureEndsAt; + private final long settleEndsAt; + private final int messageCount; + private final int retentionMinutes; + private final long stoppedAt; + + Definition(String id, long deviceId, long startedAt, long captureEndsAt, + long settleEndsAt, int messageCount, int retentionMinutes) { + this(id, deviceId, startedAt, captureEndsAt, settleEndsAt, messageCount, + retentionMinutes, 0L); + } + + private Definition(String id, long deviceId, long startedAt, long captureEndsAt, + long settleEndsAt, int messageCount, int retentionMinutes, + long stoppedAt) { + this.id = id; this.deviceId = deviceId; this.startedAt = startedAt; + this.captureEndsAt = captureEndsAt; this.settleEndsAt = settleEndsAt; + this.messageCount = messageCount; this.retentionMinutes = retentionMinutes; + this.stoppedAt = stoppedAt; + } + + private Map toHash() { + Map value = new LinkedHashMap<>(); + value.put("deviceId", String.valueOf(deviceId)); + value.put("startedAt", String.valueOf(startedAt)); + value.put("captureEndsAt", String.valueOf(captureEndsAt)); + value.put("settleEndsAt", String.valueOf(settleEndsAt)); + value.put("messageCount", String.valueOf(messageCount)); + value.put("retentionMinutes", String.valueOf(retentionMinutes)); + value.put("stoppedAt", String.valueOf(stoppedAt)); + return value; + } + + private static Definition from(String id, Map map) { + return new Definition(id, longValue(map.get("deviceId")), longValue(map.get("startedAt")), + longValue(map.get("captureEndsAt")), longValue(map.get("settleEndsAt")), + (int) longValue(map.get("messageCount")), + (int) longValue(map.get("retentionMinutes")), longValue(map.get("stoppedAt"))); + } + + private long ttlSeconds(long now) { + long remainingMillis = Math.max(0L, settleEndsAt - now); + return Math.max(60L, (remainingMillis + 999L) / 1000L + + retentionMinutes * 60L); + } + private String status(long now) { + if (stoppedAt > 0L) return "CANCELLED"; + if (now < startedAt) return "SCHEDULED"; + if (now <= captureEndsAt) return "RUNNING"; + if (now <= settleEndsAt) return "DRAINING"; + return "COMPLETED"; + } + String getId() { return id; } + long getDeviceId() { return deviceId; } + long getStartedAt() { return startedAt; } + long getCaptureEndsAt() { return captureEndsAt; } + long getSettleEndsAt() { return settleEndsAt; } + int getMessageCount() { return messageCount; } + } +} diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/RawLogCaptureService.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/RawLogCaptureService.java new file mode 100644 index 0000000..2da6470 --- /dev/null +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/RawLogCaptureService.java @@ -0,0 +1,798 @@ +package com.common.service; + +import com.Modules.NormalData.LogNormalProcessor; +import com.common.entity.SyslogNonNormalMessage; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.http.HttpStatus; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Service; +import org.springframework.web.bind.annotation.ResponseStatus; + +import javax.annotation.PostConstruct; +import java.time.Clock; +import java.time.Instant; +import java.time.LocalDateTime; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.atomic.AtomicLong; + +/** + * 面向现场诊断的有界原始日志采样器。 + * + *

普通消息按到达顺序填充固定容量；容量满后，新的异常消息会淘汰最早的非异常消息。 + * 异常消息之间仍保持先到先得，避免高流量设备导致内存随消息数增长。

+ */ +@Service +public class RawLogCaptureService { + + public static final int MAX_DURATION_SECONDS = 600; + public static final int MAX_MESSAGE_COUNT = 50; + + @Value("${syslog.raw-capture.enabled:true}") + private boolean enabled = true; + + @Value("${syslog.raw-capture.max-active-sessions:20}") + private int maxActiveSessions = 20; + + @Value("${syslog.raw-capture.max-retained-sessions:100}") + private int maxRetainedSessions = 100; + + @Value("${syslog.raw-capture.settle-seconds:30}") + private int settleSeconds = 30; + + @Value("${syslog.raw-capture.result-retention-minutes:30}") + private int resultRetentionMinutes = 30; + + private final Clock clock; + private final ConcurrentHashMap sessions = new ConcurrentHashMap<>(); + private final ConcurrentHashMap> sessionsByDevice = + new ConcurrentHashMap<>(); + private final AtomicLong sequence = new AtomicLong(); + + /** 跨实例同步由独立存储组件完成；单元测试或明确关闭时退化为单实例模式。 */ + @Autowired(required = false) + private RawLogCaptureClusterStore clusterStore; + + @Value("${syslog.raw-capture.cluster-enabled:true}") + private boolean clusterEnabled = true; + + public RawLogCaptureService() { + this(Clock.systemUTC()); + } + + RawLogCaptureService(Clock clock) { + this.clock = clock; + } + + @PostConstruct + public void validateConfiguration() { + if (maxActiveSessions < 1 || maxActiveSessions > 1000) { + throw new IllegalArgumentException("syslog.raw-capture.max-active-sessions 必须在 1~1000 之间"); + } + if (maxRetainedSessions < maxActiveSessions || maxRetainedSessions > 5000) { + throw new IllegalArgumentException( + "syslog.raw-capture.max-retained-sessions 必须不小于活动会话上限且不超过 5000"); + } + if (settleSeconds < 0 || settleSeconds > 300) { + throw new IllegalArgumentException("syslog.raw-capture.settle-seconds 必须在 0~300 之间"); + } + if (resultRetentionMinutes < 1 || resultRetentionMinutes > 1440) { + throw new IllegalArgumentException("syslog.raw-capture.result-retention-minutes 必须在 1~1440 之间"); + } + } + + public CaptureSnapshot start(long deviceId, int durationSeconds, int messageCount) { + if (!enabled) throw new CaptureDisabledException(); + if (deviceId <= 0L) throw new IllegalArgumentException("deviceId 必须大于 0"); + if (durationSeconds < 1 || durationSeconds > MAX_DURATION_SECONDS) { + throw new IllegalArgumentException("durationSeconds 必须在 1~600 之间"); + } + if (messageCount < 1 || messageCount > MAX_MESSAGE_COUNT) { + throw new IllegalArgumentException("messageCount 必须在 1~50 之间"); + } + if (useClusterStore()) { + RawLogCaptureClusterStore.Definition definition = clusterStore.create(deviceId, + durationSeconds, settleSeconds, messageCount, maxActiveSessions, + resultRetentionMinutes); + ensureClusterSession(definition); + return clusterStore.get(definition.getId()); + } + cleanup(clock.millis()); + synchronized (this) { + long startedAt = clock.millis(); + if (activeSessionCount(startedAt) >= maxActiveSessions) { + throw new CaptureCapacityException(maxActiveSessions); + } + makeRoomForRetainedSession(startedAt); + CaptureSession session = new CaptureSession(UUID.randomUUID().toString(), deviceId, + startedAt, startedAt + durationSeconds * 1000L, + startedAt + (durationSeconds + settleSeconds) * 1000L, messageCount); + sessions.put(session.id, session); + sessionsByDevice.computeIfAbsent(deviceId, ignored -> new CopyOnWriteArrayList<>()) + .add(session); + return session.snapshot(startedAt); + } + } + + public CaptureSnapshot get(String sessionId) { + if (useClusterStore()) return clusterStore.get(sessionId); + long now = clock.millis(); + cleanup(now); + CaptureSession session = sessions.get(sessionId); + return session == null ? null : session.snapshot(now); + } + + public CaptureSnapshot stop(String sessionId) { + if (useClusterStore()) { + CaptureSnapshot snapshot = clusterStore.stop(sessionId); + CaptureSession local = sessions.get(sessionId); + if (local != null) local.stop(clock.millis()); + if (local != null) removeFromActiveIndex(local); + return snapshot; + } + long now = clock.millis(); + CaptureSession session = sessions.get(sessionId); + if (session == null) return null; + session.stop(now); + removeFromActiveIndex(session); + return session.snapshot(now); + } + + /** 在解密并识别设备后调用；只保存固定容量的最早样本。 */ + public void observeReceived(long observedAtEpochMillis, Long deviceId, Long deviceCollectId, + String eventId, String topic, LocalDateTime receivedAt, + String rawMessage) { + if (!enabled || deviceId == null) return; + List active = sessionsByDevice.get(deviceId); + if (active == null || active.isEmpty()) return; + ReceivedLog received = new ReceivedLog(orderSequence(observedAtEpochMillis), observedAtEpochMillis, + deviceId, deviceCollectId, eventId, topic, receivedAt, rawMessage); + for (CaptureSession session : active) session.received(received); + } + + /** 记录规则处理结果；非标结果具有异常优先级。 */ + public void observeNormalization(long observedAtEpochMillis, Long deviceId, Long deviceCollectId, + String eventId, String topic, LocalDateTime receivedAt, + String rawMessage, + LogNormalProcessor.NormalizationResult result) { + if (result == null) return; + if (!result.getNonNormalMessages().isEmpty()) { + SyslogNonNormalMessage first = result.getNonNormalMessages().get(0); + observeOutcome(observedAtEpochMillis, deviceId, deviceCollectId, eventId, topic, + receivedAt, rawMessage, "NON_NORMAL", true, + first.getReason(), first.getReasonDetail()); + } else if (result.getOutcome() == LogNormalProcessor.ProcessingOutcome.FILTERED) { + observeOutcome(observedAtEpochMillis, deviceId, deviceCollectId, eventId, topic, + receivedAt, rawMessage, "FILTERED", false, "规则过滤", null); + } else { + observeOutcome(observedAtEpochMillis, deviceId, deviceCollectId, eventId, topic, + receivedAt, rawMessage, "SUCCESS", false, null, null); + } + } + + /** 记录 Retry/DLT、规则执行或持久化异常；异常可替换已有普通样本。 */ + public void observeFailure(long observedAtEpochMillis, Long deviceId, Long deviceCollectId, + String eventId, String topic, LocalDateTime receivedAt, + String rawMessage, String stage, String terminalState, + Throwable failure) { + String status = terminalState == null || terminalState.trim().isEmpty() + ? "ERROR" : terminalState; + String reason = stage == null || stage.trim().isEmpty() ? "处理异常" : stage; + String detail = failure == null ? null : failure.getClass().getName() + + (failure.getMessage() == null ? "" : ": " + failure.getMessage()); + observeOutcome(observedAtEpochMillis, deviceId, deviceCollectId, eventId, topic, + receivedAt, rawMessage, status, true, reason, abbreviate(detail, 2000)); + } + + private void observeOutcome(long observedAtEpochMillis, Long deviceId, Long deviceCollectId, + String eventId, String topic, LocalDateTime receivedAt, + String rawMessage, String status, boolean anomaly, + String reason, String detail) { + if (!enabled || deviceId == null) return; + List active = sessionsByDevice.get(deviceId); + if (active == null || active.isEmpty()) return; + ReceivedLog received = new ReceivedLog(orderSequence(observedAtEpochMillis), observedAtEpochMillis, + deviceId, deviceCollectId, eventId, topic, receivedAt, rawMessage); + long now = clock.millis(); + for (CaptureSession session : active) { + session.outcome(received, status, anomaly, reason, detail, now); + } + } + + /** + * 多实例每200ms同步一次会话定义和变化过的有界候选集；单实例降级模式用同一定时器 + * 主动清理过期结果。API创建时预留1秒激活时间，避免轮询传播丢掉窗口开头的日志。 + */ + @Scheduled(fixedDelayString = "${syslog.raw-capture.cluster-sync-interval-ms:200}") + public void synchronizeClusterSessions() { + if (!enabled) return; + if (!useClusterStore()) { + cleanup(clock.millis()); + return; + } + try { + long now = clock.millis(); + List definitions = clusterStore.active(now); + Map activeIds = new LinkedHashMap<>(); + for (RawLogCaptureClusterStore.Definition definition : definitions) { + activeIds.put(definition.getId(), Boolean.TRUE); + CaptureSession session = ensureClusterSession(definition); + CaptureCandidates candidates = session.candidatesIfChanged(); + if (candidates != null) { + clusterStore.publish(candidates); + session.markPublished(candidates.getRevision()); + } + } + for (CaptureSession session : sessions.values()) { + if (session.clusterSession && !activeIds.containsKey(session.id)) { + clusterStore.publish(session.finalCandidates()); + session.stop(now); + removeFromActiveIndex(session); + sessions.remove(session.id, session); + } + } + } catch (RuntimeException failure) { + clusterStore.recordSyncFailure(failure); + } + } + + private boolean useClusterStore() { + return clusterEnabled && clusterStore != null; + } + + private CaptureSession ensureClusterSession(RawLogCaptureClusterStore.Definition definition) { + CaptureSession existing = sessions.get(definition.getId()); + if (existing != null) return existing; + CaptureSession created = new CaptureSession(definition.getId(), definition.getDeviceId(), + definition.getStartedAt(), definition.getCaptureEndsAt(), + definition.getSettleEndsAt(), definition.getMessageCount(), true); + CaptureSession winner = sessions.putIfAbsent(created.id, created); + CaptureSession session = winner == null ? created : winner; + if (winner == null) { + sessionsByDevice.computeIfAbsent(created.deviceId, + ignored -> new CopyOnWriteArrayList<>()).add(created); + } + return session; + } + + private synchronized int activeSessionCount(long now) { + int count = 0; + for (CaptureSession session : sessions.values()) { + if (session.isActive(now)) count++; + } + return count; + } + + /** 总结果数满时优先淘汰最早完成的会话，活动会话绝不被淘汰。 */ + private void makeRoomForRetainedSession(long now) { + while (sessions.size() >= maxRetainedSessions) { + CaptureSession oldestFinished = null; + for (CaptureSession session : sessions.values()) { + if (session.isActive(now)) continue; + if (oldestFinished == null || session.finishedAt() < oldestFinished.finishedAt()) { + oldestFinished = session; + } + } + if (oldestFinished == null) { + throw new CaptureCapacityException(maxRetainedSessions); + } + sessions.remove(oldestFinished.id, oldestFinished); + removeFromActiveIndex(oldestFinished); + } + } + + private void cleanup(long now) { + for (CaptureSession session : sessions.values()) { + if (!session.isActive(now)) removeFromActiveIndex(session); + if (session.removableAt(resultRetentionMinutes) < now) sessions.remove(session.id, session); + } + } + + private void removeFromActiveIndex(CaptureSession session) { + CopyOnWriteArrayList active = sessionsByDevice.get(session.deviceId); + if (active == null) return; + active.remove(session); + if (active.isEmpty()) sessionsByDevice.remove(session.deviceId, active); + } + + private static String abbreviate(String value, int maxChars) { + if (value == null || value.length() <= maxChars) return value; + return value.substring(0, maxChars) + "...[truncated]"; + } + + static String instant(long epochMillis) { + return Instant.ofEpochMilli(epochMillis).toString(); + } + + private long orderSequence(long observedAtEpochMillis) { + return observedAtEpochMillis * 1_000_000L + sequence.incrementAndGet() % 1_000_000L; + } + + static List mergeCandidates(List firstCandidates, + List anomalyCandidates, + int limit) { + Map uniqueFirst = uniqueByEvent(firstCandidates); + List first = new ArrayList<>(uniqueFirst.values()); + first.sort(Comparator.comparingLong(CapturedLogSnapshot::getSequence)); + if (first.size() > limit) first = new ArrayList<>(first.subList(0, limit)); + + Map uniqueAnomalies = uniqueByEvent(anomalyCandidates); + List anomalies = new ArrayList<>(uniqueAnomalies.values()); + anomalies.sort(Comparator.comparingLong(CapturedLogSnapshot::getSequence)); + if (anomalies.size() > limit) anomalies = anomalies.subList(0, limit); + + Map selected = uniqueByEvent(first); + for (CapturedLogSnapshot anomaly : anomalies) { + String key = eventKey(anomaly); + if (selected.containsKey(key)) { + selected.put(key, anomaly); + continue; + } + if (selected.size() < limit) { + selected.put(key, anomaly); + continue; + } + String oldestNormal = null; + long oldestSequence = Long.MAX_VALUE; + for (Map.Entry entry : selected.entrySet()) { + CapturedLogSnapshot candidate = entry.getValue(); + if (!candidate.isAnomaly() && candidate.getSequence() < oldestSequence) { + oldestNormal = entry.getKey(); + oldestSequence = candidate.getSequence(); + } + } + if (oldestNormal == null) break; + selected.remove(oldestNormal); + selected.put(key, anomaly); + } + List result = new ArrayList<>(selected.values()); + result.sort(Comparator.comparingLong(CapturedLogSnapshot::getSequence)); + return result; + } + + private static Map uniqueByEvent( + List candidates) { + Map result = new LinkedHashMap<>(); + if (candidates == null) return result; + for (CapturedLogSnapshot candidate : candidates) { + String key = eventKey(candidate); + CapturedLogSnapshot existing = result.get(key); + if (existing == null || (!existing.isAnomaly() && candidate.isAnomaly())) { + result.put(key, candidate); + } + } + return result; + } + + private static String eventKey(CapturedLogSnapshot candidate) { + return candidate.getEventId() == null + ? "sequence:" + candidate.getSequence() : "event:" + candidate.getEventId(); + } + + private static List snapshots(List logs) { + List result = new ArrayList<>(logs.size()); + for (CapturedLog log : logs) result.add(log.snapshot()); + return result; + } + + private static int countAnomalies(List messages) { + int count = 0; + for (CapturedLogSnapshot message : messages) if (message.isAnomaly()) count++; + return count; + } + + private static long replacementCount(List first, + List anomalies, int limit) { + Map firstSet = uniqueByEvent(first); + int laterAnomalies = 0; + for (CapturedLogSnapshot anomaly : uniqueByEvent(anomalies).values()) { + if (!firstSet.containsKey(eventKey(anomaly))) laterAnomalies++; + } + int availableNormals = 0; + List globalFirst = new ArrayList<>(firstSet.values()); + globalFirst.sort(Comparator.comparingLong(CapturedLogSnapshot::getSequence)); + for (int i = 0; i < Math.min(limit, globalFirst.size()); i++) { + if (!globalFirst.get(i).isAnomaly()) availableNormals++; + } + return Math.min(laterAnomalies, availableNormals); + } + + private static final class ReceivedLog { + private final long sequence; + private final long observedAt; + private final Long deviceId; + private final Long deviceCollectId; + private final String eventId; + private final String topic; + private final LocalDateTime receivedAt; + private final String rawMessage; + + private ReceivedLog(long sequence, long observedAt, Long deviceId, Long deviceCollectId, + String eventId, String topic, LocalDateTime receivedAt, String rawMessage) { + this.sequence = sequence; + this.observedAt = observedAt; + this.deviceId = deviceId; + this.deviceCollectId = deviceCollectId; + this.eventId = eventId; + this.topic = topic; + this.receivedAt = receivedAt; + this.rawMessage = rawMessage; + } + } + + private static final class CaptureSession { + private final String id; + private final long deviceId; + private final long startedAt; + private final long captureEndsAt; + private final long settleEndsAt; + private final int messageCount; + private final List firstMessages = new ArrayList<>(); + private final List firstAnomalies = new ArrayList<>(); + private final Map firstByEventId = new LinkedHashMap<>(); + private final Map anomalyByEventId = new LinkedHashMap<>(); + private final boolean clusterSession; + private long revision; + private long publishedRevision = -1L; + private long stoppedAt; + private long receivedCount; + private long processedCount; + private long observedAnomalyCount; + private long replacementCount; + + private CaptureSession(String id, long deviceId, long startedAt, long captureEndsAt, + long settleEndsAt, int messageCount) { + this(id, deviceId, startedAt, captureEndsAt, settleEndsAt, messageCount, false); + } + + private CaptureSession(String id, long deviceId, long startedAt, long captureEndsAt, + long settleEndsAt, int messageCount, boolean clusterSession) { + this.id = id; + this.deviceId = deviceId; + this.startedAt = startedAt; + this.captureEndsAt = captureEndsAt; + this.settleEndsAt = settleEndsAt; + this.messageCount = messageCount; + this.clusterSession = clusterSession; + } + + private synchronized void received(ReceivedLog log) { + if (!accepts(log.observedAt) || stoppedAt > 0L) return; + receivedCount++; + if (findFirst(log.eventId) != null || firstMessages.size() >= messageCount) return; + CapturedLog captured = new CapturedLog(log); + firstMessages.add(captured); + if (captured.eventId != null) firstByEventId.put(captured.eventId, captured); + revision++; + } + + private synchronized void outcome(ReceivedLog log, String status, boolean anomaly, + String reason, String detail, long now) { + if (!accepts(log.observedAt) || stoppedAt > 0L || now > settleEndsAt) return; + processedCount++; + if (anomaly) observedAnomalyCount++; + CapturedLog first = findFirst(log.eventId); + if (first != null) { + first.update(status, anomaly, reason, detail, now); + revision++; + } + CapturedLog existingAnomaly = findAnomaly(log.eventId); + if (existingAnomaly != null) { + existingAnomaly.update(status, anomaly, reason, detail, now); + revision++; + } + if (!anomaly || existingAnomaly != null || firstAnomalies.size() >= messageCount) return; + CapturedLog captured = first == null ? new CapturedLog(log) : first.copy(); + captured.update(status, true, reason, detail, now); + firstAnomalies.add(captured); + if (captured.eventId != null) anomalyByEventId.put(captured.eventId, captured); + revision++; + } + + private boolean accepts(long observedAt) { + return observedAt >= startedAt && observedAt <= captureEndsAt; + } + + private CapturedLog findFirst(String eventId) { + return eventId == null ? null : firstByEventId.get(eventId); + } + + private CapturedLog findAnomaly(String eventId) { + return eventId == null ? null : anomalyByEventId.get(eventId); + } + + private synchronized void stop(long now) { + if (stoppedAt == 0L) stoppedAt = now; + } + + private synchronized boolean isActive(long now) { + return stoppedAt == 0L && now <= settleEndsAt; + } + + private synchronized long removableAt(int retentionMinutes) { + return finishedAt() + retentionMinutes * 60_000L; + } + + private synchronized long finishedAt() { + return stoppedAt > 0L ? stoppedAt : settleEndsAt; + } + + private synchronized CaptureSnapshot snapshot(long now) { + List first = snapshots(firstMessages); + List anomalies = snapshots(firstAnomalies); + List logs = mergeCandidates(first, anomalies, messageCount); + int retainedAnomalies = countAnomalies(logs); + replacementCount = replacementCount(first, anomalies, messageCount); + return new CaptureSnapshot(id, deviceId, status(now), instant(startedAt), + instant(captureEndsAt), instant(settleEndsAt), messageCount, + receivedCount, processedCount, observedAnomalyCount, + retainedAnomalies, replacementCount, logs); + } + + private synchronized CaptureCandidates candidatesIfChanged() { + if (!clusterSession || revision == publishedRevision) return null; + return candidates(); + } + + private synchronized CaptureCandidates finalCandidates() { + return candidates(); + } + + private CaptureCandidates candidates() { + return new CaptureCandidates(id, deviceId, startedAt, captureEndsAt, settleEndsAt, + messageCount, receivedCount, processedCount, observedAnomalyCount, + snapshots(firstMessages), snapshots(firstAnomalies), revision); + } + + private synchronized void markPublished(long candidateRevision) { + if (candidateRevision > publishedRevision) publishedRevision = candidateRevision; + } + + private String status(long now) { + if (stoppedAt > 0L) return "CANCELLED"; + if (now < startedAt) return "SCHEDULED"; + if (now <= captureEndsAt) return "RUNNING"; + if (now <= settleEndsAt) return "DRAINING"; + return "COMPLETED"; + } + } + + private static final class CapturedLog { + private final long sequence; + private final String eventId; + private final Long deviceId; + private final Long deviceCollectId; + private final String topic; + private final LocalDateTime receivedAt; + private final String observedAt; + private final String rawMessage; + private String processingStatus = "PENDING"; + private boolean anomaly; + private String reason; + private String reasonDetail; + private String processedAt; + + private CapturedLog(ReceivedLog log) { + this.sequence = log.sequence; + this.eventId = log.eventId; + this.deviceId = log.deviceId; + this.deviceCollectId = log.deviceCollectId; + this.topic = log.topic; + this.receivedAt = log.receivedAt; + this.observedAt = instant(log.observedAt); + this.rawMessage = log.rawMessage; + } + + private CapturedLog copy() { + return new CapturedLog(sequence, eventId, deviceId, deviceCollectId, topic, + receivedAt, observedAt, rawMessage, processingStatus, anomaly, + reason, reasonDetail, processedAt); + } + + private CapturedLog(long sequence, String eventId, Long deviceId, Long deviceCollectId, + String topic, LocalDateTime receivedAt, String observedAt, + String rawMessage, String processingStatus, boolean anomaly, + String reason, String reasonDetail, String processedAt) { + this.sequence = sequence; + this.eventId = eventId; + this.deviceId = deviceId; + this.deviceCollectId = deviceCollectId; + this.topic = topic; + this.receivedAt = receivedAt; + this.observedAt = observedAt; + this.rawMessage = rawMessage; + this.processingStatus = processingStatus; + this.anomaly = anomaly; + this.reason = reason; + this.reasonDetail = reasonDetail; + this.processedAt = processedAt; + } + + private void update(String status, boolean anomaly, String reason, String detail, long now) { + if (this.anomaly && !anomaly) return; + this.processingStatus = status; + this.anomaly = anomaly; + this.reason = reason; + this.reasonDetail = detail; + this.processedAt = instant(now); + } + + private CapturedLogSnapshot snapshot() { + return new CapturedLogSnapshot(sequence, eventId, deviceId, deviceCollectId, topic, + receivedAt, observedAt, processingStatus, anomaly, reason, reasonDetail, + processedAt, rawMessage); + } + } + + static final class CaptureCandidates { + private final String sessionId; + private final long deviceId; + private final long startedAt; + private final long captureEndsAt; + private final long settleEndsAt; + private final int messageCount; + private final long receivedCount; + private final long processedCount; + private final long observedAnomalyCount; + private final List firstMessages; + private final List firstAnomalies; + private final long revision; + + CaptureCandidates(String sessionId, long deviceId, long startedAt, long captureEndsAt, + long settleEndsAt, int messageCount, long receivedCount, + long processedCount, long observedAnomalyCount, + List firstMessages, + List firstAnomalies, long revision) { + this.sessionId = sessionId; + this.deviceId = deviceId; + this.startedAt = startedAt; + this.captureEndsAt = captureEndsAt; + this.settleEndsAt = settleEndsAt; + this.messageCount = messageCount; + this.receivedCount = receivedCount; + this.processedCount = processedCount; + this.observedAnomalyCount = observedAnomalyCount; + this.firstMessages = firstMessages; + this.firstAnomalies = firstAnomalies; + this.revision = revision; + } + + String getSessionId() { return sessionId; } + long getDeviceId() { return deviceId; } + long getStartedAt() { return startedAt; } + long getCaptureEndsAt() { return captureEndsAt; } + long getSettleEndsAt() { return settleEndsAt; } + int getMessageCount() { return messageCount; } + long getReceivedCount() { return receivedCount; } + long getProcessedCount() { return processedCount; } + long getObservedAnomalyCount() { return observedAnomalyCount; } + List getFirstMessages() { return firstMessages; } + List getFirstAnomalies() { return firstAnomalies; } + long getRevision() { return revision; } + } + + public static final class CaptureSnapshot { + private final String sessionId; + private final long deviceId; + private final String status; + private final String startedAt; + private final String captureEndsAt; + private final String settleEndsAt; + private final int messageCount; + private final long receivedCount; + private final long processedCount; + private final long observedAnomalyCount; + private final int retainedAnomalyCount; + private final long replacementCount; + private final List messages; + + CaptureSnapshot(String sessionId, long deviceId, String status, String startedAt, + String captureEndsAt, String settleEndsAt, int messageCount, + long receivedCount, long processedCount, long observedAnomalyCount, + int retainedAnomalyCount, long replacementCount, + List messages) { + this.sessionId = sessionId; + this.deviceId = deviceId; + this.status = status; + this.startedAt = startedAt; + this.captureEndsAt = captureEndsAt; + this.settleEndsAt = settleEndsAt; + this.messageCount = messageCount; + this.receivedCount = receivedCount; + this.processedCount = processedCount; + this.observedAnomalyCount = observedAnomalyCount; + this.retainedAnomalyCount = retainedAnomalyCount; + this.replacementCount = replacementCount; + this.messages = Collections.unmodifiableList(messages); + } + + public String getSessionId() { return sessionId; } + public long getDeviceId() { return deviceId; } + public String getStatus() { return status; } + public String getStartedAt() { return startedAt; } + public String getCaptureEndsAt() { return captureEndsAt; } + public String getSettleEndsAt() { return settleEndsAt; } + public int getMessageCount() { return messageCount; } + public long getReceivedCount() { return receivedCount; } + public long getProcessedCount() { return processedCount; } + public long getObservedAnomalyCount() { return observedAnomalyCount; } + public int getRetainedAnomalyCount() { return retainedAnomalyCount; } + public long getReplacementCount() { return replacementCount; } + public List getMessages() { return messages; } + } + + public static final class CapturedLogSnapshot { + private final long sequence; + private final String eventId; + private final Long deviceId; + private final Long deviceCollectId; + private final String topic; + private final LocalDateTime receivedAt; + private final String observedAt; + private final String processingStatus; + private final boolean anomaly; + private final String reason; + private final String reasonDetail; + private final String processedAt; + private final String rawMessage; + + CapturedLogSnapshot(long sequence, String eventId, Long deviceId, + Long deviceCollectId, String topic, LocalDateTime receivedAt, + String observedAt, String processingStatus, boolean anomaly, + String reason, String reasonDetail, String processedAt, + String rawMessage) { + this.sequence = sequence; + this.eventId = eventId; + this.deviceId = deviceId; + this.deviceCollectId = deviceCollectId; + this.topic = topic; + this.receivedAt = receivedAt; + this.observedAt = observedAt; + this.processingStatus = processingStatus; + this.anomaly = anomaly; + this.reason = reason; + this.reasonDetail = reasonDetail; + this.processedAt = processedAt; + this.rawMessage = rawMessage; + } + + public long getSequence() { return sequence; } + public String getEventId() { return eventId; } + public Long getDeviceId() { return deviceId; } + public Long getDeviceCollectId() { return deviceCollectId; } + public String getTopic() { return topic; } + public LocalDateTime getReceivedAt() { return receivedAt; } + public String getObservedAt() { return observedAt; } + public String getProcessingStatus() { return processingStatus; } + public boolean isAnomaly() { return anomaly; } + public String getReason() { return reason; } + public String getReasonDetail() { return reasonDetail; } + public String getProcessedAt() { return processedAt; } + public String getRawMessage() { return rawMessage; } + } + + @ResponseStatus(HttpStatus.SERVICE_UNAVAILABLE) + public static final class CaptureDisabledException extends RuntimeException { + public CaptureDisabledException() { super("现场原始日志采样功能未启用"); } + } + + @ResponseStatus(HttpStatus.TOO_MANY_REQUESTS) + public static final class CaptureCapacityException extends RuntimeException { + public CaptureCapacityException(int max) { super("采样会话容量已达到上限: " + max); } + } + + @ResponseStatus(HttpStatus.SERVICE_UNAVAILABLE) + public static final class CaptureCoordinationException extends RuntimeException { + public CaptureCoordinationException(Throwable cause) { + super("跨实例原始日志采样暂不可用，请检查 Redis 连接", cause); + } + } +} diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/controllers/RawLogCaptureController.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/controllers/RawLogCaptureController.java new file mode 100644 index 0000000..83385a9 --- /dev/null +++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/controllers/RawLogCaptureController.java @@ -0,0 +1,104 @@ +package com.controllers; + +import com.common.service.RawLogCaptureService; +import org.springframework.http.ResponseEntity; +import org.springframework.validation.annotation.Validated; +import org.springframework.web.bind.annotation.DeleteMapping; +import org.springframework.web.bind.annotation.ExceptionHandler; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.PathVariable; +import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.RequestBody; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RestController; + +import javax.validation.Valid; +import javax.validation.constraints.Max; +import javax.validation.constraints.Min; +import javax.validation.constraints.NotNull; +import java.net.URI; +import java.util.Collections; +import java.util.Map; + +/** 现场按设备开启短时间、异常优先的原始日志采样。 */ +@Validated +@RestController +@RequestMapping("/api/syslog/raw-captures") +public class RawLogCaptureController { + + private final RawLogCaptureService captureService; + + public RawLogCaptureController(RawLogCaptureService captureService) { + this.captureService = captureService; + } + + @PostMapping + public ResponseEntity start( + @Valid @RequestBody StartCaptureRequest request) { + RawLogCaptureService.CaptureSnapshot snapshot = captureService.start( + request.getDeviceId(), request.getDurationSeconds(), request.getMessageCount()); + return ResponseEntity.accepted() + .location(URI.create("/api/syslog/raw-captures/" + snapshot.getSessionId())) + .body(snapshot); + } + + @GetMapping("/{sessionId}") + public ResponseEntity get( + @PathVariable String sessionId) { + RawLogCaptureService.CaptureSnapshot snapshot = captureService.get(sessionId); + return snapshot == null ? ResponseEntity.notFound().build() : ResponseEntity.ok(snapshot); + } + + @DeleteMapping("/{sessionId}") + public ResponseEntity stop( + @PathVariable String sessionId) { + RawLogCaptureService.CaptureSnapshot snapshot = captureService.stop(sessionId); + return snapshot == null ? ResponseEntity.notFound().build() : ResponseEntity.ok(snapshot); + } + + @ExceptionHandler(RawLogCaptureService.CaptureDisabledException.class) + public ResponseEntity> disabled( + RawLogCaptureService.CaptureDisabledException failure) { + return ResponseEntity.status(503) + .body(Collections.singletonMap("message", failure.getMessage())); + } + + @ExceptionHandler(RawLogCaptureService.CaptureCapacityException.class) + public ResponseEntity> capacity( + RawLogCaptureService.CaptureCapacityException failure) { + return ResponseEntity.status(429) + .body(Collections.singletonMap("message", failure.getMessage())); + } + + @ExceptionHandler(RawLogCaptureService.CaptureCoordinationException.class) + public ResponseEntity> coordination( + RawLogCaptureService.CaptureCoordinationException failure) { + return ResponseEntity.status(503) + .body(Collections.singletonMap("message", failure.getMessage())); + } + + public static final class StartCaptureRequest { + @NotNull(message = "deviceId 不能为空") + @Min(value = 1, message = "deviceId 必须大于 0") + private Long deviceId; + + @NotNull(message = "durationSeconds 不能为空") + @Min(value = 1, message = "durationSeconds 最小为 1") + @Max(value = RawLogCaptureService.MAX_DURATION_SECONDS, + message = "durationSeconds 不能超过 600") + private Integer durationSeconds; + + @NotNull(message = "messageCount 不能为空") + @Min(value = 1, message = "messageCount 最小为 1") + @Max(value = RawLogCaptureService.MAX_MESSAGE_COUNT, + message = "messageCount 不能超过 50") + private Integer messageCount; + + public Long getDeviceId() { return deviceId; } + public void setDeviceId(Long deviceId) { this.deviceId = deviceId; } + public Integer getDurationSeconds() { return durationSeconds; } + public void setDurationSeconds(Integer durationSeconds) { this.durationSeconds = durationSeconds; } + public Integer getMessageCount() { return messageCount; } + public void setMessageCount(Integer messageCount) { this.messageCount = messageCount; } + } +} diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/resources/application.properties b/haobang-security-dm/syslog-consumer-rule/src/main/resources/application.properties index fb93523..6e783a0 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/main/resources/application.properties +++ b/haobang-security-dm/syslog-consumer-rule/src/main/resources/application.properties @@ -38,6 +38,19 @@ influxdb.raw.query-max-response-bytes=${INFLUXDB_RAW_QUERY_MAX_RESPONSE_BYTES:10 influxdb.raw.message-max-utf8-bytes=${INFLUXDB_RAW_MESSAGE_MAX_UTF8_BYTES:262144} # Influx 同步写按 Line Protocol 估算字节切批，避免少量大日志形成超大 HTTP 请求。 influxdb.raw.write-batch-max-bytes=${INFLUXDB_RAW_WRITE_BATCH_MAX_BYTES:8388608} + +# 现场短时原始日志采样：单会话硬限制仍为最长600秒、最多50条。 +syslog.raw-capture.enabled=${SYSLOG_RAW_CAPTURE_ENABLED:true} +syslog.raw-capture.cluster-enabled=${SYSLOG_RAW_CAPTURE_CLUSTER_ENABLED:true} +syslog.raw-capture.max-active-sessions=${SYSLOG_RAW_CAPTURE_MAX_ACTIVE_SESSIONS:20} +# 包含已完成且仍可查询的会话，满载时淘汰最早完成的结果。 +syslog.raw-capture.max-retained-sessions=${SYSLOG_RAW_CAPTURE_MAX_RETAINED_SESSIONS:100} +# 给所有消费实例预留会话同步时间；API返回SCHEDULED后从startedAt正式开始计时。 +syslog.raw-capture.cluster-activation-delay-ms=${SYSLOG_RAW_CAPTURE_CLUSTER_ACTIVATION_DELAY_MS:1000} +syslog.raw-capture.cluster-sync-interval-ms=${SYSLOG_RAW_CAPTURE_CLUSTER_SYNC_INTERVAL_MS:200} +# 采集停止后继续接收窗口内消息的迟到处理结果，便于异常样本替换普通样本。 +syslog.raw-capture.settle-seconds=${SYSLOG_RAW_CAPTURE_SETTLE_SECONDS:30} +syslog.raw-capture.result-retention-minutes=${SYSLOG_RAW_CAPTURE_RESULT_RETENTION_MINUTES:30} # InfluxDB 2.7 连接超时配置 influxdb.connection.timeout=30s influxdb.connection.read-timeout=30s diff --git a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/LogNormalProcessorRuleSelectionTest.java b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/LogNormalProcessorRuleSelectionTest.java index 02e947e..3a4e937 100644 --- a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/LogNormalProcessorRuleSelectionTest.java +++ b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/LogNormalProcessorRuleSelectionTest.java @@ -106,37 +106,117 @@ class LogNormalProcessorRuleSelectionTest { } @Test - void invalidJsonStopsAtFirstJsonRuleAndGoesDirectlyToDeferredNonNormal() { + void invalidJsonCandidateDoesNotStopLaterMatchingRule() { RuleSource invalidJson = jsonRule(10L, "json-first", "", equalMapping(), ""); - RuleSource regexWouldMatch = regexRule(20L, "regex-second", "value=(\\S+)"); + RuleSource regexWouldMatch = regexRuleWithCropper( + 20L, "regex-second", "value=(\\S+)", "value="); Fixture fixture = fixture(Arrays.asList(invalidJson, regexWouldMatch), config(true)); LogNormalProcessor.NormalizationResult result = fixture.prepareActive( "value=ok {\"broken\":\"unterminated"); - assertEquals(LogNormalProcessor.ProcessingOutcome.FILTERED, result.getOutcome()); - assertTrue(result.getRows().isEmpty()); - assertEquals(1, result.getNonNormalMessages().size()); - assertEquals("JSON解析异常", result.getNonNormalMessages().get(0).getReason()); - assertTrue(result.getNonNormalMessages().get(0).getReasonDetail().contains("ruleId=10")); + assertEquals(LogNormalProcessor.ProcessingOutcome.SUCCESS, result.getOutcome()); + assertEquals(1, result.getRows().size()); + assertEquals(20L, result.getRows().get(0).get("normalize_rule_id")); + assertTrue(result.getNonNormalMessages().isEmpty()); verify(fixture.nonNormalService, never()).saveMessage(any()); } @Test - void jsonWithoutObjectBoundaryAlsoGoesDirectlyToDeferredNonNormal() { + void invalidJsonCandidateDoesNotStopLaterMatchingRuleInLegacyParser() { + RuleSource invalidJson = jsonRule(10L, "json-first", "", equalMapping(), ""); + RuleSource regexWouldMatch = regexRuleWithCropper( + 20L, "regex-second", "value=(\\S+)", "value="); + Fixture fixture = fixture(Arrays.asList(invalidJson, regexWouldMatch), config(false)); + + LogNormalProcessor.NormalizationResult result = fixture.prepareActive( + "value=ok {\"broken\":\"unterminated"); + + assertEquals(LogNormalProcessor.ProcessingOutcome.SUCCESS, result.getOutcome()); + assertEquals(1, result.getRows().size()); + assertEquals(20L, result.getRows().get(0).get("normalize_rule_id")); + assertTrue(result.getNonNormalMessages().isEmpty()); + verify(fixture.nonNormalService, never()).saveMessage(any()); + } + + @Test + void jsonWithoutObjectBoundaryDoesNotStopLaterMatchingRule() { RuleSource invalidJson = new RuleSource(10L, "json-no-cropper", "json-no-cropper", "json", "{\"mappers\":[" + equalMapping() + "],\"complete_columns\":[]}", true, "0"); - RuleSource regexWouldMatch = regexRule(20L, "regex-second", "value=(\\S+)"); + RuleSource regexWouldMatch = regexRuleWithCropper( + 20L, "regex-second", "value=(\\S+)", "value="); Fixture fixture = fixture(Arrays.asList(invalidJson, regexWouldMatch), config(true)); LogNormalProcessor.NormalizationResult result = fixture.prepareActive("value=ok"); + assertEquals(LogNormalProcessor.ProcessingOutcome.SUCCESS, result.getOutcome()); + assertEquals(1, result.getRows().size()); + assertEquals(20L, result.getRows().get(0).get("normalize_rule_id")); + assertTrue(result.getNonNormalMessages().isEmpty()); + verify(fixture.nonNormalService, never()).saveMessage(any()); + } + + @Test + void threatbookUsesTdpRuleAfterFixedOffsetHoneypotRulesAreRejected() { + String honeypotMapping = "{\"origin_field\":\"title\",\"dest_field\":\"device_name\"," + + "\"action\":{\"type\":\"equal\",\"param\":{}}}"; + String tdpMapping = "{\"origin_field\":\"machine\",\"dest_field\":\"victim_ip\"," + + "\"action\":{\"type\":\"equal\",\"param\":{}}}"; + RuleSource honeypotV12 = jsonRuleWithCropper(158L, "weibujsomv2", + "<", 58, honeypotMapping); + RuleSource honeypotV13 = jsonRuleWithCropper(159L, "weibujson", + "<", 57, honeypotMapping); + RuleSource tdp = jsonRuleWithCropper(150L, "weibu_TDP_alarmlog_JSON", + "{", -1, tdpMapping); + Fixture fixture = fixture(Arrays.asList(honeypotV12, honeypotV13, tdp), config(true)); + + LogNormalProcessor.NormalizationResult result = fixture.prepareActive( + "<1>2026-09-14T03:34:12+08:00 localhost.localdomain Threatbook[2477960]: " + + "{\"time\":1789328052,\"machine\":\"10.174.2.19\"," + + "\"net\":{\"http\":{\"reqs_body\":" + + "\"{\\\"operatorID\\\":\\\"MA59RW1N3\\\"}\"}}}"); + + assertEquals(LogNormalProcessor.ProcessingOutcome.SUCCESS, result.getOutcome()); + assertEquals(1, result.getRows().size()); + assertEquals(150L, result.getRows().get(0).get("normalize_rule_id")); + assertEquals("10.174.2.19", result.getRows().get(0).get("victim_ip")); + assertTrue(result.getNonNormalMessages().isEmpty()); + } + + @Test + void skyeyeCropperPrefixIsNarrowedToOuterJsonBeforeJackson() { + RuleSource skyeye = jsonRuleWithCropper(129L, "skyeye", + "|!alarm|!{", -1, equalMapping()); + RuleSource unrelated = regexRule(200L, "unrelated", "never=(\\S+)"); + Fixture fixture = fixture(Arrays.asList(skyeye, unrelated), config(true)); + + LogNormalProcessor.NormalizationResult result = fixture.prepareActive( + "<158>Sep 14 11:07:16 host SyslogClient[1]: " + + "2026-09-14 11:07:16|!10.56.170.107|!alarm|!" + + "{\"status\":\"ok\",\"payload\":{\"rsp_header\":\"a\\r\\nb\"}}"); + + assertEquals(LogNormalProcessor.ProcessingOutcome.SUCCESS, result.getOutcome()); + assertEquals(1, result.getRows().size()); + assertEquals(129L, result.getRows().get(0).get("normalize_rule_id")); + assertEquals("ok", result.getRows().get(0).get("action")); + } + + @Test + void allApplicableJsonCandidatesInvalidStillProduceJsonException() { + RuleSource first = jsonRuleWithCropper(10L, "first", "marker={", -1, + equalMapping()); + RuleSource second = jsonRuleWithCropper(20L, "second", "marker={", -1, + equalMapping()); + Fixture fixture = fixture(Arrays.asList(first, second), config(true)); + + LogNormalProcessor.NormalizationResult result = fixture.prepareActive( + "marker={\"broken\":\"unterminated"); + assertEquals(LogNormalProcessor.ProcessingOutcome.FILTERED, result.getOutcome()); - assertTrue(result.getRows().isEmpty()); assertEquals(1, result.getNonNormalMessages().size()); assertEquals("JSON解析异常", result.getNonNormalMessages().get(0).getReason()); - verify(fixture.nonNormalService, never()).saveMessage(any()); + assertTrue(result.getNonNormalMessages().get(0).getReasonDetail().contains("ruleId=10")); } @Test @@ -250,6 +330,26 @@ class LogNormalProcessorRuleSelectionTest { return new RuleSource(id, name, name, "regex", content, true, "0"); } + private static RuleSource regexRuleWithCropper(long id, String name, String regexp, + String headKey) { + String content = "{\"cropper_params\":{\"head_key\":\"" + headKey + + "\",\"head_offset\":0,\"tail_key\":\"\",\"tail_offset\":0}," + + "\"regexp\":\"" + regexp.replace("\\", "\\\\") + "\"," + + "\"mappers\":[{\"origin_field\":\"1\",\"dest_field\":\"value\"," + + "\"action\":{\"type\":\"equal\",\"param\":{}}}]}"; + return new RuleSource(id, name, name, "regex", content, true, "0"); + } + + private static RuleSource jsonRuleWithCropper(long id, String name, String headKey, + int headOffset, String mapper) { + String escapedHeadKey = headKey.replace("\\", "\\\\").replace("\"", "\\\""); + String content = "{\"cropper_params\":{\"head_key\":\"" + escapedHeadKey + + "\",\"head_offset\":" + headOffset + + ",\"tail_key\":\"\",\"tail_offset\":0}," + + "\"mappers\":[" + mapper + "],\"complete_columns\":[]}"; + return new RuleSource(id, name, name, "json", content, true, "0"); + } + private static String equalMapping() { return "{\"origin_field\":\"status\",\"dest_field\":\"action\"," + "\"action\":{\"type\":\"equal\",\"param\":{}}}"; diff --git a/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/RawLogCaptureServiceTest.java b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/RawLogCaptureServiceTest.java new file mode 100644 index 0000000..f2cad8a --- /dev/null +++ b/haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/RawLogCaptureServiceTest.java @@ -0,0 +1,154 @@ +package com.common.service; + +import org.junit.jupiter.api.Test; + +import java.time.Clock; +import java.time.Instant; +import java.time.LocalDateTime; +import java.time.ZoneId; +import java.time.ZoneOffset; +import java.util.Arrays; +import java.util.List; +import java.util.stream.Collectors; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class RawLogCaptureServiceTest { + + @Test + void keepsFirstMessagesThenReplacesOldestNormalWithLaterAnomalies() { + MutableClock clock = new MutableClock(1_700_000_000_000L); + RawLogCaptureService service = new RawLogCaptureService(clock); + RawLogCaptureService.CaptureSnapshot started = service.start(7L, 60, 3); + + receive(service, clock, 7L, "e1"); + receive(service, clock, 7L, "e2"); + receive(service, clock, 7L, "e3"); + receive(service, clock, 7L, "e4"); + assertEquals(Arrays.asList("e1", "e2", "e3"), ids(service.get(started.getSessionId()))); + + fail(service, clock, 7L, "e4"); + fail(service, clock, 7L, "e5"); + fail(service, clock, 7L, "e6"); + RawLogCaptureService.CaptureSnapshot anomalies = service.get(started.getSessionId()); + + assertEquals(Arrays.asList("e4", "e5", "e6"), ids(anomalies)); + assertEquals(3, anomalies.getRetainedAnomalyCount()); + assertEquals(3L, anomalies.getReplacementCount()); + + fail(service, clock, 7L, "e7"); + assertEquals(Arrays.asList("e4", "e5", "e6"), + ids(service.get(started.getSessionId()))); + } + + @Test + void upgradesSelectedMessageToAnomalyWithoutReplacement() { + MutableClock clock = new MutableClock(1_700_000_000_000L); + RawLogCaptureService service = new RawLogCaptureService(clock); + RawLogCaptureService.CaptureSnapshot started = service.start(7L, 60, 2); + + receive(service, clock, 7L, "e1"); + receive(service, clock, 7L, "e2"); + fail(service, clock, 7L, "e2"); + + RawLogCaptureService.CaptureSnapshot snapshot = service.get(started.getSessionId()); + assertEquals(Arrays.asList("e1", "e2"), ids(snapshot)); + assertEquals(1, snapshot.getRetainedAnomalyCount()); + assertEquals(0L, snapshot.getReplacementCount()); + assertFalse(snapshot.getMessages().get(0).isAnomaly()); + assertTrue(snapshot.getMessages().get(1).isAnomaly()); + } + + @Test + void isolatesDevicesAndAcceptsLateOutcomeOnlyDuringSettlingPeriod() { + MutableClock clock = new MutableClock(1_700_000_000_000L); + RawLogCaptureService service = new RawLogCaptureService(clock); + RawLogCaptureService.CaptureSnapshot started = service.start(7L, 2, 1); + long observedInsideWindow = clock.millis(); + + receive(service, clock, 8L, "other-device"); + receive(service, clock, 7L, "normal"); + clock.advanceMillis(3_000L); + assertEquals("DRAINING", service.get(started.getSessionId()).getStatus()); + + service.observeFailure(observedInsideWindow, 7L, 1L, "late-error", "topic", + LocalDateTime.ofInstant(Instant.ofEpochMilli(observedInsideWindow), ZoneOffset.UTC), + "raw-late-error", "RULE_PROCESSING", "DLT_PUBLISHED", + new IllegalStateException("bad json")); + RawLogCaptureService.CaptureSnapshot draining = service.get(started.getSessionId()); + assertEquals(Arrays.asList("late-error"), ids(draining)); + assertEquals(1L, draining.getReplacementCount()); + + clock.advanceMillis(31_000L); + assertEquals("COMPLETED", service.get(started.getSessionId()).getStatus()); + } + + @Test + void validatesHardApiLimits() { + RawLogCaptureService service = new RawLogCaptureService( + new MutableClock(1_700_000_000_000L)); + assertThrows(IllegalArgumentException.class, () -> service.start(0L, 1, 1)); + assertThrows(IllegalArgumentException.class, () -> service.start(1L, 601, 1)); + assertThrows(IllegalArgumentException.class, () -> service.start(1L, 1, 51)); + } + + @Test + void mergesCandidatesAcrossConsumerInstancesBeforeApplyingAnomalyReplacement() { + List first = Arrays.asList( + snapshot(2, "a2", false), snapshot(3, "a3", false), snapshot(4, "a4", false), + snapshot(1, "b1", false), snapshot(100, "b100", false), + snapshot(101, "b101", false)); + List anomalies = + Arrays.asList(snapshot(200, "a200", true)); + + List selected = + RawLogCaptureService.mergeCandidates(first, anomalies, 3); + + assertEquals(Arrays.asList("a2", "a3", "a200"), selected.stream() + .map(RawLogCaptureService.CapturedLogSnapshot::getEventId) + .collect(Collectors.toList())); + } + + private static RawLogCaptureService.CapturedLogSnapshot snapshot( + long sequence, String eventId, boolean anomaly) { + return new RawLogCaptureService.CapturedLogSnapshot(sequence, eventId, 7L, 1L, + "topic", null, "2026-09-14T00:00:00Z", + anomaly ? "NON_NORMAL" : "SUCCESS", anomaly, + anomaly ? "JSON解析异常" : null, null, null, "raw-" + eventId); + } + + private static void receive(RawLogCaptureService service, Clock clock, + long deviceId, String eventId) { + service.observeReceived(clock.millis(), deviceId, 1L, eventId, "topic", + LocalDateTime.ofInstant(Instant.ofEpochMilli(clock.millis()), ZoneOffset.UTC), + "raw-" + eventId); + } + + private static void fail(RawLogCaptureService service, Clock clock, + long deviceId, String eventId) { + service.observeFailure(clock.millis(), deviceId, 1L, eventId, "topic", + LocalDateTime.ofInstant(Instant.ofEpochMilli(clock.millis()), ZoneOffset.UTC), + "raw-" + eventId, "RULE_PROCESSING", "DLT_PUBLISHED", + new IllegalArgumentException("broken")); + } + + private static List ids(RawLogCaptureService.CaptureSnapshot snapshot) { + return snapshot.getMessages().stream() + .map(RawLogCaptureService.CapturedLogSnapshot::getEventId) + .collect(Collectors.toList()); + } + + private static final class MutableClock extends Clock { + private long millis; + + private MutableClock(long millis) { this.millis = millis; } + private void advanceMillis(long value) { millis += value; } + @Override public ZoneId getZone() { return ZoneOffset.UTC; } + @Override public Clock withZone(ZoneId zone) { return this; } + @Override public Instant instant() { return Instant.ofEpochMilli(millis); } + @Override public long millis() { return millis; } + } +} diff --git a/haobang-security-dm/upgrade-doc/README.md b/haobang-security-dm/upgrade-doc/README.md index 73d5074..45f3eee 100644 --- a/haobang-security-dm/upgrade-doc/README.md +++ b/haobang-security-dm/upgrade-doc/README.md @@ -8,12 +8,15 @@ | `v2/` | 从提交 `b61ec2e` 升级到当前源码所需的增量资料 | | `v3/` | 日统计异步投影，以及大日志独立 Kafka 通道、字节背压和存储截断升级资料 | | `v4/` | Kafka 积压分级采样、告警优先落库和 InfluxDB 非阻塞旁路升级资料 | +| `v5/` | 按设备短时采集未截断原始日志，并将异常样本优先落盘的现场诊断工具 | -新环境按 v1、v2、v3、v4 的顺序执行；已经完成 v3 的现场只执行 v4。已经执行过早期 v3 日统计升级的 -现场，只需补做 v3 中的大日志 Topic、应用升级和验证步骤，再执行 v4；不要重复执行数据库升级脚本。 +新环境按 v1、v2、v3、v4、v5 的顺序执行；已经完成 v4 的现场只需升级应用并执行 v5。已经执行过早期 +v3 日统计升级的现场，只需补做 v3 中的大日志 Topic、应用升级和验证步骤，再执行 v4、v5；不要重复执行 +数据库升级脚本。 数据库脚本不要重复执行。执行过程中有报错时，先确认已经落库的对象，再决定从哪里继续。 v2 的执行步骤见 [v2/README.md](./v2/README.md)。 v3 的执行步骤见 [v3/README.md](./v3/README.md)。 v4 的执行步骤见 [v4/README.md](./v4/README.md)。 +v5 的执行步骤见 [v5/README.md](./v5/README.md)。 diff --git a/haobang-security-dm/upgrade-doc/v2/config/syslog-consumer-rule.env.example b/haobang-security-dm/upgrade-doc/v2/config/syslog-consumer-rule.env.example index 143b3a9..5350533 100644 --- a/haobang-security-dm/upgrade-doc/v2/config/syslog-consumer-rule.env.example +++ b/haobang-security-dm/upgrade-doc/v2/config/syslog-consumer-rule.env.example @@ -18,6 +18,16 @@ NORMALIZATION_RULE_SELECTION_DETAIL_LOG_ENABLED=false NORMALIZATION_RULE_SELECTION_EXPLICIT_ENABLED=true NORMALIZATION_RULE_SELECTION_AUTOMATIC_FALLBACK_ENABLED=true +# 现场原始日志短时采样；三个消费实例通过同一个 Redis 汇聚候选结果。 +SYSLOG_RAW_CAPTURE_ENABLED=true +SYSLOG_RAW_CAPTURE_CLUSTER_ENABLED=true +SYSLOG_RAW_CAPTURE_MAX_ACTIVE_SESSIONS=20 +SYSLOG_RAW_CAPTURE_MAX_RETAINED_SESSIONS=100 +SYSLOG_RAW_CAPTURE_CLUSTER_ACTIVATION_DELAY_MS=1000 +SYSLOG_RAW_CAPTURE_CLUSTER_SYNC_INTERVAL_MS=200 +SYSLOG_RAW_CAPTURE_SETTLE_SECONDS=30 +SYSLOG_RAW_CAPTURE_RESULT_RETENTION_MINUTES=30 + # 统计投影解耦：三个 syslog-consumer-rule 实例保持完全相同的 Topic 与 Group ID。 STATS_MATERIALIZATION_ENABLED=true STATS_MATERIALIZATION_TOPIC=agent-01-syslog-topic.STATS.MATERIALIZE diff --git a/haobang-security-dm/upgrade-doc/v3/config/01-new-environment-variables.md b/haobang-security-dm/upgrade-doc/v3/config/01-new-environment-variables.md index f4a0978..45b0535 100644 --- a/haobang-security-dm/upgrade-doc/v3/config/01-new-environment-variables.md +++ b/haobang-security-dm/upgrade-doc/v3/config/01-new-environment-variables.md @@ -23,9 +23,23 @@ syslog-serve/docker-compose/docker-compose-serve-prod.yaml - SYSLOG_MESSAGE_MAX_BYTES=6291456 - SYSLOG_LARGE_MESSAGE_THRESHOLD_BYTES=131072 - # WAL 和 Kafka 投递最多各占用 256 MiB 在途内存。 - - SYSLOG_WAL_MAX_PENDING_BYTES=268435456 - - SYSLOG_DISPATCH_MAX_INFLIGHT_BYTES=268435456 + # 200KiB/条容量基线；WAL 高水位约在 14,000 条、2.7~2.8GiB 对齐触发。 + - SYSLOG_WAL_QUEUE_CAPACITY=20000 + - SYSLOG_WAL_MAX_PENDING_BYTES=4294967296 + - SYSLOG_WAL_HIGH_WATERMARK_PERCENT=70 + - SYSLOG_WAL_LOW_WATERMARK_PERCENT=40 + - SYSLOG_WAL_FLUSH_RECORDS=200 + - SYSLOG_WAL_FLUSH_INTERVAL_MS=100 + - SYSLOG_WAL_SEGMENT_MAX_BYTES=4294967296 + + # Kafka 投递限制正文最多 1GiB 在途，并增加加密后的 Producer 缓冲。 + - SYSLOG_DISPATCH_WORKER_THREADS=8 + - SYSLOG_DISPATCH_MAX_INFLIGHT=5000 + - SYSLOG_DISPATCH_MAX_INFLIGHT_BYTES=1073741824 + - SPRING_KAFKA_PRODUCER_PROPERTIES_BUFFER_MEMORY=536870912 + + # Dockerfile 不展开 JAVA_OPTS，使用 JVM 自动读取的 JAVA_TOOL_OPTIONS。 + - JAVA_TOOL_OPTIONS=-Xms8g -Xmx24g -XX:+UseG1GC -XX:MaxGCPauseMillis=200 -XX:MaxDirectMemorySize=8g -XX:+ExitOnOutOfMemoryError # 大日志 Kafka Topic 及 Producer 请求上限。 - KAFKA_PRODUCER_LARGE_TOPIC=agent-01-syslog-topic.large @@ -37,7 +51,9 @@ syslog-serve/docker-compose/docker-compose-serve-prod.yaml - `SYSLOG_MAX_FRAME_LENGTH`：TCP 单帧最大长度； - `SYSLOG_MESSAGE_MAX_BYTES`：所有来源的单条日志硬上限，超过后拒绝； - `SYSLOG_LARGE_MESSAGE_THRESHOLD_BYTES`：达到该值后走大日志 Topic； -- `SYSLOG_WAL_MAX_PENDING_BYTES`、`SYSLOG_DISPATCH_MAX_INFLIGHT_BYTES`：限制 JVM 在途数据量。 +- `SYSLOG_WAL_QUEUE_CAPACITY`、`SYSLOG_WAL_MAX_PENDING_BYTES`：共同限制 WAL 接入条数和原文字节量； +- `SYSLOG_DISPATCH_MAX_INFLIGHT_BYTES`：限制 Kafka Dispatcher 在途正文总量； +- `JAVA_TOOL_OPTIONS`：设置实际 JVM 堆和直接内存上限，不能替换成当前 Dockerfile 不会展开的 `JAVA_OPTS`。 `SYSLOG_MAX_FRAME_LENGTH` 必须大于或等于 `SYSLOG_MESSAGE_MAX_BYTES`。如果设备发送 GBK，应将 `SYSLOG_CHARSET` 改为 `GBK`。 diff --git a/haobang-security-dm/upgrade-doc/v3/kafka/03-query-kafka-consumer-lag.md b/haobang-security-dm/upgrade-doc/v3/kafka/03-query-kafka-consumer-lag.md new file mode 100644 index 0000000..31c6789 --- /dev/null +++ b/haobang-security-dm/upgrade-doc/v3/kafka/03-query-kafka-consumer-lag.md @@ -0,0 +1,45 @@ +# 查询 Kafka 当前积压 + +在 Kafka 的 `bin` 目录执行。查询操作只读，不会修改消息或消费组 offset。 + +## 查询主日志消费积压 + +```bash +./kafka-consumer-groups.sh \ + --bootstrap-server 10.141.121.35:9092 \ + --group agent-01-syslog-group-dm \ + --describe +``` + +如果现场主消费组仍使用旧名称，将组名改为 `agent-01-syslog-group`。 + +## 查询所有消费组的总积压 + +```bash +./kafka-consumer-groups.sh \ + --bootstrap-server 10.141.121.35:9092 \ + --all-groups \ + --describe \ + | awk '$6 ~ /^[0-9]+$/ {lag[$1]+=$6} END {for (g in lag) print lag[g], g}' \ + | sort -nr +``` + +输出第一列是总积压条数，第二列是消费组名称，积压最多的排在最前面。 + +需要查看某个消费组各分区的明细时执行： + +```bash +./kafka-consumer-groups.sh \ + --bootstrap-server 10.141.121.35:9092 \ + --group 消费组名称 \ + --describe +``` + +重点查看： + +- `CURRENT-OFFSET`：已经消费并提交的位置； +- `LOG-END-OFFSET`：Kafka 当前最新位置； +- `LAG`：该分区积压条数。 + +间隔一分钟查询两次：`LAG` 下降表示正在追赶；`LAG` 上升表示消费速度低于数据进入速度； +`CURRENT-OFFSET` 不变且 `LAG` 大于零，表示消费没有进展。 diff --git a/haobang-security-dm/upgrade-doc/v3/kafka/04-query-backlog-sampling-dropped.md b/haobang-security-dm/upgrade-doc/v3/kafka/04-query-backlog-sampling-dropped.md new file mode 100644 index 0000000..f92f416 --- /dev/null +++ b/haobang-security-dm/upgrade-doc/v3/kafka/04-query-backlog-sampling-dropped.md @@ -0,0 +1,61 @@ +# 查询主动丢弃的非告警数据 + +积压采样只丢弃部分普通日志和非标消息，告警全部保留。审计数据保存在达梦表 +`syslog_backlog_retention_hourly`。 + +## 查看当前运行累计值 + +```bash +docker exec syslog-consumer-rule-dm sh -c \ + 'curl -fsS http://127.0.0.1:8289/xdrrule/api/syslog/backlog-sampling/status' +``` + +主要字段： + +- `normal_rows_dropped_total`：本次应用启动后主动丢弃的普通日志行数； +- `non_normal_rows_dropped_total`：本次应用启动后主动丢弃的非标消息行数； +- `raw_dropped_total`：本次应用启动后没有写入 InfluxDB 的原文条数。 + +## 查询指定一天的丢弃总数 + +以下查询统计上海时间 `2026-09-13` 一整天： + +```sql +SELECT + NVL(SUM(normal_rows_dropped), 0) AS normal_dropped, + NVL(SUM(non_normal_rows_dropped), 0) AS non_normal_dropped, + NVL(SUM(normal_rows_dropped), 0) + + NVL(SUM(non_normal_rows_dropped), 0) AS non_alarm_rows_dropped, + NVL(SUM(raw_dropped), 0) AS raw_dropped, + NVL(SUM(alarm_rows_total), 0) AS alarm_total, + NVL(SUM(alarm_rows_saved), 0) AS alarm_saved +FROM syslog_backlog_retention_hourly +WHERE source_hour >= TIMESTAMP '2026-09-13 00:00:00' + AND source_hour < TIMESTAMP '2026-09-14 00:00:00'; +``` + +## 按小时统计 + +```sql +SELECT + source_hour, + SUM(normal_rows_dropped) AS normal_dropped, + SUM(non_normal_rows_dropped) AS non_normal_dropped, + SUM(normal_rows_dropped) + SUM(non_normal_rows_dropped) + AS non_alarm_rows_dropped, + SUM(raw_dropped) AS raw_dropped, + SUM(alarm_rows_total) AS alarm_total, + SUM(alarm_rows_saved) AS alarm_saved, + MAX(max_lag_ms) AS max_lag_ms +FROM syslog_backlog_retention_hourly +WHERE source_hour >= TIMESTAMP '2026-09-13 00:00:00' + AND source_hour < TIMESTAMP '2026-09-14 00:00:00' +GROUP BY source_hour +ORDER BY source_hour; +``` + +`non_alarm_rows_dropped` 是普通和非标两类达梦业务行的合计。`raw_dropped` 是原始消息口径， +可能与这两类记录重叠，因此不要再把 `raw_dropped` 加到 `non_alarm_rows_dropped` 中。 + +正常情况下 `alarm_total` 应等于 `alarm_saved`。查询结果为零或没有小时记录，表示该时间段没有触发 +积压采样，或者采样审计尚未成功写入。 diff --git a/haobang-security-dm/upgrade-doc/v5/README.md b/haobang-security-dm/upgrade-doc/v5/README.md new file mode 100644 index 0000000..53b5c8a --- /dev/null +++ b/haobang-security-dm/upgrade-doc/v5/README.md @@ -0,0 +1,139 @@ +# v5 现场设备原始日志采样 + +## 功能说明 + +v5 为 `syslog-consumer-rule` 增加按设备短时采集原始日志的诊断接口，并提供自动落盘脚本。 + +- 单次采集最长 600 秒； +- 单次最多返回 50 条； +- 先保留采集窗口内最早的 N 条日志； +- 后续出现非标、规则执行异常、InfluxDB 写入异常、达梦写入异常或 Retry/DLT 时，用异常日志逐条替换最早的正常样本； +- 返回内容来自规则解析和非标字段截断之前，不受 `HB_XDR_TRUNCATED` 影响； +- 三个 `syslog-consumer-rule` 实例通过相同 Redis 汇聚有界候选集； +- 不新增数据库表，也不需要执行达梦 SQL。 + +## 升级与配置 + +滚动升级全部 `syslog-consumer-rule` 实例，并确保各实例连接同一个 Redis。原始日志采集和多实例汇聚默认 +均已启用；建议在所有实例显式保留以下配置，便于现场识别和后续调整： + +```yaml +- SYSLOG_RAW_CAPTURE_ENABLED=true +- SYSLOG_RAW_CAPTURE_CLUSTER_ENABLED=true +``` + +推荐保留默认配置： + +```yaml +- SYSLOG_RAW_CAPTURE_MAX_ACTIVE_SESSIONS=20 +- SYSLOG_RAW_CAPTURE_MAX_RETAINED_SESSIONS=100 +- SYSLOG_RAW_CAPTURE_CLUSTER_ACTIVATION_DELAY_MS=1000 +- SYSLOG_RAW_CAPTURE_CLUSTER_SYNC_INTERVAL_MS=200 +- SYSLOG_RAW_CAPTURE_SETTLE_SECONDS=30 +- SYSLOG_RAW_CAPTURE_RESULT_RETENTION_MINUTES=30 +``` + +如果多个 Java 进程运行在同一主机且没有不同的容器 `HOSTNAME`，应为每个实例配置不同标识： + +```yaml +- SYSLOG_RAW_CAPTURE_NODE_ID=consumer-rule-01 +``` + +接口会返回完整原文，应只允许受信任的运维网络或现有管理网关访问。 + +## 一键采集 + +脚本依赖 `bash`、`curl` 和 `python3`。先增加执行权限： + +```bash +chmod 700 /升级资料路径/v5/capture-device-raw-logs.sh +``` + +基本用法： + +```bash +/升级资料路径/v5/capture-device-raw-logs.sh \ + 38 \ + 60 \ + 10 \ + http://127.0.0.1:8289/xdrrule \ + /data/haobang/raw-log-captures +``` + +参数依次是： + +1. 设备 ID； +2. 采集秒数，默认 60，最大 600； +3. 样本数，默认 10，最大 50； +4. `syslog-consumer-rule` 基础地址，默认 `http://127.0.0.1:8289/xdrrule`； +5. 本地输出目录，默认当前目录下的 `raw-log-captures`。 + +如果接口经过鉴权网关，可通过环境变量传入请求头： + +```bash +RAW_CAPTURE_AUTH_HEADER='Authorization: Bearer 替换为现场令牌' \ + ./capture-device-raw-logs.sh 38 120 20 https://管理地址/xdrrule /data/raw-samples +``` + +## 输出文件 + +每次运行生成独立目录： + +```text +device-38_20260914-190000_会话ID/ +├── start.json +├── latest.json +├── result.json +├── samples.jsonl +├── session-id.txt +├── summary.txt +└── raw/ + ├── 001_normal_事件ID.log + └── 002_anomaly_事件ID.log +``` + +- `result.json`：最终完整 API 响应，是最完整的交付文件； +- `samples.jsonl`：每行一条样本及其处理状态、异常原因和原文； +- `raw/*.log`：每条原始报文单独保存，内容不添加分隔符，也不主动截断； +- `summary.txt`：会话时间、收到的消息数、异常数和替换数； +- `latest.json`：脚本被中断或等待超时时保留的最后一次有效快照。 + +脚本设置 `umask 077`，新生成文件默认只允许执行脚本的账号访问。 + +## 生命周期 + +状态依次为： + +```text +SCHEDULED -> RUNNING -> DRAINING -> COMPLETED +``` + +采集时间到达后不再接收新日志，随后默认等待 30 秒，让窗口内日志的异步处理异常完成替换。进入 +`COMPLETED` 后结果默认继续保留 30 分钟，之后 Redis TTL 自动删除。 + +脚本会一直轮询到 `COMPLETED` 并自动导出，不需要人工再次查询。关闭终端或脚本等待超时不会使服务端会话 +永久运行，会话仍按上述时间自动结束。 + +## 常见问题 + +### 返回 503：现场原始日志采样功能未启用 + +确认实例没有通过环境变量设置 `SYSLOG_RAW_CAPTURE_ENABLED=false`，并确认实例已经升级到 v5 JAR。 + +### 返回 503：跨实例原始日志采样暂不可用 + +检查所有 `syslog-consumer-rule` 实例到 Redis 的网络、密码、数据库编号是否一致。Redis 故障只影响诊断 +采集，不会中断主 Kafka 消费。 + +### 返回 429 + +活动采集会话达到上限。等待已有会话进入 `COMPLETED`，或核对是否有人重复创建采集任务。 + +### 脚本退出码为 2 + +表示客户端等待超时，但已经把最后一次有效快照写入输出目录。服务端会话仍会自动结束，可在结果保留期内 +使用 `session-id.txt` 中的 ID 再查询： + +```bash +curl 'http://127.0.0.1:8289/xdrrule/api/syslog/raw-captures/会话ID' +``` diff --git a/haobang-security-dm/upgrade-doc/v5/capture-device-raw-logs.sh b/haobang-security-dm/upgrade-doc/v5/capture-device-raw-logs.sh new file mode 100755 index 0000000..e005724 --- /dev/null +++ b/haobang-security-dm/upgrade-doc/v5/capture-device-raw-logs.sh @@ -0,0 +1,229 @@ +#!/usr/bin/env bash +set -Eeuo pipefail + +# 原始日志可能包含敏感字段，确保新建目录和文件默认只允许当前用户读取。 +umask 077 + +usage() { + cat <<'EOF' +用法: + capture-device-raw-logs.sh DEVICE_ID [DURATION_SECONDS] [MESSAGE_COUNT] [BASE_URL] [OUTPUT_DIR] + +参数: + DEVICE_ID 必填，设备ID + DURATION_SECONDS 可选，采集秒数，1~600，默认60 + MESSAGE_COUNT 可选，样本条数，1~50，默认10 + BASE_URL 可选，默认 http://127.0.0.1:8289/xdrrule + OUTPUT_DIR 可选，默认当前目录下 raw-log-captures + +可选环境变量: + RAW_CAPTURE_AUTH_HEADER 例如 Authorization: Bearer xxx + RAW_CAPTURE_POLL_SECONDS 查询间隔，默认2秒 + RAW_CAPTURE_WAIT_TIMEOUT_SECONDS 总等待超时；默认采集时间加420秒 + RAW_CAPTURE_HTTP_MAX_TIME_SECONDS 单次HTTP超时，默认120秒 +EOF +} + +fail() { + printf 'ERROR: %s\n' "$*" >&2 + exit 1 +} + +require_command() { + command -v "$1" >/dev/null 2>&1 || fail "缺少命令: $1" +} + +positive_integer() { + [[ "$1" =~ ^[0-9]+$ ]] && (( 10#$1 > 0 )) +} + +if [[ "${1:-}" == "-h" || "${1:-}" == "--help" ]]; then + usage + exit 0 +fi + +DEVICE_ID="${1:-}" +DURATION_SECONDS="${2:-60}" +MESSAGE_COUNT="${3:-10}" +BASE_URL="${4:-http://127.0.0.1:8289/xdrrule}" +OUTPUT_DIR="${5:-$(pwd)/raw-log-captures}" +POLL_SECONDS="${RAW_CAPTURE_POLL_SECONDS:-2}" +HTTP_MAX_TIME_SECONDS="${RAW_CAPTURE_HTTP_MAX_TIME_SECONDS:-120}" + +[[ -n "$DEVICE_ID" ]] || { usage >&2; fail "必须提供 DEVICE_ID"; } +positive_integer "$DEVICE_ID" || fail "DEVICE_ID 必须是大于0的整数" +positive_integer "$DURATION_SECONDS" || fail "DURATION_SECONDS 必须是大于0的整数" +positive_integer "$MESSAGE_COUNT" || fail "MESSAGE_COUNT 必须是大于0的整数" +positive_integer "$POLL_SECONDS" || fail "RAW_CAPTURE_POLL_SECONDS 必须是大于0的整数" +positive_integer "$HTTP_MAX_TIME_SECONDS" || fail "RAW_CAPTURE_HTTP_MAX_TIME_SECONDS 必须是大于0的整数" +(( 10#$DURATION_SECONDS <= 600 )) || fail "DURATION_SECONDS 不能超过600" +(( 10#$MESSAGE_COUNT <= 50 )) || fail "MESSAGE_COUNT 不能超过50" +(( 10#$POLL_SECONDS <= 30 )) || fail "RAW_CAPTURE_POLL_SECONDS 不能超过30" + +DEFAULT_WAIT_TIMEOUT=$((10#$DURATION_SECONDS + 420)) +WAIT_TIMEOUT_SECONDS="${RAW_CAPTURE_WAIT_TIMEOUT_SECONDS:-$DEFAULT_WAIT_TIMEOUT}" +positive_integer "$WAIT_TIMEOUT_SECONDS" || fail "RAW_CAPTURE_WAIT_TIMEOUT_SECONDS 必须是大于0的整数" + +require_command curl +require_command python3 +require_command mktemp + +BASE_URL="${BASE_URL%/}" +if [[ "$BASE_URL" == */api/syslog/raw-captures ]]; then + CAPTURE_API="$BASE_URL" +else + CAPTURE_API="$BASE_URL/api/syslog/raw-captures" +fi + +mkdir -p -- "$OUTPUT_DIR" +TEMP_DIR="$(mktemp -d "${TMPDIR:-/tmp}/hb-raw-capture.XXXXXX")" +SESSION_ID="" +CAPTURE_DIR="" + +cleanup() { + rm -rf -- "$TEMP_DIR" +} +trap cleanup EXIT + +CURL_HEADERS=(-H 'Accept: application/json') +if [[ -n "${RAW_CAPTURE_AUTH_HEADER:-}" ]]; then + CURL_HEADERS+=(-H "$RAW_CAPTURE_AUTH_HEADER") +fi + +http_request() { + local method="$1" url="$2" output="$3" body="${4:-}" + local args=(-sS -o "$output" -w '%{http_code}' --connect-timeout 10 + --max-time "$HTTP_MAX_TIME_SECONDS" -X "$method" "${CURL_HEADERS[@]}") + if [[ -n "$body" ]]; then + args+=(-H 'Content-Type: application/json' --data "$body") + fi + curl "${args[@]}" "$url" +} + +START_RESPONSE="$TEMP_DIR/start.json" +REQUEST_BODY="{\"deviceId\":$((10#$DEVICE_ID)),\"durationSeconds\":$((10#$DURATION_SECONDS)),\"messageCount\":$((10#$MESSAGE_COUNT))}" + +printf '创建采集会话: deviceId=%s, duration=%ss, count=%s\n' \ + "$DEVICE_ID" "$DURATION_SECONDS" "$MESSAGE_COUNT" +if ! HTTP_CODE="$(http_request POST "$CAPTURE_API" "$START_RESPONSE" "$REQUEST_BODY")"; then + fail "无法连接采集API: $CAPTURE_API" +fi +if [[ "$HTTP_CODE" != "202" && "$HTTP_CODE" != "200" ]]; then + printf 'API响应(%s):\n' "$HTTP_CODE" >&2 + python3 -m json.tool "$START_RESPONSE" 2>/dev/null >&2 || sed -n '1,20p' "$START_RESPONSE" >&2 + fail "创建采集会话失败" +fi + +SESSION_ID="$(python3 - "$START_RESPONSE" <<'PY' +import json, sys +with open(sys.argv[1], encoding="utf-8") as source: + value = json.load(source) +print(value.get("sessionId", "")) +PY +)" +[[ -n "$SESSION_ID" ]] || fail "API响应缺少 sessionId" + +START_STAMP="$(date '+%Y%m%d-%H%M%S')" +CAPTURE_DIR="$OUTPUT_DIR/device-${DEVICE_ID}_${START_STAMP}_${SESSION_ID}" +mkdir -p -- "$CAPTURE_DIR/raw" +cp -- "$START_RESPONSE" "$CAPTURE_DIR/start.json" +printf '%s\n' "$SESSION_ID" > "$CAPTURE_DIR/session-id.txt" +printf '会话已创建: %s\n结果目录: %s\n' "$SESSION_ID" "$CAPTURE_DIR" + +LATEST_TEMP="$TEMP_DIR/latest.json" +DEADLINE=$(( $(date +%s) + 10#$WAIT_TIMEOUT_SECONDS )) +FINAL_STATUS="" + +while (( $(date +%s) <= DEADLINE )); do + if ! HTTP_CODE="$(http_request GET "$CAPTURE_API/$SESSION_ID" "$LATEST_TEMP")"; then + printf 'WARN: 本次查询失败，%s秒后重试\n' "$POLL_SECONDS" >&2 + sleep "$POLL_SECONDS" + continue + fi + if [[ "$HTTP_CODE" != "200" ]]; then + printf 'WARN: 查询返回HTTP %s，%s秒后重试\n' "$HTTP_CODE" "$POLL_SECONDS" >&2 + sleep "$POLL_SECONDS" + continue + fi + + cp -- "$LATEST_TEMP" "$CAPTURE_DIR/latest.json.tmp" + mv -- "$CAPTURE_DIR/latest.json.tmp" "$CAPTURE_DIR/latest.json" + FINAL_STATUS="$(python3 - "$LATEST_TEMP" <<'PY' +import json, sys +with open(sys.argv[1], encoding="utf-8") as source: + print(json.load(source).get("status", "UNKNOWN")) +PY +)" + COUNTS="$(python3 - "$LATEST_TEMP" <<'PY' +import json, sys +with open(sys.argv[1], encoding="utf-8") as source: + value = json.load(source) +print("samples=%s, anomalies=%s, received=%s" % ( + len(value.get("messages") or []), value.get("retainedAnomalyCount", 0), + value.get("receivedCount", 0))) +PY +)" + printf '状态: %-10s %s\n' "$FINAL_STATUS" "$COUNTS" + if [[ "$FINAL_STATUS" == "COMPLETED" || "$FINAL_STATUS" == "CANCELLED" ]]; then + break + fi + sleep "$POLL_SECONDS" +done + +if [[ ! -f "$CAPTURE_DIR/latest.json" ]]; then + fail "等待期间没有取得有效结果；sessionId=$SESSION_ID" +fi + +cp -- "$CAPTURE_DIR/latest.json" "$CAPTURE_DIR/result.json" + +python3 - "$CAPTURE_DIR/result.json" "$CAPTURE_DIR" <<'PY' +import json +import os +import re +import sys + +result_path, output_dir = sys.argv[1], sys.argv[2] +with open(result_path, encoding="utf-8") as source: + result = json.load(source) + +messages = result.get("messages") or [] +jsonl_path = os.path.join(output_dir, "samples.jsonl") +with open(jsonl_path, "w", encoding="utf-8", newline="\n") as target: + for message in messages: + target.write(json.dumps(message, ensure_ascii=False, separators=(",", ":")) + "\n") + +raw_dir = os.path.join(output_dir, "raw") +os.makedirs(raw_dir, exist_ok=True) +for index, message in enumerate(messages, 1): + event_id = str(message.get("eventId") or "no-event-id") + safe_event_id = re.sub(r"[^A-Za-z0-9._-]", "_", event_id)[:80] + kind = "anomaly" if message.get("anomaly") else "normal" + path = os.path.join(raw_dir, "%03d_%s_%s.log" % (index, kind, safe_event_id)) + raw = message.get("rawMessage") + with open(path, "w", encoding="utf-8", newline="") as target: + target.write("" if raw is None else str(raw)) + +summary = [ + "sessionId=%s" % result.get("sessionId", ""), + "deviceId=%s" % result.get("deviceId", ""), + "status=%s" % result.get("status", ""), + "startedAt=%s" % result.get("startedAt", ""), + "captureEndsAt=%s" % result.get("captureEndsAt", ""), + "settleEndsAt=%s" % result.get("settleEndsAt", ""), + "receivedCount=%s" % result.get("receivedCount", 0), + "sampleCount=%s" % len(messages), + "retainedAnomalyCount=%s" % result.get("retainedAnomalyCount", 0), + "replacementCount=%s" % result.get("replacementCount", 0), +] +with open(os.path.join(output_dir, "summary.txt"), "w", encoding="utf-8", newline="\n") as target: + target.write("\n".join(summary) + "\n") +PY + +if [[ "$FINAL_STATUS" != "COMPLETED" && "$FINAL_STATUS" != "CANCELLED" ]]; then + printf 'WARN: 等待超时，已保存最后一次快照；会话仍会在服务端自动结束。\n' >&2 + printf '结果目录: %s\n' "$CAPTURE_DIR" + exit 2 +fi + +printf '采集完成。\n完整结果: %s/result.json\n原文目录: %s/raw\n' \ + "$CAPTURE_DIR" "$CAPTURE_DIR"