feat(syslog-consumer-rule): 增加多节点原始日志采集并修复天眼/微步JSON规则匹配

一、增加现场原始日志采集功能 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参数调整。
2026-09-14 23:11:45 +08:00
parent 7e6b2fc4dc
commit 1c07acc01d
17 changed files with 3131 additions and 43 deletions
@@ -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<RuleSelectionEngine.Candidate<ExecutionPlan>> candidates = collectCandidates(ctx);
-        if (ctx.invalidJson) {
+        if (candidates.isEmpty() && ctx.invalidJson) {
            return invalidJsonResult(ctx, deviceId);
        }
        RuleSelectionEngine.Decision<ExecutionPlan> decision =
@@ -505,9 +502,9 @@ public class LogNormalProcessor {
    /** 关闭切换开关时仅做归属影子计算，不改变旧的多结果行为。 */
    private void shadowSelect(ProcessContext ctx) {
        try {
-            List<RuleSelectionEngine.Candidate<ExecutionPlan>> candidates = collectCandidates(ctx);
-            if (ctx.invalidJson) return;
-            ruleSelectionEngine.select(candidates, ctx.strLogMsg, true);
+            ProcessContext shadow = shadowContext(ctx);
+            List<RuleSelectionEngine.Candidate<ExecutionPlan>> 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<String, Object> 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<String, Object> 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<String, Object> result1 = NestedJsonParserUtil.safeParseJson(strJson);
                if (result1 == null) return null;
                LinkedHashMap<String, Object> 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<Map<String, Object>> getRuleList(long device_id) {
        List<Map<String, Object>> rulelst = dmNormalizeRuleService.selectByDeviceId((long) 1);
        if (rulelst == null) {
@@ -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<PreparedMessage, RuntimeException> 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<String, String> record;
@@ -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<String, Object> 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<String, Object> 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<String, Object> map(ParsedMessage message) {
        Map<String, Object> destination = new LinkedHashMap<>();
        for (Map.Entry<String, Object> source : message.getFields().entrySet()) {
@@ -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<String, Object>> MAP_TYPE =
+            new TypeReference<Map<String, Object>>() { };
+
+    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<Definition> active(long now) {
+        redis.opsForZSet().removeRangeByScore(ACTIVE_KEY, 0D, now - 1D);
+        Set<String> ids = redis.opsForZSet().rangeByScore(ACTIVE_KEY, now, Double.MAX_VALUE);
+        if (ids == null || ids.isEmpty()) return Collections.emptyList();
+        List<Definition> 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<String> nodes = redis.opsForSet().members(nodesKey(sessionId));
+            List<RawLogCaptureService.CapturedLogSnapshot> first = new ArrayList<>();
+            List<RawLogCaptureService.CapturedLogSnapshot> 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<String, Object> 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<RawLogCaptureService.CapturedLogSnapshot> 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<Object, Object> values = redis.opsForHash().entries(metaKey(id));
+        return values == null || values.isEmpty() ? null : Definition.from(id, values);
+    }
+
+    private Map<String, Object> toMap(RawLogCaptureService.CaptureCandidates candidates) {
+        Map<String, Object> 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<RawLogCaptureService.CapturedLogSnapshot> messages(Object value) {
+        if (!(value instanceof List)) return Collections.emptyList();
+        List<RawLogCaptureService.CapturedLogSnapshot> result = new ArrayList<>();
+        for (Object item : (List<?>) value) {
+            if (!(item instanceof Map)) continue;
+            Map<String, Object> map = (Map<String, Object>) 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<RawLogCaptureService.CapturedLogSnapshot> first,
+                                  List<RawLogCaptureService.CapturedLogSnapshot> anomalies, int limit) {
+        int finalAnomalies = 0, initialAnomalies = 0;
+        for (RawLogCaptureService.CapturedLogSnapshot value
+                : RawLogCaptureService.mergeCandidates(first, anomalies, limit)) {
+            if (value.isAnomaly()) finalAnomalies++;
+        }
+        List<RawLogCaptureService.CapturedLogSnapshot> 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<String, String> toHash() {
+            Map<String, String> 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<Object, Object> 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; }
+    }
+}
@@ -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;
+
+/**
+ * 面向现场诊断的有界原始日志采样器。
+ *
+ * <p>普通消息按到达顺序填充固定容量；容量满后，新的异常消息会淘汰最早的非异常消息。
+ * 异常消息之间仍保持先到先得，避免高流量设备导致内存随消息数增长。</p>
+ */
+@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<String, CaptureSession> sessions = new ConcurrentHashMap<>();
+    private final ConcurrentHashMap<Long, CopyOnWriteArrayList<CaptureSession>> 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<CaptureSession> 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<CaptureSession> 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<RawLogCaptureClusterStore.Definition> definitions = clusterStore.active(now);
+            Map<String, Boolean> 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<CaptureSession> 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<CapturedLogSnapshot> mergeCandidates(List<CapturedLogSnapshot> firstCandidates,
+                                                      List<CapturedLogSnapshot> anomalyCandidates,
+                                                      int limit) {
+        Map<String, CapturedLogSnapshot> uniqueFirst = uniqueByEvent(firstCandidates);
+        List<CapturedLogSnapshot> first = new ArrayList<>(uniqueFirst.values());
+        first.sort(Comparator.comparingLong(CapturedLogSnapshot::getSequence));
+        if (first.size() > limit) first = new ArrayList<>(first.subList(0, limit));
+
+        Map<String, CapturedLogSnapshot> uniqueAnomalies = uniqueByEvent(anomalyCandidates);
+        List<CapturedLogSnapshot> anomalies = new ArrayList<>(uniqueAnomalies.values());
+        anomalies.sort(Comparator.comparingLong(CapturedLogSnapshot::getSequence));
+        if (anomalies.size() > limit) anomalies = anomalies.subList(0, limit);
+
+        Map<String, CapturedLogSnapshot> 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<String, CapturedLogSnapshot> 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<CapturedLogSnapshot> result = new ArrayList<>(selected.values());
+        result.sort(Comparator.comparingLong(CapturedLogSnapshot::getSequence));
+        return result;
+    }
+
+    private static Map<String, CapturedLogSnapshot> uniqueByEvent(
+            List<CapturedLogSnapshot> candidates) {
+        Map<String, CapturedLogSnapshot> 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<CapturedLogSnapshot> snapshots(List<CapturedLog> logs) {
+        List<CapturedLogSnapshot> result = new ArrayList<>(logs.size());
+        for (CapturedLog log : logs) result.add(log.snapshot());
+        return result;
+    }
+
+    private static int countAnomalies(List<CapturedLogSnapshot> messages) {
+        int count = 0;
+        for (CapturedLogSnapshot message : messages) if (message.isAnomaly()) count++;
+        return count;
+    }
+
+    private static long replacementCount(List<CapturedLogSnapshot> first,
+                                         List<CapturedLogSnapshot> anomalies, int limit) {
+        Map<String, CapturedLogSnapshot> firstSet = uniqueByEvent(first);
+        int laterAnomalies = 0;
+        for (CapturedLogSnapshot anomaly : uniqueByEvent(anomalies).values()) {
+            if (!firstSet.containsKey(eventKey(anomaly))) laterAnomalies++;
+        }
+        int availableNormals = 0;
+        List<CapturedLogSnapshot> 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<CapturedLog> firstMessages = new ArrayList<>();
+        private final List<CapturedLog> firstAnomalies = new ArrayList<>();
+        private final Map<String, CapturedLog> firstByEventId = new LinkedHashMap<>();
+        private final Map<String, CapturedLog> 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<CapturedLogSnapshot> first = snapshots(firstMessages);
+            List<CapturedLogSnapshot> anomalies = snapshots(firstAnomalies);
+            List<CapturedLogSnapshot> 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<CapturedLogSnapshot> firstMessages;
+        private final List<CapturedLogSnapshot> 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<CapturedLogSnapshot> firstMessages,
+                          List<CapturedLogSnapshot> 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<CapturedLogSnapshot> getFirstMessages() { return firstMessages; }
+        List<CapturedLogSnapshot> 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<CapturedLogSnapshot> 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<CapturedLogSnapshot> 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<CapturedLogSnapshot> 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);
+        }
+    }
+}
@@ -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<RawLogCaptureService.CaptureSnapshot> 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<RawLogCaptureService.CaptureSnapshot> get(
+            @PathVariable String sessionId) {
+        RawLogCaptureService.CaptureSnapshot snapshot = captureService.get(sessionId);
+        return snapshot == null ? ResponseEntity.notFound().build() : ResponseEntity.ok(snapshot);
+    }
+
+    @DeleteMapping("/{sessionId}")
+    public ResponseEntity<RawLogCaptureService.CaptureSnapshot> 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<Map<String, String>> disabled(
+            RawLogCaptureService.CaptureDisabledException failure) {
+        return ResponseEntity.status(503)
+                .body(Collections.singletonMap("message", failure.getMessage()));
+    }
+
+    @ExceptionHandler(RawLogCaptureService.CaptureCapacityException.class)
+    public ResponseEntity<Map<String, String>> capacity(
+            RawLogCaptureService.CaptureCapacityException failure) {
+        return ResponseEntity.status(429)
+                .body(Collections.singletonMap("message", failure.getMessage()));
+    }
+
+    @ExceptionHandler(RawLogCaptureService.CaptureCoordinationException.class)
+    public ResponseEntity<Map<String, String>> 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; }
+    }
+}
@@ -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
@@ -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\":{}}}";
@@ -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<RawLogCaptureService.CapturedLogSnapshot> 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<RawLogCaptureService.CapturedLogSnapshot> anomalies =
+                Arrays.asList(snapshot(200, "a200", true));
+
+        List<RawLogCaptureService.CapturedLogSnapshot> 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<String> 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; }
+    }
+}