source = new HashMap<>();
+ source.put("_time", eventTime);
+ source.put("title", ALARM_TITLES[random.nextInt(ALARM_TITLES.length)]);
+ source.put("alarm_sev", 1 + random.nextInt(5));
+ source.put("src_ip", randomIp(random));
+ source.put("dst_ip", DST_IPS[random.nextInt(DST_IPS.length)]);
+ source.put("src_port", 1024 + random.nextInt(60000));
+ source.put("dst_port", 80 + random.nextInt(1000));
+ source.put("proto", PROTOS[random.nextInt(PROTOS.length)]);
+ source.put("rule_id", "TY-RULE-" + (1000 + random.nextInt(9000)));
+ source.put("event_id", "ty_al_" + eventTime + "_" + seq + "_" + random.nextInt(99999));
+ source.put("attack_type", ALARM_TITLES[random.nextInt(ALARM_TITLES.length)]);
+ return source;
+ }
+
+ /**
+ * 随机内网 IP
+ */
+ private static String randomIp(Random random) {
+ String prefix = SRC_IP_PREFIX[random.nextInt(SRC_IP_PREFIX.length)];
+ return prefix + "." + (random.nextInt(254) + 1);
+ }
+
+ /**
+ * 随机十六进制字符串（md5/sha256）
+ */
+ private static String randomHex(Random random, int length) {
+ StringBuilder sb = new StringBuilder();
+ String chars = "0123456789abcdef";
+ for (int i = 0; i < length; i++) {
+ sb.append(chars.charAt(random.nextInt(chars.length())));
+ }
+ return sb.toString();
+ }
+}
diff --git a/haobang-security-dm/syslog-collector-api/src/main/java/com/mock/TianyanMockController.java b/haobang-security-dm/syslog-collector-api/src/main/java/com/mock/TianyanMockController.java
new file mode 100644
index 0000000..9ab387b
--- /dev/null
+++ b/haobang-security-dm/syslog-collector-api/src/main/java/com/mock/TianyanMockController.java
@@ -0,0 +1,216 @@
+package com.mock;
+
+import com.alibaba.fastjson2.JSONArray;
+import com.alibaba.fastjson2.JSONObject;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestHeader;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.Map;
+
+/**
+ * 天眼(Tianyan) API Mock 接口
+ *
+ * 模拟真实天眼的认证与日志查询接口，供功能测试使用。
+ * 采集器将 device_api_config.host_api 配置为
+ * http://127.0.0.1:8290/collectorapi/mock/tianyan 即可无缝对接。
+ *
+ * 模拟接口：
+ * 1. POST /mock/tianyan/skyeye/v1/admin/auth 免密登录，返回 access_token
+ * 2. GET /mock/tianyan/skyeye/v1/admin/auth 提取 csrf_token
+ * 3. GET /mock/tianyan/skyeye/v1/analysis/log-search/list 日志查询（weblog）
+ * 4. GET /mock/tianyan/skyeye/v1/alarm/alarm/list 告警查询（alarm）
+ *
+ * 开关配置：mock.server.enabled=true 时启用，默认关闭（生产不暴露）。
+ */
+@RestController
+@RequestMapping("/mock/tianyan")
+@ConditionalOnProperty(name = "mock.server.enabled", havingValue = "true")
+public class TianyanMockController {
+
+ private static final Logger log = LoggerFactory.getLogger(TianyanMockController.class);
+
+ /** 模拟数据总量（每类），分页测试用，默认250条 */
+ @Value("${mock.server.total-count:250}")
+ private int totalCount;
+
+ /**
+ * 免密登录，返回 access_token
+ * 请求：POST /skyeye/v1/admin/auth
+ * Headers: X-Authorization=SHA256(JSON(jsonBody)+timestamp+clientSecret)
+ * Body: {"client_id":"xxx","username":"xxx"}
+ */
+ @PostMapping("/skyeye/v1/admin/auth")
+ public String auth(@RequestBody(required = false) String body,
+ @RequestHeader(value = "X-Authorization", required = false) String xAuthorization) {
+ log.info("[Mock] 天眼认证请求, X-Authorization={}, body={}", maskAuth(xAuthorization), body);
+ JSONObject resp = new JSONObject();
+ resp.put("status", 1000);
+ resp.put("message", "success");
+ JSONObject inner = new JSONObject();
+ inner.put("access_token", "mock_tianyan_token_" + System.currentTimeMillis());
+ JSONObject data = new JSONObject();
+ data.put("data", inner);
+ resp.put("data", data);
+ return resp.toJSONString();
+ }
+
+ /**
+ * 提取 csrf_token
+ * 请求：GET /skyeye/v1/admin/auth?token=access_token
+ */
+ @GetMapping("/skyeye/v1/admin/auth")
+ public String getCsrf(@RequestParam(value = "token", required = false) String token) {
+ log.info("[Mock] 天眼获取 csrf_token, token={}", maskToken(token));
+ JSONObject resp = new JSONObject();
+ resp.put("status", 1000);
+ resp.put("message", "success");
+ JSONObject data = new JSONObject();
+ data.put("csrf_token", "mock_tianyan_csrf_" + System.currentTimeMillis());
+ resp.put("data", data);
+ return resp.toJSONString();
+ }
+
+ /**
+ * 日志查询（weblog / alarm）
+ * 请求：GET /skyeye/v1/analysis/log-search/list
+ * 参数：branch_id, start_time(ms), end_time(ms), index, category, offset, limit, mode, csrf_token, r
+ * 响应：{"status":1000,"data":{"data":{"search":{"hits":[{"_source":{...}}]}}}}
+ *
+ * 可选场景参数 mock_scenario：
+ * - empty 返回空数据
+ * - csrf_expired 返回 csrf_token error（触发采集器重新认证）
+ * - delay 延迟 5 秒返回（模拟超时）
+ */
+ @GetMapping("/skyeye/v1/analysis/log-search/list")
+ public String searchLog(@RequestParam(value = "index", required = false) String index,
+ @RequestParam(value = "offset", defaultValue = "1") int offset,
+ @RequestParam(value = "limit", defaultValue = "100") int limit,
+ @RequestParam(value = "start_time", required = false) Long startTime,
+ @RequestParam(value = "end_time", required = false) Long endTime,
+ @RequestParam(value = "mock_scenario", required = false) String scenario) throws InterruptedException {
+ boolean isAlarm = index != null && index.toLowerCase().contains("alarm");
+ String dataType = isAlarm ? "alarm" : "weblog";
+ log.info("[Mock] 天眼日志查询: index={}, dataType={}, offset={}, limit={}, start={}, end={}, scenario={}",
+ index, dataType, offset, limit, startTime, endTime, scenario);
+
+ JSONObject resp = new JSONObject();
+
+ // 模拟场景：csrf_token 过期
+ if ("csrf_expired".equalsIgnoreCase(scenario)) {
+ resp.put("status", 2000);
+ resp.put("message", "csrf_token error");
+ return resp.toJSONString();
+ }
+
+ // 模拟场景：响应延迟
+ if ("delay".equalsIgnoreCase(scenario)) {
+ Thread.sleep(5000);
+ }
+
+ resp.put("status", 1000);
+ resp.put("message", "success");
+
+ // 模拟场景：空数据
+ long now = System.currentTimeMillis();
+ long start = startTime != null && startTime > 0 ? startTime : now - 60 * 60 * 1000L;
+ long end = endTime != null && endTime > 0 ? endTime : now;
+ int size = "empty".equalsIgnoreCase(scenario) ? 0
+ : Math.max(0, Math.min(limit, totalCount - (offset - 1)));
+
+ JSONArray hits = new JSONArray();
+ for (int i = 0; i < size; i++) {
+ Map hit = MockDataGenerator.buildTianyanHit(dataType, start, end, offset + i);
+ hits.add(new JSONObject(hit));
+ }
+
+ JSONObject search = new JSONObject();
+ search.put("hits", hits);
+ JSONObject inner = new JSONObject();
+ inner.put("search", search);
+ JSONObject data = new JSONObject();
+ data.put("data", inner);
+ resp.put("data", data);
+ return resp.toJSONString();
+ }
+
+ /**
+ * 告警查询（alarm）
+ * 请求：GET /skyeye/v1/alarm/alarm/list
+ * 参数：start_time(ms), end_time(ms), hazard_level, status, host_state, limit, offset, csrf_token, r
+ * 响应：{"status":1000,"data":{"data":{"rows":[{"_source":{...}}]}}}
+ *
+ * 可选场景参数 mock_scenario：
+ * - empty 返回空数据
+ * - csrf_expired 返回 csrf_token error（触发采集器重新认证）
+ * - delay 延迟 5 秒返回（模拟超时）
+ */
+ @GetMapping("/skyeye/v1/alarm/alarm/list")
+ public String alarmList(@RequestParam(value = "offset", defaultValue = "1") int offset,
+ @RequestParam(value = "limit", defaultValue = "100") int limit,
+ @RequestParam(value = "start_time", required = false) Long startTime,
+ @RequestParam(value = "end_time", required = false) Long endTime,
+ @RequestParam(value = "mock_scenario", required = false) String scenario) throws InterruptedException {
+ log.info("[Mock] 天眼告警查询: offset={}, limit={}, start={}, end={}, scenario={}",
+ offset, limit, startTime, endTime, scenario);
+
+ JSONObject resp = new JSONObject();
+
+ // 模拟场景：csrf_token 过期
+ if ("csrf_expired".equalsIgnoreCase(scenario)) {
+ resp.put("status", 2000);
+ resp.put("message", "csrf_token error");
+ return resp.toJSONString();
+ }
+
+ // 模拟场景：响应延迟
+ if ("delay".equalsIgnoreCase(scenario)) {
+ Thread.sleep(5000);
+ }
+
+ resp.put("status", 1000);
+ resp.put("message", "success");
+
+ // 模拟场景：空数据
+ long now = System.currentTimeMillis();
+ long start = startTime != null && startTime > 0 ? startTime : now - 60 * 60 * 1000L;
+ long end = endTime != null && endTime > 0 ? endTime : now;
+ int size = "empty".equalsIgnoreCase(scenario) ? 0
+ : Math.max(0, Math.min(limit, totalCount - (offset - 1)));
+
+ JSONArray rows = new JSONArray();
+ for (int i = 0; i < size; i++) {
+ Map row = MockDataGenerator.buildTianyanHit("alarm", start, end, offset + i);
+ rows.add(new JSONObject(row));
+ }
+
+ JSONObject inner = new JSONObject();
+ inner.put("rows", rows);
+ JSONObject data = new JSONObject();
+ data.put("data", inner);
+ resp.put("data", data);
+ return resp.toJSONString();
+ }
+
+ private String maskAuth(String v) {
+ if (v == null || v.length() < 8) {
+ return v;
+ }
+ return v.substring(0, 8) + "***";
+ }
+
+ private String maskToken(String v) {
+ if (v == null || v.length() < 8) {
+ return v;
+ }
+ return v.substring(0, 8) + "***";
+ }
+}
diff --git a/haobang-security-dm/syslog-collector-api/src/main/java/com/mock/WeibuMockController.java b/haobang-security-dm/syslog-collector-api/src/main/java/com/mock/WeibuMockController.java
new file mode 100644
index 0000000..4659f72
--- /dev/null
+++ b/haobang-security-dm/syslog-collector-api/src/main/java/com/mock/WeibuMockController.java
@@ -0,0 +1,137 @@
+package com.mock;
+
+import com.alibaba.fastjson2.JSONArray;
+import com.alibaba.fastjson2.JSONObject;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+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.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.Map;
+
+/**
+ * 微步(Weibu) API Mock 接口
+ *
+ * 模拟真实微步的数据查询接口，供功能测试使用。
+ * 采集器将 device_api_config.host_api 配置为
+ * http://127.0.0.1:8290/collectorapi/mock/weibu 即可无缝对接。
+ *
+ * 模拟接口：
+ * 1. POST /mock/weibu/api/v1/log/searchBySql 数据查询（attack/info）
+ *
+ * 开关配置：mock.server.enabled=true 时启用，默认关闭（生产不暴露）。
+ */
+@RestController
+@RequestMapping("/mock/weibu")
+@ConditionalOnProperty(name = "mock.server.enabled", havingValue = "true")
+public class WeibuMockController {
+
+ private static final Logger log = LoggerFactory.getLogger(WeibuMockController.class);
+
+ /** 模拟数据总量（每类），分页测试用，默认250条 */
+ @Value("${mock.server.total-count:250}")
+ private int totalCount;
+
+ /**
+ * 数据查询
+ * 请求：POST /api/v1/log/searchBySql?api_key=xxx&auth_timestamp=xxx&sign=xxx
+ * Body: {"net_data_type":["attack"],"time_from":秒,"time_to":秒,
+ * "assets_group":[],"search_connection":false,"sql":"","columns":["*"],
+ * "asc":true,"page":{"cur_page":1,"page_size":100}}
+ * 响应：{"response_code":0,"response_msg":"OK","data":{"data":[...]}}
+ *
+ * 可选场景参数 mock_scenario：
+ * - empty 返回空数据
+ * - sign_error 返回签名错误（response_code 非0，触发采集失败分支）
+ * - delay 延迟 5 秒返回（模拟超时）
+ */
+ @PostMapping("/api/v1/log/searchBySql")
+ public String searchBySql(@RequestParam(value = "api_key", required = false) String apiKey,
+ @RequestParam(value = "auth_timestamp", required = false) String authTimestamp,
+ @RequestParam(value = "sign", required = false) String sign,
+ @RequestParam(value = "mock_scenario", required = false) String scenario,
+ @RequestBody(required = false) String body) throws InterruptedException {
+ JSONObject req = body != null ? JSONObject.parseObject(body) : new JSONObject();
+ log.info("[Mock] 微步查询请求: api_key={}, ts={}, sign={}, body={}, scenario={}",
+ maskAuth(apiKey), authTimestamp, maskAuth(sign), req, scenario);
+
+ JSONObject resp = new JSONObject();
+
+ // 模拟场景：签名错误
+ if ("sign_error".equalsIgnoreCase(scenario)) {
+ resp.put("response_code", 401);
+ resp.put("response_msg", "signature verification failed");
+ return resp.toJSONString();
+ }
+
+ // 模拟场景：响应延迟
+ if ("delay".equalsIgnoreCase(scenario)) {
+ Thread.sleep(5000);
+ }
+
+ // 解析 net_data_type（attack/info）
+ String dataType = "attack";
+ JSONArray netDataType = req.getJSONArray("net_data_type");
+ if (netDataType != null && !netDataType.isEmpty()) {
+ String first = netDataType.getString(0);
+ if (first != null) {
+ dataType = first.toLowerCase();
+ }
+ }
+
+ // 解析时间窗口（秒级 -> 毫秒），缺失时默认最近1小时
+ long nowSec = System.currentTimeMillis() / 1000;
+ long timeFromSec = req.getLongValue("time_from", 0L);
+ long timeToSec = req.getLongValue("time_to", 0L);
+ long start;
+ long end;
+ if (timeFromSec > 0 && timeToSec > 0 && timeToSec >= timeFromSec) {
+ start = timeFromSec * 1000;
+ end = timeToSec * 1000;
+ } else {
+ start = (nowSec - 60 * 60) * 1000;
+ end = nowSec * 1000;
+ }
+
+ // 解析分页
+ int curPage = 1;
+ int pageSize = 100;
+ JSONObject page = req.getJSONObject("page");
+ if (page != null) {
+ curPage = page.getIntValue("cur_page", 1);
+ pageSize = page.getIntValue("page_size", 100);
+ }
+
+ log.info("[Mock] 微步查询: dataType={}, start={}, end={}, curPage={}, pageSize={}",
+ dataType, start, end, curPage, pageSize);
+
+ // 模拟场景：空数据
+ int size = "empty".equalsIgnoreCase(scenario) ? 0
+ : Math.max(0, Math.min(pageSize, totalCount - (curPage - 1) * pageSize));
+
+ JSONArray dataList = new JSONArray();
+ for (int i = 0; i < size; i++) {
+ Map item = MockDataGenerator.buildWeibuItem(dataType, start, end, (curPage - 1) * pageSize + i + 1);
+ dataList.add(new JSONObject(item));
+ }
+
+ resp.put("response_code", 0);
+ resp.put("response_msg", "OK");
+ JSONObject data = new JSONObject();
+ data.put("data", dataList);
+ resp.put("data", data);
+ return resp.toJSONString();
+ }
+
+ private String maskAuth(String v) {
+ if (v == null || v.length() < 8) {
+ return v;
+ }
+ return v.substring(0, 8) + "***";
+ }
+}
diff --git a/haobang-security-dm/syslog-collector-api/src/main/java/com/util/HttpUtil.java b/haobang-security-dm/syslog-collector-api/src/main/java/com/util/HttpUtil.java
new file mode 100644
index 0000000..3969e1d
--- /dev/null
+++ b/haobang-security-dm/syslog-collector-api/src/main/java/com/util/HttpUtil.java
@@ -0,0 +1,130 @@
+package com.util;
+
+import org.springframework.http.HttpEntity;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpMethod;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.client.RestTemplate;
+
+import java.net.URI;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * HTTP 请求工具（基于 RestTemplate）
+ */
+public class HttpUtil {
+
+ private HttpUtil() {
+ }
+
+ /**
+ * 将已构建好的 URL 字符串转为 URI 对象
+ * URL 中的参数已按平台要求预编码（如 branch_id 中的 %2F），
+ * 必须以 URI 类型传给 RestTemplate，否则默认 UriTemplateHandler 会对 % 再次编码（%2F → %252F）
+ */
+ private static URI toUri(String url) {
+ return URI.create(url);
+ }
+
+ /**
+ * 发送 GET 请求
+ * @param restTemplate RestTemplate
+ * @param url URL
+ * @param headers 请求头（可为 null）
+ * @return 响应体字符串
+ */
+ public static String get(RestTemplate restTemplate, String url, Map headers) {
+ HttpHeaders httpHeaders = new HttpHeaders();
+ if (headers != null) {
+ headers.forEach(httpHeaders::set);
+ }
+ HttpEntity entity = new HttpEntity<>(httpHeaders);
+ ResponseEntity response = restTemplate.exchange(toUri(url), HttpMethod.GET, entity, String.class);
+ return response.getBody();
+ }
+
+ /**
+ * 发送 GET 请求，同时返回响应体与 Set-Cookie（供需要登录 Cookie 的接口使用，如天眼免密登录）
+ * @param restTemplate RestTemplate
+ * @param url URL
+ * @param headers 请求头（可为 null）
+ * @return 响应结果（body + Set-Cookie 原始列表）
+ */
+ public static GetResult getWithCookies(RestTemplate restTemplate, String url, Map headers) {
+ HttpHeaders httpHeaders = new HttpHeaders();
+ if (headers != null) {
+ headers.forEach(httpHeaders::set);
+ }
+ HttpEntity entity = new HttpEntity<>(httpHeaders);
+ ResponseEntity response = restTemplate.exchange(toUri(url), HttpMethod.GET, entity, String.class);
+ List setCookies = response.getHeaders() != null
+ ? response.getHeaders().get(HttpHeaders.SET_COOKIE) : null;
+ return new GetResult(response.getBody(), setCookies);
+ }
+
+ /**
+ * GET 请求响应结果封装
+ */
+ public static class GetResult {
+ private final String body;
+ private final List setCookies;
+
+ GetResult(String body, List setCookies) {
+ this.body = body;
+ this.setCookies = setCookies;
+ }
+
+ public String getBody() {
+ return body;
+ }
+
+ /** Set-Cookie 原始串列表（可能为 null/空） */
+ public List getSetCookies() {
+ return setCookies == null ? Collections.emptyList() : setCookies;
+ }
+ }
+
+ /**
+ * 发送 POST 请求（JSON 体）
+ * @param restTemplate RestTemplate
+ * @param url URL
+ * @param jsonBody JSON 请求体
+ * @param headers 请求头（可为 null）
+ * @return 响应体字符串
+ */
+ public static String postJson(RestTemplate restTemplate, String url, String jsonBody, Map headers) {
+ HttpHeaders httpHeaders = new HttpHeaders();
+ if (headers != null) {
+ headers.forEach(httpHeaders::set);
+ }
+ httpHeaders.set(HttpHeaders.CONTENT_TYPE, "application/json");
+ HttpEntity entity = new HttpEntity<>(jsonBody, httpHeaders);
+ ResponseEntity response = restTemplate.exchange(toUri(url), HttpMethod.POST, entity, String.class);
+ return response.getBody();
+ }
+
+ /**
+ * 发送 POST 请求（表单体）
+ * @param restTemplate RestTemplate
+ * @param url URL
+ * @param formParams 表单参数
+ * @param headers 请求头（可为 null）
+ * @return 响应体字符串
+ */
+ public static String postForm(RestTemplate restTemplate, String url, Map formParams, Map headers) {
+ HttpHeaders httpHeaders = new HttpHeaders();
+ if (headers != null) {
+ headers.forEach(httpHeaders::set);
+ }
+ httpHeaders.set(HttpHeaders.CONTENT_TYPE, "application/x-www-form-urlencoded");
+ org.springframework.util.LinkedMultiValueMap body = new org.springframework.util.LinkedMultiValueMap<>();
+ if (formParams != null) {
+ formParams.forEach(body::add);
+ }
+ HttpEntity> entity = new HttpEntity<>(body, httpHeaders);
+ ResponseEntity response = restTemplate.exchange(toUri(url), HttpMethod.POST, entity, String.class);
+ return response.getBody();
+ }
+}
diff --git a/haobang-security-dm/syslog-collector-api/src/main/java/com/util/Sm4Util.java b/haobang-security-dm/syslog-collector-api/src/main/java/com/util/Sm4Util.java
new file mode 100644
index 0000000..5d03f88
--- /dev/null
+++ b/haobang-security-dm/syslog-collector-api/src/main/java/com/util/Sm4Util.java
@@ -0,0 +1,114 @@
+package com.util;
+
+import cn.hutool.core.util.HexUtil;
+import cn.hutool.core.util.RandomUtil;
+import cn.hutool.crypto.symmetric.SM4;
+import cn.hutool.crypto.symmetric.SymmetricCrypto;
+import org.bouncycastle.jce.provider.BouncyCastleProvider;
+
+import javax.crypto.SecretKey;
+import javax.crypto.spec.IvParameterSpec;
+import javax.crypto.spec.SecretKeySpec;
+import java.security.Security;
+import java.util.Base64;
+
+/**
+ * 国密 SM4 加解密工具类
+ * 基于 Hutool + Bouncy Castle 实现，与 syslog-serve / syslog-consumer 保持一致
+ * 支持 ECB 和 CBC 模式；推荐 CBC 模式，输出格式 "Base64(IV):Base64(密文)"
+ */
+public class Sm4Util {
+
+ static {
+ // 统一注册 Provider，确保全局生效
+ if (Security.getProvider(BouncyCastleProvider.PROVIDER_NAME) == null) {
+ Security.addProvider(new BouncyCastleProvider());
+ }
+ }
+
+ /**
+ * 生成 SM4 密钥（16字节，128位）
+ *
+ * @return 十六进制字符串格式的密钥
+ */
+ public static String generateKey() {
+ byte[] key = cn.hutool.crypto.SmUtil.sm4().getSecretKey().getEncoded();
+ return HexUtil.encodeHexStr(key);
+ }
+
+ // ==================== ECB 模式（电子密码本模式） ====================
+ // 注意：ECB 模式不需要 IV，但安全性较低，不建议用于加密大量数据
+
+ /**
+ * ECB 模式加密
+ *
+ * @param plainText 明文
+ * @param hexKey 十六进制字符串格式的密钥
+ * @return Base64 编码的密文
+ */
+ public static String encryptEcb(String plainText, String hexKey) {
+ SM4 sm4 = new SM4(HexUtil.decodeHex(hexKey));
+ return sm4.encryptBase64(plainText);
+ }
+
+ /**
+ * ECB 模式解密
+ *
+ * @param cipherText Base64 编码的密文
+ * @param hexKey 十六进制字符串格式的密钥
+ * @return 明文
+ */
+ public static String decryptEcb(String cipherText, String hexKey) {
+ SM4 sm4 = new SM4(HexUtil.decodeHex(hexKey));
+ return sm4.decryptStr(cipherText);
+ }
+
+ // ==================== CBC 模式（密码分组链接模式，推荐） ====================
+ // 需要随机 IV（初始化向量），安全性更高；密文携带 IV，格式 "Base64(IV):Base64(密文)"
+
+ /**
+ * CBC 模式加密（推荐）
+ *
+ * @param plainText 明文
+ * @param hexKey 十六进制字符串格式的密钥
+ * @return 格式为 "Base64(IV):Base64(密文)" 的字符串
+ */
+ public static String encryptCbc(String plainText, String hexKey) {
+ // 1. 随机生成16字节 IV
+ byte[] iv = RandomUtil.randomBytes(16);
+ IvParameterSpec ivSpec = new IvParameterSpec(iv);
+ // 2. 将十六进制密钥转为字节数组并包装为 SecretKey
+ byte[] keyBytes = HexUtil.decodeHex(hexKey);
+ SecretKey secretKey = new SecretKeySpec(keyBytes, "SM4");
+ // 3. 创建加解密器
+ SymmetricCrypto sm4 = new SymmetricCrypto("SM4/CBC/PKCS5Padding", secretKey, ivSpec);
+ // 4. 执行加密
+ String encryptedBase64 = sm4.encryptBase64(plainText);
+ String ivBase64 = Base64.getEncoder().encodeToString(iv);
+ return ivBase64 + ":" + encryptedBase64;
+ }
+
+ /**
+ * CBC 模式解密
+ *
+ * @param cipherTextWithIv 由 {@link #encryptCbc(String, String)} 生成的字符串
+ * @param hexKey 十六进制字符串格式的密钥
+ * @return 明文
+ */
+ public static String decryptCbc(String cipherTextWithIv, String hexKey) {
+ String[] parts = cipherTextWithIv.split(":");
+ if (parts.length != 2) {
+ throw new IllegalArgumentException("无效的密文格式，应为 'Base64(IV):Base64(密文)'");
+ }
+ // 1. 解析 IV
+ byte[] iv = Base64.getDecoder().decode(parts[0]);
+ IvParameterSpec ivSpec = new IvParameterSpec(iv);
+ // 2. 解析密钥
+ byte[] keyBytes = HexUtil.decodeHex(hexKey);
+ SecretKey secretKey = new SecretKeySpec(keyBytes, "SM4");
+ // 3. 创建解密器（算法参数必须与加密时完全一致）
+ SymmetricCrypto sm4 = new SymmetricCrypto("SM4/CBC/PKCS5Padding", secretKey, ivSpec);
+ // 4. 解密
+ return sm4.decryptStr(parts[1]);
+ }
+}
diff --git a/haobang-security-dm/syslog-collector-api/src/main/java/com/util/SpringContextUtil.java b/haobang-security-dm/syslog-collector-api/src/main/java/com/util/SpringContextUtil.java
new file mode 100644
index 0000000..395d2f0
--- /dev/null
+++ b/haobang-security-dm/syslog-collector-api/src/main/java/com/util/SpringContextUtil.java
@@ -0,0 +1,40 @@
+package com.util;
+
+import org.springframework.beans.BeansException;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+import org.springframework.stereotype.Component;
+
+/**
+ * Spring 上下文工具类
+ */
+@Component
+public class SpringContextUtil implements ApplicationContextAware {
+
+ private static ApplicationContext applicationContext;
+
+ @Override
+ public void setApplicationContext(ApplicationContext context) throws BeansException {
+ applicationContext = context;
+ }
+
+ public static ApplicationContext getApplicationContext() {
+ return applicationContext;
+ }
+
+ public static Object getBean(String name) {
+ return applicationContext.getBean(name);
+ }
+
+ public static T getBean(Class clazz) {
+ return applicationContext.getBean(clazz);
+ }
+
+ public static T getBean(String name, Class clazz) {
+ return applicationContext.getBean(name, clazz);
+ }
+
+ public static boolean containsBean(String name) {
+ return applicationContext.containsBean(name);
+ }
+}
diff --git a/haobang-security-dm/syslog-collector-api/src/main/java/com/util/TimeUtils.java b/haobang-security-dm/syslog-collector-api/src/main/java/com/util/TimeUtils.java
new file mode 100644
index 0000000..4503c3c
--- /dev/null
+++ b/haobang-security-dm/syslog-collector-api/src/main/java/com/util/TimeUtils.java
@@ -0,0 +1,80 @@
+package com.util;
+
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+
+/**
+ * 时间工具类
+ */
+public class TimeUtils {
+
+ /** 默认时间格式 */
+ public static final DateTimeFormatter DEFAULT_FORMATTER =
+ DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
+
+ /** 紧凑时间格式：yyyyMMddHHmmss */
+ public static final DateTimeFormatter COMPACT_FORMATTER =
+ DateTimeFormatter.ofPattern("yyyyMMddHHmmss");
+
+ /** 带毫秒格式：yyyyMMddHHmmssSSS */
+ public static final DateTimeFormatter MS_FORMATTER =
+ DateTimeFormatter.ofPattern("yyyyMMddHHmmssSSS");
+
+ /** 天眼时间格式：yyyy-MM-dd'T'HH:mm:ss.SSSZ */
+ public static final DateTimeFormatter TIANYAN_FORMATTER =
+ DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'");
+
+ private TimeUtils() {
+ }
+
+ /**
+ * 毫秒时间戳 -> LocalDateTime
+ */
+ public static LocalDateTime toLocalDateTime(long timestampMs) {
+ return LocalDateTime.ofInstant(Instant.ofEpochMilli(timestampMs), ZoneId.systemDefault());
+ }
+
+ /**
+ * LocalDateTime -> 毫秒时间戳
+ */
+ public static long toTimestamp(LocalDateTime dateTime) {
+ return dateTime.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
+ }
+
+ /**
+ * 当前毫秒时间戳
+ */
+ public static long now() {
+ return System.currentTimeMillis();
+ }
+
+ /**
+ * 当前秒级时间戳
+ */
+ public static long nowSecond() {
+ return System.currentTimeMillis() / 1000;
+ }
+
+ /**
+ * 格式化 LocalDateTime -> yyyy-MM-dd HH:mm:ss
+ */
+ public static String format(LocalDateTime dateTime) {
+ return dateTime.format(DEFAULT_FORMATTER);
+ }
+
+ /**
+ * 毫秒时间戳 -> yyyy-MM-dd HH:mm:ss
+ */
+ public static String format(long timestampMs) {
+ return toLocalDateTime(timestampMs).format(DEFAULT_FORMATTER);
+ }
+
+ /**
+ * 毫秒时间戳 -> yyyyMMddHHmmssSSS（用于 receive_time）
+ */
+ public static String formatCompactMs(long timestampMs) {
+ return toLocalDateTime(timestampMs).format(MS_FORMATTER);
+ }
+}
diff --git a/haobang-security-dm/syslog-collector-api/src/main/resources/application-prod.properties b/haobang-security-dm/syslog-collector-api/src/main/resources/application-prod.properties
new file mode 100644
index 0000000..c766fdd
--- /dev/null
+++ b/haobang-security-dm/syslog-collector-api/src/main/resources/application-prod.properties
@@ -0,0 +1,114 @@
+#Server Configuration
+server.port=8290
+server.servlet.context-path=/collectorapi
+server.tomcat.uri-encoding=UTF-8
+server.error.include-message=always
+server.error.include-binding-errors=always
+
+# ============================================
+# 采集调度配置
+# ============================================
+# 定时采集间隔（毫秒），默认1分钟
+device.collect.interval=60000
+# 全量回补功能开关（false=禁用，POST /api/collect/full 将拒绝执行）
+device.collect.full.enabled=false
+# 重试配置
+device.collect.retry.max-attempts=3
+device.collect.retry.initial-delay=5000
+device.collect.retry.multiplier=2.0
+# Kafka 推送 topic
+device.collect.kafka.topic=${KAFKA_COLLECTOR_TOPIC:test-topic}
+# 采集线程池大小：各设备/数据类型并行采集，慢设备不阻塞其他设备
+device.collect.pool-size=${COLLECT_POOL_SIZE:15}
+# 微步 info（日志流量）单轮采集窗口上限（分钟），积压时逐轮追赶，0=不限(用默认30分钟上限)
+device.weibu.info-max-window-minutes=${WEIBU_INFO_MAX_WINDOW_MINUTES:5}
+#采集探针ID
+app.service.device_collect_id=${DEVICE_COLLECT_ID:1}
+# ============================================
+# 天眼采集配置
+# ============================================
+device.tianyan.host=${TIANYAN_HOST:https://10.56.170.107}
+# 免密登录 login_key（由平台管理员分配）
+device.tianyan.login-key=${TIANYAN_LOGIN_KEY:QuJm4NwF3GMaXb2Z}
+device.tianyan.username=${TIANYAN_USERNAME:xxadmin}
+device.tianyan.branch-id=${TIANYAN_BRANCH_ID:QbJK%2FsLHi}
+# 采集数据类型，逗号分隔：weblog(访问日志),alarm(告警日志)
+device.tianyan.data-types=${TIANYAN_DATA_TYPES:weblog,alarm}
+device.tianyan.page-size=${TIANYAN_PAGE_SIZE:1000}
+# 天眼单次查询子窗口跨度（分钟），采集窗口切片预防平台深分页上限，<=0 不切片
+device.tianyan.sub-window-minutes=${TIANYAN_SUB_WINDOW_MINUTES:5}
+
+# ============================================
+# 微步采集配置
+# ============================================
+device.weibu.host=${WEIBU_HOST:https://10.56.171.164}
+device.weibu.api-key=${WEIBU_API_KEY:57d8fb2b556a471ea36b8b4441fa1c0c}
+device.weibu.secret=${WEIBU_SECRET:9d967876f51d4e16bbf8bde069d12a9b}
+# 采集数据类型，逗号分隔：attack(攻击日志),info(情报日志)
+device.weibu.data-types=${WEIBU_DATA_TYPES:attack,info}
+# 微步限流：每秒最大请求数
+device.weibu.rate-limit=${WEIBU_RATE_LIMIT:10}
+# 微步单次取数上限（size 两段式取数一次取全的条数上限，超出对半拆分窗口重采）
+device.weibu.fetch-size-cap=${WEIBU_FETCH_SIZE_CAP:20000}
+
+# ============================================
+# XDR 采集配置（暂缓）
+# ============================================
+device.xdr.host=${XDR_HOST:https://127.0.0.1}
+device.xdr.auth-code=${XDR_AUTH_CODE:}
+device.xdr.api-base=${XDR_API_BASE:/api/openapi/}
+device.xdr.data-types=${XDR_DATA_TYPES:}
+
+# ============================================
+# Kafka 配置
+# ============================================
+spring.kafka.producer.bootstrap-servers=192.168.4.99:9092
+spring.kafka.producer.properties.retries=10
+spring.kafka.producer.properties.retry.backoff.ms=500
+spring.kafka.producer.properties.connections.max.idle.ms=600000
+spring.kafka.producer.properties.socket.keepalive.enable=true
+spring.kafka.producer.properties.request.timeout.ms=30000
+spring.kafka.producer.properties.delivery.timeout.ms=120000
+
+# SM4 加密密钥（十六进制，128位），与 syslog-serve / syslog-consumer 保持一致
+# 配置后 Kafka 消息以 SM4-CBC 加密推送（格式 Base64(IV):Base64(密文)）；留空则明文推送
+syslog.sm4.generateKey=${SYSLOG_SM4_KEY:f79548ab6fa8a304fc0115e17230358a}
+
+
+#database Configuration
+spring.datasource.url=jdbc:dm://192.168.4.99:5237
+spring.datasource.username=SYSDBA
+spring.datasource.password=caZ2TcmXNSW8L2Ap
+spring.datasource.driver-class-name=dm.jdbc.driver.DmDriver
+spring.datasource.hikari.schema=\"PUBLIC\"
+
+# mybatis Configuration
+mybatis.mapper-locations=classpath:mapper/*.xml
+mybatis.type-aliases-package=com.common.entity
+mybatis.configuration.map-underscore-to-camel-case=true
+
+# Redis 缓存配置
+spring.redis.host=192.168.4.99
+spring.redis.port=6379
+# 密码（如果没有设置密码，可以省略）
+spring.redis.password=redis_GdGWte
+spring.redis.database=0
+spring.redis.timeout=2000
+spring.redis.lettuce.pool.max-active=8
+spring.redis.lettuce.pool.max-wait=-1
+spring.redis.lettuce.pool.max-idle=8
+spring.redis.lettuce.pool.min-idle=0
+spring.cache.redis.time-to-live=600000
+
+# 定时任务线程池
+spring.task.scheduling.pool.size=5
+# ============================================
+# Mock 服务器配置（模拟天眼/微步接口，供功能测试）
+# 开启后暴露 /mock/tianyan /mock/weibu 接口，采集器 host_api 指向：
+# http://127.0.0.1:8290/collectorapi/mock/tianyan
+# http://127.0.0.1:8290/collectorapi/mock/weibu
+# 生产环境务必保持关闭
+# ============================================
+mock.server.enabled=${MOCK_SERVER_ENABLED:false}
+# 模拟数据总量（每类），分页测试用
+mock.server.total-count=${MOCK_SERVER_TOTAL_COUNT:100}
diff --git a/haobang-security-dm/syslog-collector-api/src/main/resources/application.properties b/haobang-security-dm/syslog-collector-api/src/main/resources/application.properties
new file mode 100644
index 0000000..ff98f6d
--- /dev/null
+++ b/haobang-security-dm/syslog-collector-api/src/main/resources/application.properties
@@ -0,0 +1,114 @@
+#Server Configuration
+server.port=8290
+server.servlet.context-path=/collectorapi
+server.tomcat.uri-encoding=UTF-8
+server.error.include-message=always
+server.error.include-binding-errors=always
+
+# ============================================
+# 采集调度配置
+# ============================================
+# 定时采集间隔（毫秒），默认1分钟
+device.collect.interval=60000
+# 全量回补功能开关（false=禁用，POST /api/collect/full 将拒绝执行）
+device.collect.full.enabled=false
+# 重试配置
+device.collect.retry.max-attempts=3
+device.collect.retry.initial-delay=5000
+device.collect.retry.multiplier=2.0
+# Kafka 推送 topic
+device.collect.kafka.topic=${KAFKA_COLLECTOR_TOPIC:test-topic}
+# 采集线程池大小：各设备/数据类型并行采集，慢设备不阻塞其他设备
+device.collect.pool-size=${COLLECT_POOL_SIZE:15}
+# 微步 info（日志流量）单轮采集窗口上限（分钟），积压时逐轮追赶，0=不限(用默认30分钟上限)
+device.weibu.info-max-window-minutes=${WEIBU_INFO_MAX_WINDOW_MINUTES:5}
+
+#采集探针ID
+app.service.device_collect_id=${DEVICE_COLLECT_ID:1}
+# ============================================
+# 天眼采集配置
+# ============================================
+device.tianyan.host=${TIANYAN_HOST:https://10.56.170.107}
+# 免密登录 login_key（由平台管理员分配）
+device.tianyan.login-key=${TIANYAN_LOGIN_KEY:QuJm4NwF3GMaXb2Z}
+device.tianyan.username=${TIANYAN_USERNAME:xxadmin}
+device.tianyan.branch-id=${TIANYAN_BRANCH_ID:QbJK%2FsLHi}
+# 采集数据类型，逗号分隔：weblog(访问日志),alarm(告警日志)
+device.tianyan.data-types=${TIANYAN_DATA_TYPES:weblog,alarm}
+device.tianyan.page-size=${TIANYAN_PAGE_SIZE:1000}
+# 天眼单次查询子窗口跨度（分钟），采集窗口切片预防平台深分页上限，<=0 不切片
+device.tianyan.sub-window-minutes=${TIANYAN_SUB_WINDOW_MINUTES:5}
+
+# ============================================
+# 微步采集配置
+# ============================================
+device.weibu.host=${WEIBU_HOST:https://10.56.171.164}
+device.weibu.api-key=${WEIBU_API_KEY:57d8fb2b556a471ea36b8b4441fa1c0c}
+device.weibu.secret=${WEIBU_SECRET:9d967876f51d4e16bbf8bde069d12a9b}
+# 采集数据类型，逗号分隔：attack(攻击日志),info(情报日志)
+device.weibu.data-types=${WEIBU_DATA_TYPES:attack,info}
+# 微步限流：每秒最大请求数
+device.weibu.rate-limit=${WEIBU_RATE_LIMIT:10}
+# 微步单次取数上限（size 两段式取数一次取全的条数上限，超出对半拆分窗口重采）
+device.weibu.fetch-size-cap=${WEIBU_FETCH_SIZE_CAP:10000}
+
+# ============================================
+# XDR 采集配置（暂缓）
+# ============================================
+device.xdr.host=${XDR_HOST:https://127.0.0.1}
+device.xdr.auth-code=${XDR_AUTH_CODE:}
+device.xdr.api-base=${XDR_API_BASE:/api/openapi/}
+device.xdr.data-types=${XDR_DATA_TYPES:}
+
+# ============================================
+# Kafka 配置
+# ============================================
+spring.kafka.producer.bootstrap-servers=192.168.4.99:9092
+spring.kafka.producer.properties.retries=10
+spring.kafka.producer.properties.retry.backoff.ms=500
+spring.kafka.producer.properties.connections.max.idle.ms=600000
+spring.kafka.producer.properties.socket.keepalive.enable=true
+spring.kafka.producer.properties.request.timeout.ms=30000
+spring.kafka.producer.properties.delivery.timeout.ms=120000
+
+# SM4 加密密钥（十六进制，128位），与 syslog-serve / syslog-consumer 保持一致
+# 配置后 Kafka 消息以 SM4-CBC 加密推送（格式 Base64(IV):Base64(密文)）；留空则明文推送
+syslog.sm4.generateKey=${SYSLOG_SM4_KEY:f79548ab6fa8a304fc0115e17230358a}
+
+#database Configuration
+spring.datasource.url=jdbc:dm://192.168.4.99:5237
+spring.datasource.username=SYSDBA
+spring.datasource.password=caZ2TcmXNSW8L2Ap
+spring.datasource.driver-class-name=dm.jdbc.driver.DmDriver
+spring.datasource.hikari.schema=ECOSYS
+
+# mybatis Configuration
+mybatis.mapper-locations=classpath:mapper/*.xml
+mybatis.type-aliases-package=com.common.entity
+mybatis.configuration.map-underscore-to-camel-case=true
+
+# Redis 缓存配置
+spring.redis.host=localhost
+spring.redis.port=6379
+spring.redis.password=123
+spring.redis.database=0
+spring.redis.timeout=2000
+spring.redis.lettuce.pool.max-active=8
+spring.redis.lettuce.pool.max-wait=-1
+spring.redis.lettuce.pool.max-idle=8
+spring.redis.lettuce.pool.min-idle=0
+spring.cache.redis.time-to-live=600000
+
+# 定时任务线程池
+spring.task.scheduling.pool.size=5
+
+# ============================================
+# Mock 服务器配置（模拟天眼/微步接口，供功能测试）
+# 开启后暴露 /mock/tianyan /mock/weibu 接口，采集器 host_api 指向：
+# http://127.0.0.1:8290/collectorapi/mock/tianyan
+# http://127.0.0.1:8290/collectorapi/mock/weibu
+# 生产环境务必保持关闭
+# ============================================
+mock.server.enabled=${MOCK_SERVER_ENABLED:false}
+# 模拟数据总量（每类），分页测试用
+mock.server.total-count=${MOCK_SERVER_TOTAL_COUNT:100}
diff --git a/haobang-security-dm/syslog-collector-api/src/main/resources/logback.xml b/haobang-security-dm/syslog-collector-api/src/main/resources/logback.xml
new file mode 100644
index 0000000..498627a
--- /dev/null
+++ b/haobang-security-dm/syslog-collector-api/src/main/resources/logback.xml
@@ -0,0 +1,52 @@
+
+
+
+
+
+
+
+
+
+
+ ${LOG_PATTERN}
+
+
+
+
+
+ logs/syslog-collector-api.log
+
+
+ logs/syslog-collector-api.%d{yyyy-MM-dd-HH}.log
+
+ 120
+
+ 50GB
+
+ true
+
+
+ ${LOG_PATTERN}
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/haobang-security-dm/syslog-collector-api/src/main/resources/mapper/DeviceApiCollectLogMapper.xml b/haobang-security-dm/syslog-collector-api/src/main/resources/mapper/DeviceApiCollectLogMapper.xml
new file mode 100644
index 0000000..be0311d
--- /dev/null
+++ b/haobang-security-dm/syslog-collector-api/src/main/resources/mapper/DeviceApiCollectLogMapper.xml
@@ -0,0 +1,61 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ id, device_id, device_type, data_type, start_time, end_time,
+ total_count, success_count, fail_count, cost_ms, status, error_msg, create_time
+
+
+
+ INSERT INTO device_api_collect_log
+ (device_id, device_type, data_type, start_time, end_time,
+ total_count, success_count, fail_count, cost_ms, status, error_msg, create_time)
+ VALUES
+ (#{deviceId}, #{deviceType}, #{dataType}, #{startTime}, #{endTime},
+ #{totalCount}, #{successCount}, #{failCount}, #{costMs}, #{status}, #{errorMsg}, NOW())
+
+
+
+
+
+
+
+
+
+ DELETE FROM device_api_collect_log
+ WHERE start_time < #{startTime}
+
+
+
diff --git a/haobang-security-dm/syslog-collector-api/src/main/resources/mapper/DeviceApiConfigMapper.xml b/haobang-security-dm/syslog-collector-api/src/main/resources/mapper/DeviceApiConfigMapper.xml
new file mode 100644
index 0000000..17b757e
--- /dev/null
+++ b/haobang-security-dm/syslog-collector-api/src/main/resources/mapper/DeviceApiConfigMapper.xml
@@ -0,0 +1,88 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ id, device_id, device_type, device_name, device_ip, host_api, auth_config, data_types,
+ collect_interval_sec, enabled, remark, create_time, update_time
+
+
+
+
+
+
+
+
+
+
+
+ INSERT INTO device_api_config
+ (device_id, device_type, device_name, device_ip, host_api, auth_config, data_types,
+ collect_interval_sec, enabled, remark, create_time, update_time)
+ VALUES
+ (#{deviceId}, #{deviceType}, #{deviceName}, #{deviceIp}, #{hostApi}, #{authConfig}, #{dataTypes},
+ #{collectIntervalSec}, #{enabled}, #{remark}, NOW(), NOW())
+
+
+
+ UPDATE device_api_config
+ SET device_id = #{deviceId},
+ device_type = #{deviceType},
+ device_name = #{deviceName},
+ device_ip = #{deviceIp},
+ host_api = #{hostApi},
+ auth_config = #{authConfig},
+ data_types = #{dataTypes},
+ collect_interval_sec = #{collectIntervalSec},
+ enabled = #{enabled},
+ remark = #{remark},
+ update_time = NOW()
+ WHERE id = #{id}
+
+
+
+ UPDATE device_api_config
+ SET enabled = #{enabled},
+ update_time = NOW()
+ WHERE id = #{id}
+
+
+
+ DELETE FROM device_api_config WHERE id = #{id}
+
+
+
diff --git a/haobang-security-dm/syslog-consumer-rule/docker-compose/docker-compose-consumer-prod.yaml b/haobang-security-dm/syslog-consumer-rule/docker-compose/docker-compose-consumer-prod.yaml
index 151c367..19e6142 100644
--- a/haobang-security-dm/syslog-consumer-rule/docker-compose/docker-compose-consumer-prod.yaml
+++ b/haobang-security-dm/syslog-consumer-rule/docker-compose/docker-compose-consumer-prod.yaml
@@ -13,8 +13,8 @@ services:
build:
context: ./consumer-rule
dockerfile: Dockerfile
- image: syslog-consumer-rule:1.0.1
- container_name: syslog-consumer-rule
+ image: syslog-consumer-rule-dm:1.0.1
+ container_name: syslog-consumer-rule-dm
restart: unless-stopped
environment:
# 环境配置
diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/impl/OfflineAnalysisEngine.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/impl/OfflineAnalysisEngine.java
index d5b7b98..5c869a5 100644
--- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/impl/OfflineAnalysisEngine.java
+++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/impl/OfflineAnalysisEngine.java
@@ -151,7 +151,12 @@ public class OfflineAnalysisEngine implements AnalysisEngine {
if (!alarms.isEmpty()) {
//String tableName = "alarm_" + dataStartTime.format(DateTimeFormatter.ofPattern("yyyyMMdd"));
String tableName = "alarm";
- alarmMapper.batchInsert( alarms);
+ // 分批插入，每批1000条，防止达梦参数超过65535上限
+ int alarmBatchSize = 1000;
+ for (int i = 0; i < alarms.size(); i += alarmBatchSize) {
+ int end = Math.min(i + alarmBatchSize, alarms.size());
+ alarmMapper.batchInsert(alarms.subList(i, end));
+ }
alarmCount = alarms.size();
}
}
diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/impl/RealtimeAnalysisEngine.java b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/impl/RealtimeAnalysisEngine.java
index c1b6407..96b8439 100644
--- a/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/impl/RealtimeAnalysisEngine.java
+++ b/haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/impl/RealtimeAnalysisEngine.java
@@ -162,7 +162,12 @@ public class RealtimeAnalysisEngine implements AnalysisEngine {
if (!alarms.isEmpty()) {
//String tableName = "alarm_" + dataStartTime.format(DateTimeFormatter.ofPattern("yyyyMMdd"));
String tableName = "alarm";
- alarmMapper.batchInsert( alarms);
+ // 分批插入，每批1000条，防止达梦参数超过65535上限
+ int alarmBatchSize = 1000;
+ for (int i = 0; i < alarms.size(); i += alarmBatchSize) {
+ int end = Math.min(i + alarmBatchSize, alarms.size());
+ alarmMapper.batchInsert(alarms.subList(i, end));
+ }
alarmCount = alarms.size();
}
}
diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/resources/application.properties b/haobang-security-dm/syslog-consumer-rule/src/main/resources/application.properties
index a9b6b6f..acb3896 100644
--- a/haobang-security-dm/syslog-consumer-rule/src/main/resources/application.properties
+++ b/haobang-security-dm/syslog-consumer-rule/src/main/resources/application.properties
@@ -20,7 +20,7 @@ syslog.sm4.generateKey=f79548ab6fa8a304fc0115e17230358a
influxdb.url=http://192.168.4.99:8087
influxdb.token=JsUyvU8vhQEFlMM_el4Drm87fyh707IhwJNsPBucPghSdbVmdQ-UvmPcyP5NTzWxsRfEz0T51Rw4ebZUuUrmZg==
influxdb.org=influxdb
-influxdb.bucket=yelangbucket
+influxdb.bucket=influxdb
influxdb.batch.size=1000
influxdb.flush.interval=1000
influxdb.retry.attempts=3
@@ -53,7 +53,7 @@ mybatis-plus.type-handlers-package=com.Modules.etl.handler
# kafka Configuration
spring.kafka.consumer.bootstrap-servers=192.168.4.99:9092
-spring.kafka.consumer.group-id=test-group-1
+spring.kafka.consumer.group-id=test-group-dev
spring.kafka.consumer.auto-offset-reset=latest
spring.kafka.consumer.enable-auto-commit=false
spring.kafka.consumer.auto-commit-interval=1000
@@ -124,5 +124,5 @@ spring.datasource.hikari.max-lifetime=900000
spring.datasource.hikari.connection-test-query=SELECT 1 FROM DUAL
spring.datasource.hikari.validation-timeout=5000
spring.datasource.hikari.leak-detection-threshold=30000
-spring.datasource.hikari.pool-name=HikariPool-SyslogConsumer-rule
+spring.datasource.hikari.pool-name=HikariPool-SyslogConsumer-rule-dev
spring.datasource.hikari.auto-commit=false
\ No newline at end of file
diff --git a/haobang-security-dm/syslog-consumer-rule/src/main/resources/logback.xml b/haobang-security-dm/syslog-consumer-rule/src/main/resources/logback.xml
index 9196a6f..bebdaba 100644
--- a/haobang-security-dm/syslog-consumer-rule/src/main/resources/logback.xml
+++ b/haobang-security-dm/syslog-consumer-rule/src/main/resources/logback.xml
@@ -9,10 +9,10 @@
logs/syslog-consumer-rule-dm.log
- logs/syslog-consumer-rule-dm.%d{yyyy-MM-dd}.log
+ logs/syslog-consumer-rule-dm.%d{yyyy-MM-dd-HH}.log
-
- 1
+
+ 24
15GB
diff --git a/haobang-security-dm/syslog-consumer/src/main/resources/application.properties b/haobang-security-dm/syslog-consumer/src/main/resources/application.properties
index d32fc09..094d0b8 100644
--- a/haobang-security-dm/syslog-consumer/src/main/resources/application.properties
+++ b/haobang-security-dm/syslog-consumer/src/main/resources/application.properties
@@ -7,8 +7,7 @@ server.error.include-message=always
server.error.include-binding-errors=always
#run.environment: dev|test|pro
-server.run.environment=pro
-
+server.run.environment=dev
# Syslog Server Configuration
syslog.tcp.port=514
@@ -16,7 +15,6 @@ syslog.udp.port=514
syslog.max.frame.length=65536
syslog.buffer.size=1000
syslog.sm4.generateKey=f79548ab6fa8a304fc0115e17230358a
-
# InfluxDB 2.7 Configuration
influxdb.url=http://192.168.4.99:8087
influxdb.token=JsUyvU8vhQEFlMM_el4Drm87fyh707IhwJNsPBucPghSdbVmdQ-UvmPcyP5NTzWxsRfEz0T51Rw4ebZUuUrmZg==
@@ -41,10 +39,11 @@ spring.datasource.url=jdbc:dm://192.168.4.99:5237
spring.datasource.username=SYSDBA
spring.datasource.password=caZ2TcmXNSW8L2Ap
spring.datasource.driver-class-name=dm.jdbc.driver.DmDriver
-spring.datasource.hikari.schema=\"PUBLIC\"
+spring.datasource.hikari.schema=ECOSYS
# mybatis Configuration
mybatis.mapper-locations=classpath:mapper/*.xml
mybatis.type-aliases-package=com.common.entity
+
#mybatis handler 类
mybatis.configuration.default-statement-timeout=30
mybatis.configuration.default-fetch-size=1000
@@ -52,26 +51,25 @@ mybatis.configuration.map-underscore-to-camel-case=true
mybatis.type-handlers-package=com.Modules.etl.handler
mybatis-plus.configuration.map-underscore-to-camel-case=true
mybatis-plus.type-handlers-package=com.Modules.etl.handler
-
# kafka Configuration
spring.kafka.consumer.bootstrap-servers=192.168.4.99:9092
-spring.kafka.consumer.group-id=agent-syslog-group
+spring.kafka.consumer.group-id=test-group
spring.kafka.consumer.auto-offset-reset=latest
spring.kafka.consumer.enable-auto-commit=false
spring.kafka.consumer.auto-commit-interval=1000
-spring.kafka.consumer.topic=agent-syslog-topic
-
+spring.kafka.consumer.topic=test-topic
spring.kafka.consumer.max-poll-records=1000
spring.kafka.consumer.properties.max.poll.interval.ms=300000
spring.kafka.consumer.properties.session.timeout.ms=45000
+#spring.kafka.consumer.key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+#spring.kafka.consumer.value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.fetch-min-size= 1048576
spring.kafka.listener.ack-mode= manual
spring.kafka.listener.concurrency= 2
spring.kafka.listener.type=batch
-
# 定时任务配置
spring.task.scheduling.pool.size=10
@@ -84,23 +82,21 @@ partition.check.tomorrow.enabled=true
partition.check.future.days=7
partition.auto.create=true
-
-# 生产环境缓存配置
-spring.redis.host=192.168.4.99
+# 开发环境缓存配置
+spring.redis.host=localhost
spring.redis.port=6379
# 密码（如果没有设置密码，可以省略）
-spring.redis.password=redis_GdGWte
+spring.redis.password=
spring.redis.database=0
-spring.redis.timeout=5000
-#spring.redis.password=${REDIS_PASSWORD:default_prod_password}
+spring.redis.timeout=2000
-spring.redis.lettuce.pool.max-active=20
-spring.redis.lettuce.pool.max-wait=5000
-spring.redis.lettuce.pool.max-idle=10
-spring.redis.lettuce.pool.min-idle=5
+spring.redis.lettuce.pool.max-active=8
+spring.redis.lettuce.pool.max-wait=-1
+spring.redis.lettuce.pool.max-idle=8
+spring.redis.lettuce.pool.min-idle=0
+# 开发环境缓存时间较短，方便调试
+spring.cache.redis.time-to-live=600000
-# 生产环境缓存时间较长
-spring.cache.redis.time-to-live=3600000
# 应用处理器配置
app.processor.thread-pool.core-pool-size=10
@@ -141,10 +137,9 @@ spring.datasource.hikari.max-lifetime=900000
spring.datasource.hikari.connection-test-query=SELECT 1 FROM DUAL
spring.datasource.hikari.validation-timeout=5000
spring.datasource.hikari.leak-detection-threshold=30000
-spring.datasource.hikari.pool-name=HikariPool-SyslogConsumer
+spring.datasource.hikari.pool-name=HikariPool-SyslogConsumer-dev
spring.datasource.hikari.auto-commit=false
-
# 关联分析规则配置
analysis.realtime.enabled= true
# 检查间隔（秒） - 默认10秒
@@ -155,14 +150,13 @@ analysis.realtime.check-interval-seconds: 10
# false → 使用 RealtimeAnalysisEngine，平铺SQL（原有行为）
analysis.db.dameng-subquery-mode=true
-
# ============================================
# 探针联动API配置
# ============================================
# API-KEY认证（32位，建议使用随机生成的密钥）
interlocking.api-key=a1b2c3d4e5f6g7h8i9j0k1l2m3n4o5p6
# API接口基础URL（供syslog-serve调用）
-interlocking.api.base-url=http://192.168.4.99:8089/xdrservice/interlocking
+interlocking.api.base-url=http://localhost:8089/xdrservice/interlocking
# ============================================
# 告警健康检查配置
diff --git a/haobang-security-dm/syslog-consumer/src/main/resources/logback.xml b/haobang-security-dm/syslog-consumer/src/main/resources/logback.xml
index ef4c9d3..e4502a1 100644
--- a/haobang-security-dm/syslog-consumer/src/main/resources/logback.xml
+++ b/haobang-security-dm/syslog-consumer/src/main/resources/logback.xml
@@ -9,10 +9,10 @@
logs/syslog-consumer-dm.log
- logs/syslog-consumer-dm.%d{yyyy-MM-dd}.log
+ logs/syslog-consumer-dm.%d{yyyy-MM-dd-HH}.log
-
- 1
+
+ 24
15GB
diff --git a/haobang-security-dm/syslog-serve/docker-compose/docker-compose-serve-prod.yaml b/haobang-security-dm/syslog-serve/docker-compose/docker-compose-serve-prod.yaml
index 2e93ece..4efb049 100644
--- a/haobang-security-dm/syslog-serve/docker-compose/docker-compose-serve-prod.yaml
+++ b/haobang-security-dm/syslog-serve/docker-compose/docker-compose-serve-prod.yaml
@@ -28,8 +28,7 @@ services:
- SPRING_DATASOURCE_USERNAME=SYSDBA
- SPRING_DATASOURCE_PASSWORD=caZ2TcmXNSW8L2Ap
- SPRING_DATASOURCE_DRIVER_CLASS_NAME=dm.jdbc.driver.DmDriver
- - SPRING_DATASOURCE_HIKARI_SCHEMA=\"PUBLIC\"
-
+ - SPRING_DATASOURCE_HIKARI_SCHEMA=PUBLIC
# Redis配置
- SPRING_REDIS_HOST=192.168.4.99
diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/Modules/Device/DeviceProcess.java b/haobang-security-dm/syslog-serve/src/main/java/com/Modules/Device/DeviceProcess.java
index 1a8150f..0794166 100644
--- a/haobang-security-dm/syslog-serve/src/main/java/com/Modules/Device/DeviceProcess.java
+++ b/haobang-security-dm/syslog-serve/src/main/java/com/Modules/Device/DeviceProcess.java
@@ -107,7 +107,7 @@ public class DeviceProcess {
if(deviceList.size()>0)
{
deviceUnknownService.updateLastTime( deviceList.get(0).getId(),currentTime );
- System.out.println("更新未知设备的最后发现时间,IP:"+this.sourceIP);
+ logger.info("更新未知设备的最后发现时间,IP:"+this.sourceIP);
}
else {
//创建未知设备记录
diff --git a/haobang-security-dm/syslog-serve/src/main/java/com/kafka/kafkaProducer.java b/haobang-security-dm/syslog-serve/src/main/java/com/kafka/kafkaProducer.java
index ed8843a..2809a3d 100644
--- a/haobang-security-dm/syslog-serve/src/main/java/com/kafka/kafkaProducer.java
+++ b/haobang-security-dm/syslog-serve/src/main/java/com/kafka/kafkaProducer.java
@@ -5,6 +5,15 @@ import com.haobang.util.Sm4Util;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
+import java.io.File;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardOpenOption;
+import java.time.LocalDate;
+import java.time.format.DateTimeFormatter;
import java.util.LinkedHashMap;
import java.util.Properties;
@@ -20,157 +29,238 @@ public class kafkaProducer {
private static final Logger logger = LoggerFactory.getLogger(kafkaProducer.class);
+ // KafkaProducer 全局单例，线程安全，避免每次 send 重建连接
+ private static volatile Producer producer;
+ // 缓存的 topic 名称（初始化时从配置读取一次，避免每次 send 都查 Spring）
+ private static volatile String cachedTopic;
+ // 缓存的 SM4 密钥
+ private static volatile String cachedSm4Key;
+
+ private static final Object lock = new Object();
+
+ // ==================== 兜底备份配置 ====================
+ /** 兜底文件存储根目录 */
+ private static final String FALLBACK_DIR = "/app/syslog-fallback/";
+ /** 兜底文件日期格式 */
+ private static final DateTimeFormatter FALLBACK_DATE_FORMAT = DateTimeFormatter.ofPattern("yyyyMMdd");
+ /** 兜底写入最大重试次数 */
+ private static final int FALLBACK_MAX_RETRIES = 3;
+ /** 兜底写入重试间隔（毫秒） */
+ private static final long FALLBACK_RETRY_INTERVAL_MS = 100;
+
+ static {
+ // 注册 JVM 关闭钩子，确保 Producer 正常 flush + close
+ Runtime.getRuntime().addShutdownHook(new Thread(() -> closeProducer()));
+ }
+
private static AppProperties getProps() {
return SpringContextUtil.getBean(AppProperties.class);
}
+
+ /**
+ * 获取全局 KafkaProducer 单例（双重检查锁定）
+ */
+ private static Producer getProducer() {
+ if (producer == null) {
+ synchronized (lock) {
+ if (producer == null) {
+ AppProperties appProps = getProps();
+ Properties props = new Properties();
+ props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, appProps.getKafkaProducerBootstrap());
+ props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+ props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+ props.put(ProducerConfig.ACKS_CONFIG, "all");
+ props.put(ProducerConfig.RETRIES_CONFIG, 3);
+ props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
+ // 生产者缓冲区大小，减少网络 IO 次数
+ props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
+ props.put(ProducerConfig.LINGER_MS_CONFIG, 5);
+
+ producer = new KafkaProducer<>(props);
+ cachedTopic = appProps.getKafkaProducerTopic();
+ cachedSm4Key = appProps.getSm4Key();
+ logger.info("KafkaProducer 单例初始化完成, topic={}, brokers={}",
+ cachedTopic, appProps.getKafkaProducerBootstrap());
+ }
+ }
+ }
+ return producer;
+ }
+
+ /**
+ * 优雅关闭 KafkaProducer
+ */
+ public static void closeProducer() {
+ if (producer != null) {
+ synchronized (lock) {
+ if (producer != null) {
+ try {
+ logger.info("正在关闭 KafkaProducer...");
+ producer.flush();
+ producer.close();
+ logger.info("KafkaProducer 已关闭");
+ } catch (Exception e) {
+ logger.error("关闭 KafkaProducer 异常", e);
+ }
+ producer = null;
+ }
+ }
+ }
+ }
+
public static void main(String[] args) {
AppProperties appProps = getProps();
- // 配置生产者属性
Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, appProps.getKafkaProducerBootstrap());
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
-
- // 可选配置：提高可靠性
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.RETRIES_CONFIG, 3);
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
- // 创建生产者实例
- Producer producer = new KafkaProducer<>(props);
-
- try {
- // 发送10条消息
- //for (int i = 0; i < 1; i++) {
- String key = "key-" + "1";
- String value = "message-" + "1" + " at " + System.currentTimeMillis();
-
- // 创建生产者记录
- ProducerRecord record =
- new ProducerRecord<>(appProps.getKafkaProducerTopic(), key, value);
-
- // 发送消息（异步方式）
- producer.send(record, new Callback() {
- @Override
- public void onCompletion(RecordMetadata metadata, Exception exception) {
- if (exception == null) {
- System.out.println("消息发送成功: " +
- "主题=" + metadata.topic() +
- ", 分区=" + metadata.partition() +
- ", 偏移量=" + metadata.offset());
- } else {
- System.err.println("消息发送失败: " + exception.getMessage());
- }
+ try (Producer producer = new KafkaProducer<>(props)) {
+ String key = "key-" + "1";
+ String value = "message-" + "1" + " at " + System.currentTimeMillis();
+ ProducerRecord record =
+ new ProducerRecord<>(appProps.getKafkaProducerTopic(), key, value);
+ producer.send(record, new Callback() {
+ @Override
+ public void onCompletion(RecordMetadata metadata, Exception exception) {
+ if (exception == null) {
+ System.out.println("消息发送成功: " +
+ "主题=" + metadata.topic() +
+ ", 分区=" + metadata.partition() +
+ ", 偏移量=" + metadata.offset());
+ } else {
+ System.err.println("消息发送失败: " + exception.getMessage());
}
- });
- // 同步发送方式（如果需要）
- // RecordMetadata metadata = producer.send(record).get();
- // System.out.println("同步发送成功: " + metadata.offset());
- // }
+ }
+ });
+ producer.flush();
} catch (Exception e) {
e.printStackTrace();
- } finally {
- // 关闭生产者
- producer.close();
}
}
/**
- * * syslog 文本消息
- * * @param strSyslog
+ * syslog 文本消息推送到 Kafka（复用全局单例 Producer）
+ * @param deviceDevice 设备信息
+ * @param strSyslog 原始 syslog 消息
+ * @param strReceiveTime 接收时间
*/
-
- public static void messagePush(DeviceDevice deviceDevice, String strSyslog,String strReceiveTime)
- {
- AppProperties appProps = getProps();
- // 配置生产者属性
- Properties props = new Properties();
- props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, appProps.getKafkaProducerBootstrap());
- props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
- props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
-
- // 可选配置：提高可靠性
- props.put(ProducerConfig.ACKS_CONFIG, "all");
- props.put(ProducerConfig.RETRIES_CONFIG, 3);
- props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
-
- // 创建生产者实例
- Producer producer = new KafkaProducer<>(props);
+ public static void messagePush(DeviceDevice deviceDevice, String strSyslog, String strReceiveTime) {
+ Producer p = getProducer();
try {
+ String key = "key-" + System.currentTimeMillis();
+ String value = DeviceInfoUtil.getFullLogString(deviceDevice, strSyslog, strReceiveTime);
- String key = "key-" + System.currentTimeMillis();
- //String value = DeviceInfoUtil.getFullLogString(strSyslog);
- //采用动态获取Syslog 请求的设备信息
- String value = DeviceInfoUtil.getFullLogString(deviceDevice,strSyslog,strReceiveTime);
+ // SM4 加密
+ String sm4Message = Sm4Util.encryptCbc(value, cachedSm4Key);
- //Message进行SM4加密写入kafka
- String Sm4message= Sm4Util.encryptCbc(value, appProps.getSm4Key());
- //System.out.println("Sm4message:"+Sm4message);
+ ProducerRecord record =
+ new ProducerRecord<>(cachedTopic, key, sm4Message);
- // 创建生产者记录
- ProducerRecord record =
- new ProducerRecord<>(appProps.getKafkaProducerTopic(), key, Sm4message);
-
- // 发送消息（异步方式）
- producer.send(record, new Callback() {
- @Override
- public void onCompletion(RecordMetadata metadata, Exception exception) {
- if (exception == null) {
- System.out.println("消息发送成功: " +
- "主题=" + metadata.topic() +
- ", 分区=" + metadata.partition() +
- ", 偏移量=" + metadata.offset());
- } else {
- System.err.println("消息发送失败: " + exception.getMessage());
- logger.error("消息发送失败: " + exception.getMessage());
- }
+ // 异步发送 + 失败回调兜底：闭包捕获 value（原始明文），失败时写入本地文件
+ p.send(record, new Callback() {
+ @Override
+ public void onCompletion(RecordMetadata metadata, Exception exception) {
+ if (exception == null) {
+ logger.debug("消息发送成功: 主题={}, 分区={}, 偏移量={}",
+ metadata.topic(), metadata.partition(), metadata.offset());
+ } else {
+ logger.error("消息发送失败, 开始兜底写入本地文件: topic={}, 异常={}",
+ cachedTopic, exception.getMessage());
+ fallbackToLocal(value);
}
- });
-
- // 同步发送方式（如果需要）
- // RecordMetadata metadata = producer.send(record).get();
- // System.out.println("同步发送成功: " + metadata.offset());
+ }
+ });
} catch (Exception e) {
- e.printStackTrace();
- } finally {
- // 关闭生产者
- producer.close();
+ // 加密/组包阶段异常，同样兜底
+ logger.error("messagePush 预处理异常, 开始兜底写入本地文件", e);
+ try {
+ String value = DeviceInfoUtil.getFullLogString(deviceDevice, strSyslog, strReceiveTime);
+ fallbackToLocal(value);
+ } catch (Exception ex) {
+ logger.error("兜底写入时组包也失败, 消息已丢失", ex);
+ }
}
+ // 注意：不再 close，Producer 是全局单例，由 shutdown hook 关闭
}
/**
* 生成json字符串，补充设备配置信息
- * @param strSyslog
- * @return
*/
- private static String getLogJsonString(String strSyslog)
- {
+ private static String getLogJsonString(String strSyslog) {
ObjectMapper objectMapper = new ObjectMapper();
- // 创建设备配置信息MAP
Map map = new LinkedHashMap<>();
AppProperties appProps = getProps();
map.put("device_id", appProps.getAppServiceDeviceId());
map.put("device_name", appProps.getAppServiceDeviceName());
map.put("vendor", appProps.getAppServiceVendor());
map.put("data_type", appProps.getAppServiceDataType());
- // 添加syslogMessage
map.put("syslogMessage", strSyslog);
try {
- //生成json消息字符串
- String json = objectMapper.writeValueAsString(map);
- return json;
- }
- catch(Exception ex)
- {
- logger.error("kafkaProducer getLogJsonString(）生成json字符串 异常: syslog: "+ strSyslog);
- logger.error("Exception: : "+ ex.getMessage());
+ return objectMapper.writeValueAsString(map);
+ } catch (Exception ex) {
+ logger.error("kafkaProducer getLogJsonString() 生成json字符串异常, syslog: {}", strSyslog);
+ logger.error("Exception: {}", ex.getMessage());
}
return null;
}
+ // ==================== 兜底备份 ====================
+ /**
+ * 将原始消息写入本地兜底文件（按日期分文件，追加写入）
+ *
+ * 文件路径：/data/syslog-fallback/kafka-failed-yyyyMMdd.log
+ *
+ * @param message 原始明文消息（完整日志字符串）
+ */
+ private static void fallbackToLocal(String message) {
+ if (message == null || message.isEmpty()) {
+ logger.warn("兜底消息为空，跳过写入");
+ return;
+ }
+ for (int i = 0; i < FALLBACK_MAX_RETRIES; i++) {
+ try {
+ String dateStr = LocalDate.now().format(FALLBACK_DATE_FORMAT);
+ String fileName = "kafka-failed-" + dateStr + ".log";
+ Path filePath = Paths.get(FALLBACK_DIR, fileName);
+ // 确保目录存在
+ File dir = new File(FALLBACK_DIR);
+ if (!dir.exists()) {
+ boolean created = dir.mkdirs();
+ if (!created) {
+ logger.error("创建兜底目录失败: {}", FALLBACK_DIR);
+ return;
+ }
+ }
+
+ // 追加写入，每条一行
+ String line = message + "\n";
+ Files.write(filePath, line.getBytes(StandardCharsets.UTF_8),
+ StandardOpenOption.CREATE, StandardOpenOption.APPEND);
+
+ logger.debug("消息已兜底写入文件: {}", fileName);
+ return;
+
+ } catch (IOException e) {
+ logger.error("兜底文件写入失败(第{}次重试): {}", i + 1, e.getMessage());
+ if (i < FALLBACK_MAX_RETRIES - 1) {
+ try {
+ Thread.sleep(FALLBACK_RETRY_INTERVAL_MS);
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ break;
+ }
+ }
+ }
+ }
+ logger.error("兜底写入彻底失败，消息丢失！消息长度={}", message.length());
+ }
}
diff --git a/haobang-security-dm/syslog-serve/src/main/resources/application-prod.properties b/haobang-security-dm/syslog-serve/src/main/resources/application-prod.properties
index a3146be..d66847e 100644
--- a/haobang-security-dm/syslog-serve/src/main/resources/application-prod.properties
+++ b/haobang-security-dm/syslog-serve/src/main/resources/application-prod.properties
@@ -9,8 +9,8 @@ server.error.include-binding-errors=always
# Syslog Server Configuration
syslog.tcp.port=514
syslog.udp.port=514
-syslog.max.frame.length=2621440
-syslog.buffer.size=10000
+syslog.max.frame.length=262144
+syslog.buffer.size=1000
# syslog 消息字符编码，国内安全设备普遍使用 GBK，如需 UTF-8 改为 syslog.charset=UTF-8
syslog.charset=UTF-8
syslog.sm4.generateKey=f79548ab6fa8a304fc0115e17230358a
@@ -23,7 +23,7 @@ app.service.product_name=diting
app.service.data_type=json
#采集探针ID
app.service.device_collect_id=${DEVICE_COLLECT_ID:1}
-app.service.version=${APP_SERVICE_VER:V1.0.0-20260527}
+app.service.version=${APP_SERVICE_VER:V1.0.0-20260811}
app.service.device_collect_name=${DEVICE_COLLECT_NAME:DevCollect-01}
# kafka Configuration
diff --git a/haobang-security-dm/syslog-serve/src/main/resources/logback.xml b/haobang-security-dm/syslog-serve/src/main/resources/logback.xml
index f90779a..d03f664 100644
--- a/haobang-security-dm/syslog-serve/src/main/resources/logback.xml
+++ b/haobang-security-dm/syslog-serve/src/main/resources/logback.xml
@@ -16,10 +16,10 @@
logs/syslog-serve-dm.log
-
- logs/syslog-serve-dm.%d{yyyy-MM-dd}.log
-
- 1
+
+ logs/syslog-serve-dm.%d{yyyy-MM-dd-HH}.log
+
+ 24
15GB