feat(syslog-serve): 优化微步5秒窗口与5万条日志采集
- 将 API 拉取任务的 lookbackSeconds 通用默认值调整为 0 - 微步任务采用5秒时间窗口、50000条单窗口上限和64MiB响应保护 - 将微步适配器及管理接口的 pageSize 上限放宽至50000 - 将 API 调度扫描默认周期调整为1000毫秒 - 兼容新旧调度扫描环境变量名称 - 更新达梦数据库 PAGE_SIZE 约束及 LOOKBACK_SECONDS 默认值 - 补充 v6 数据库预检、升级和验证脚本 - 更新微步任务模板、现场操作说明和启动版本标识 - 增加配置边界及启动参数自动化测试 验证： - JDK 8 Maven测试通过：167个，失败0，错误0，跳过2 - git diff --check通过
This commit is contained in:
王绍宏
+2
-1
@@ -12,7 +12,7 @@ import org.springframework.stereotype.Component;
|
||||
public class SyslogBuildStartupLogger {
|
||||
|
||||
// 确认 JAR 包含设备级采集模式、API 拉取、探针归属过滤、微步全量查询及诊断接口。
|
||||
static final String CODE_REVISION = "WEIBU_DIAGNOSTIC_V20260917_66";
|
||||
static final String CODE_REVISION = "WEIBU_5S_50000_SCAN1S_V20260917_70";
|
||||
private final Environment environment;
|
||||
|
||||
public SyslogBuildStartupLogger(Environment environment) {
|
||||
@@ -32,6 +32,7 @@ public class SyslogBuildStartupLogger {
|
||||
+ ", contextPath=" + value("server.servlet.context-path", "/syslogserve")
|
||||
+ ", syslogEnabled=" + value("syslog.server.enabled", "true")
|
||||
+ ", apiPullEnabled=" + value("device.api.enabled", "true")
|
||||
+ ", apiSchedulerScanIntervalMs=" + value("device.api.scheduler.scan-interval-ms", "1000")
|
||||
+ ", apiTlsVerifyEnabled=" + value("device.api.tls-verify-enabled", "false")
|
||||
+ ", apiHostAllowlistEnabled=" + value("device.api.host-allowlist-enabled", "false")
|
||||
+ ", walEnabled=" + value("syslog.wal.enabled", "true")
|
||||
|
||||
+1
-1
@@ -19,7 +19,7 @@ public class ApiPullProperties {
|
||||
@Value("${device.api.scheduler.instance-id:${HOSTNAME:local}-${random.uuid}}")
|
||||
private String instanceId;
|
||||
|
||||
@Value("${device.api.scheduler.scan-interval-ms:5000}")
|
||||
@Value("${device.api.scheduler.scan-interval-ms:1000}")
|
||||
private long scanIntervalMs;
|
||||
|
||||
@Value("${device.api.scheduler.lease-seconds:120}")
|
||||
|
||||
+1
-1
@@ -79,7 +79,7 @@ public class ApiPullScheduler {
|
||||
}
|
||||
|
||||
/** 扫描到期任务；总开关关闭时不查询数据库，也不创建任何设备请求。 */
|
||||
@Scheduled(fixedDelayString = "${device.api.scheduler.scan-interval-ms:5000}")
|
||||
@Scheduled(fixedDelayString = "${device.api.scheduler.scan-interval-ms:1000}")
|
||||
public void scan() {
|
||||
if (!properties.isEnabled()) return;
|
||||
for (DeviceCollectApiConfig c
|
||||
|
||||
+6
-4
@@ -65,9 +65,9 @@ import org.springframework.stereotype.Component;
|
||||
* Runner 先读取响应 {@code total}，当窗口内总数超过 {@code pageSize} 时只丢弃探测响应并继续缩小
|
||||
* 时间窗口，直到单次响应能够完整返回。最小一秒窗口仍超限时失败且不推进 checkpoint。
|
||||
*
|
||||
* <p>externalEventId 使用厂商日志 {@code id}，事件时间使用 {@code time}。事件 ID 只由任务命名空间
|
||||
* 和厂商稳定 ID 生成，lookback 重叠查询不会制造新事件版本。适配器不把 API JSON 伪装成 Syslog，
|
||||
* 业务字段标准化由下游针对 API 的泛化规则完成。
|
||||
* <p>externalEventId 使用厂商日志 {@code id}，事件时间使用 {@code time}。通用配置默认
|
||||
* {@code lookbackSeconds=0}；确有迟到补偿需求的任务可以显式配置正数。适配器不把 API JSON
|
||||
* 伪装成 Syslog，业务字段标准化由下游针对 API 的泛化规则完成。
|
||||
*/
|
||||
@Component
|
||||
public class WeibuApiAdapter implements ApiPullAdapter {
|
||||
@@ -163,7 +163,9 @@ public class WeibuApiAdapter implements ApiPullAdapter {
|
||||
public void validateConfig(DeviceCollectApiConfig c) {
|
||||
if (c == null || !TYPE.equals(c.getAdapterType())) fail("CONFIG_SCHEMA_ERROR");
|
||||
if (!LOG.equals(c.getStreamKey())) fail("STREAM_NOT_SUPPORTED");
|
||||
range(c.getPageSize(), 1, 1000);
|
||||
// 现场微步 searchBySql 的 size 参数支持到 50000。接口没有可靠分页游标，
|
||||
// 因此 pageSize 表示单个时间窗口允许完整取回的硬上限。
|
||||
range(c.getPageSize(), 1, 50000);
|
||||
if (c.getMaxPagesPerRun() == null || c.getMaxPagesPerRun() != 1)
|
||||
fail("CONFIG_SCHEMA_ERROR");
|
||||
range(c.getPollIntervalSeconds(), 1, 86400);
|
||||
|
||||
+3
-3
@@ -51,7 +51,7 @@ public final class ApiPullDtos {
|
||||
private Integer pollIntervalSeconds = 10;
|
||||
|
||||
@Min(1)
|
||||
@Max(10000)
|
||||
@Max(50000)
|
||||
private Integer pageSize = 100;
|
||||
|
||||
@Min(1)
|
||||
@@ -80,7 +80,7 @@ public final class ApiPullDtos {
|
||||
|
||||
@Min(0)
|
||||
@Max(1209600)
|
||||
private Integer lookbackSeconds = 300;
|
||||
private Integer lookbackSeconds = 0;
|
||||
|
||||
@Min(1024)
|
||||
@Max(268435456)
|
||||
@@ -113,7 +113,7 @@ public final class ApiPullDtos {
|
||||
private Integer pollIntervalSeconds;
|
||||
|
||||
@Min(1)
|
||||
@Max(10000)
|
||||
@Max(50000)
|
||||
private Integer pageSize;
|
||||
|
||||
@Min(1)
|
||||
|
||||
@@ -131,7 +131,8 @@ mybatis.configuration.map-underscore-to-camel-case=true
|
||||
device.api.enabled=${DEVICE_API_ENABLED:true}
|
||||
# 设备 AppKey/AppSecret 由 device_collect_api_credential 表管理，配置文件不再保存厂商凭据。
|
||||
device.api.scheduler.instance-id=${DEVICE_API_INSTANCE_ID:${HOSTNAME:local}}
|
||||
device.api.scheduler.scan-interval-ms=${DEVICE_API_SCAN_INTERVAL_MS:1000}
|
||||
# 新名称与 Spring 属性一致；继续兼容旧环境变量 DEVICE_API_SCAN_INTERVAL_MS。
|
||||
device.api.scheduler.scan-interval-ms=${DEVICE_API_SCHEDULER_SCAN_INTERVAL_MS:${DEVICE_API_SCAN_INTERVAL_MS:1000}}
|
||||
device.api.scheduler.lease-seconds=${DEVICE_API_LEASE_SECONDS:120}
|
||||
# 同一任务页内严格串行；不同设备可并发。同一设备主机默认仅一个在途请求。
|
||||
device.api.scheduler.global-concurrency=${DEVICE_API_GLOBAL_CONCURRENCY:4}
|
||||
|
||||
+3
@@ -16,6 +16,7 @@ class SyslogBuildStartupLoggerTest {
|
||||
void logsCodeRevisionAndEffectiveOverrides() {
|
||||
MockEnvironment environment = new MockEnvironment()
|
||||
.withProperty("device.api.enabled", "false")
|
||||
.withProperty("device.api.scheduler.scan-interval-ms", "1000")
|
||||
.withProperty("device.api.tls-verify-enabled", "true")
|
||||
.withProperty("app.service.device_collect_id", "40")
|
||||
.withProperty("spring.kafka.producer.topic", "prod-syslog");
|
||||
@@ -34,6 +35,7 @@ class SyslogBuildStartupLoggerTest {
|
||||
assertTrue(message.contains("application=syslog-serve"));
|
||||
assertTrue(message.contains("profiles=prod"));
|
||||
assertTrue(message.contains("apiPullEnabled=false"));
|
||||
assertTrue(message.contains("apiSchedulerScanIntervalMs=1000"));
|
||||
assertTrue(message.contains("apiTlsVerifyEnabled=true"));
|
||||
assertTrue(message.contains("deviceCollectId=40"));
|
||||
assertTrue(message.contains("kafkaTopic=prod-syslog"));
|
||||
@@ -51,6 +53,7 @@ class SyslogBuildStartupLoggerTest {
|
||||
assertFalse(summary.contains("codeRevision=fake"));
|
||||
assertTrue(summary.contains("syslogEnabled=true"));
|
||||
assertTrue(summary.contains("apiPullEnabled=true"));
|
||||
assertTrue(summary.contains("apiSchedulerScanIntervalMs=1000"));
|
||||
assertTrue(summary.contains("apiTlsVerifyEnabled=false"));
|
||||
assertTrue(summary.contains("walEnabled=true"));
|
||||
}
|
||||
|
||||
+13
@@ -8,6 +8,7 @@ import com.haobang.apipull.api.ApiPullDtos.ApiPullTaskResponse;
|
||||
import com.haobang.apipull.api.ApiPullDtos.CreateApiPullTaskRequest;
|
||||
import java.io.InputStream;
|
||||
import java.util.Collections;
|
||||
import javax.validation.constraints.Max;
|
||||
import org.apache.ibatis.builder.xml.XMLMapperBuilder;
|
||||
import org.apache.ibatis.session.Configuration;
|
||||
import org.junit.jupiter.api.Test;
|
||||
@@ -37,6 +38,18 @@ class ApiPullContractTest {
|
||||
.readValue(body, CreateApiPullTaskRequest.class));
|
||||
}
|
||||
|
||||
@Test
|
||||
void newApiPullTaskDefaultsLookbackToZero() {
|
||||
assertEquals(0, new CreateApiPullTaskRequest().getLookbackSeconds());
|
||||
}
|
||||
|
||||
@Test
|
||||
void managementDtoAllowsWeibuFiftyThousandRecordWindow() throws Exception {
|
||||
Max max = CreateApiPullTaskRequest.class.getDeclaredField("pageSize").getAnnotation(Max.class);
|
||||
assertNotNull(max);
|
||||
assertEquals(50000L, max.value());
|
||||
}
|
||||
|
||||
@Test
|
||||
void mybatisMapperXmlIsSyntacticallyValidAndBindsEveryStatement() throws Exception {
|
||||
Configuration configuration = new Configuration();
|
||||
|
||||
+11
-1
@@ -251,6 +251,16 @@ class WeibuApiAdapterTest {
|
||||
assertThrows(AdapterException.class, () -> unpaged.pull(0L)).getCode());
|
||||
verifyNoInteractions(unpaged.http);
|
||||
|
||||
Fixture capacity = new Fixture();
|
||||
capacity.config.setPageSize(50000);
|
||||
capacity.adapter.validateConfig(capacity.config);
|
||||
capacity.config.setPageSize(50001);
|
||||
assertEquals(
|
||||
"CONFIG_SCHEMA_ERROR",
|
||||
assertThrows(AdapterException.class, () -> capacity.adapter.validateConfig(capacity.config))
|
||||
.getCode());
|
||||
verifyNoInteractions(capacity.http);
|
||||
|
||||
Fixture incident = new Fixture();
|
||||
incident.config.setStreamKey("WEIBU_INCIDENT");
|
||||
assertEquals(
|
||||
@@ -297,7 +307,7 @@ class WeibuApiAdapterTest {
|
||||
c.setRequestTimeoutMs(3000);
|
||||
c.setWindowSizeSeconds(3600);
|
||||
c.setSafetyDelaySeconds(120);
|
||||
c.setLookbackSeconds(300);
|
||||
c.setLookbackSeconds(0);
|
||||
c.setMaxResponseBytes(1024 * 1024L);
|
||||
c.setExtraConfigJson("{}");
|
||||
return c;
|
||||
|
||||
Reference in New Issue
Block a user