diff --git a/src/main/java/com/yunhai/platform/gateway/log/AccessLogService.java b/src/main/java/com/yunhai/platform/gateway/log/AccessLogService.java new file mode 100644 index 0000000..d8233de --- /dev/null +++ b/src/main/java/com/yunhai/platform/gateway/log/AccessLogService.java @@ -0,0 +1,267 @@ +package com.yunhai.platform.gateway.log; + +import com.yunhai.platform.gateway.entity.OpenApiLog; +import com.yunhai.platform.gateway.entity.OpenApiStatDaily; +import com.yunhai.platform.gateway.filter.SanitizeFilter; +import com.yunhai.platform.gateway.filter.SignVerifyFilter; +import com.yunhai.platform.gateway.repository.OpenApiLogRepository; +import com.yunhai.platform.gateway.repository.OpenApiStatDailyRepository; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; + +import javax.annotation.PostConstruct; +import javax.annotation.PreDestroy; +import javax.servlet.http.HttpServletRequest; +import java.time.LocalDateTime; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.Executor; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; + +/** + * 台账异步落库（T1.5，有界队列 + 背压）：全量请求观测（GET/401 也记）， + * 台账故障绝不拖垮请求线程——台账是观测设施，不能反杀业务。 + * + * 日统计维度 caller：台账 caller 为 NULL（401 无 appKey）时归入 {@value #STAT_CALLER_UNKNOWN} + * （stat 表该列 NOT NULL，观测聚合需要一个兜底桶）。 + */ +@Component +public class AccessLogService { + + public static final int QUEUE_CAPACITY = 10_000; + public static final int BATCH_SIZE = 200; + public static final long FLUSH_INTERVAL_MS = 500L; + public static final String STAT_CALLER_UNKNOWN = "unknown"; + /** ANON 判定下台账 caller 口径 */ + public static final String CALLER_ANONYMOUS = "anonymous"; + + private static final Logger log = LoggerFactory.getLogger(AccessLogService.class); + + private final ArrayBlockingQueue queue; + private final AtomicLong dropped = new AtomicLong(); + private final AtomicBoolean flushPending = new AtomicBoolean(); + + @Autowired + private OpenApiLogRepository logRepository; + + @Autowired + private OpenApiStatDailyRepository statRepository; + + /** 刷盘任务派发器：生产 = 单刷盘线程；测试注入同线程执行器作同步触发点 */ + private Executor flushDispatcher; + private ScheduledExecutorService scheduler; + + public AccessLogService() { + this(QUEUE_CAPACITY); + } + + /** 测试构造：缩小容量以复现背压（生产路径固定 QUEUE_CAPACITY） */ + AccessLogService(int capacity) { + this.queue = new ArrayBlockingQueue<>(capacity); + } + + void setFlushDispatcher(Executor flushDispatcher) { + this.flushDispatcher = flushDispatcher; + } + + int queueSize() { + return queue.size(); + } + + long droppedCount() { + return dropped.get(); + } + + @PostConstruct + void start() { + scheduler = Executors.newSingleThreadScheduledExecutor(r -> { + Thread t = new Thread(r, "gw-access-log-flush"); + t.setDaemon(true); + return t; + }); + if (flushDispatcher == null) { + flushDispatcher = scheduler; + } + scheduler.scheduleWithFixedDelay(() -> { + try { + flushOnce(); + } catch (RuntimeException e) { + log.error("[AccessLog] scheduled flush failed", e); + } + }, FLUSH_INTERVAL_MS, FLUSH_INTERVAL_MS, TimeUnit.MILLISECONDS); + } + + @PreDestroy + void stop() { + if (scheduler != null) { + scheduler.shutdown(); + try { + scheduler.awaitTermination(5, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + // 关停钩子：flush 余量（尽力而为） + try { + while (flushOnce() > 0) { + // 排空 + } + } catch (RuntimeException e) { + log.error("[AccessLog] shutdown flush failed", e); + } + } + + /** + * 请求线程入口：只入队，绝不同步写库、绝不阻塞。 + */ + public void record(OpenApiLog entry) { + if (entry.getTs() == null) { + entry.setTs(LocalDateTime.now()); + } + if (!queue.offer(entry)) { + long total = dropped.incrementAndGet(); + log.error("[AccessLog] queue FULL, entry dropped. accumulatedDropped={}", total); + return; + } + if (queue.size() >= BATCH_SIZE) { + scheduleFlush(); + } + } + + /** + * 按卡口径从请求组装并记录一条台账（T1.6 链上过滤器消费）： + * path = 剥段后路径；caller = ANON 记 anonymous、PASS 记验签 appKey、 + * 否则尽力取 {@code X-App-Key} 头（401 无 appKey 记 NULL）；status = 网关出口状态。 + * + * @param backendCode 提供方业务 code（响应为 JSON 信封时尽力解析，否则 null） + */ + public void recordRequest(HttpServletRequest req, int status, Integer backendCode, int costMs) { + OpenApiLog entry = new OpenApiLog(); + entry.setTs(LocalDateTime.now()); + entry.setTargetCode(truncate(targetOf(req), 32)); + entry.setCallerAppKey(truncate(callerOf(req), 64)); + entry.setMethod(req.getMethod()); + Object stripped = req.getAttribute(SanitizeFilter.ATTR_STRIPPED_PATH); + entry.setPath(truncate(stripped instanceof String ? (String) stripped : req.getRequestURI(), 256)); + entry.setStatus(status); + entry.setBackendCode(backendCode); + entry.setCostMs(costMs); + record(entry); + } + + private void scheduleFlush() { + Executor dispatcher = flushDispatcher; + if (dispatcher == null) { + return; // 未初始化（单元/异常场景）：500ms 周期兜底，不阻塞请求线程 + } + if (flushPending.compareAndSet(false, true)) { + dispatcher.execute(() -> { + flushPending.set(false); + try { + flushOnce(); + } catch (RuntimeException e) { + log.error("[AccessLog] batch flush failed", e); + } + }); + } + } + + /** 同步触发点（满批/周期/关停共用）：排空至多一批 → 批量插入 → 按日 upsert 统计 */ + synchronized int flushOnce() { + List batch = new ArrayList<>(BATCH_SIZE); + queue.drainTo(batch, BATCH_SIZE); + if (batch.isEmpty()) { + return 0; + } + logRepository.saveAll(batch); + upsertDaily(batch); + return batch.size(); + } + + private void upsertDaily(List batch) { + Map grouped = new LinkedHashMap<>(); + for (OpenApiLog e : batch) { + String caller = e.getCallerAppKey() == null ? STAT_CALLER_UNKNOWN : e.getCallerAppKey(); + String key = e.getTs().toLocalDate() + "|" + e.getTargetCode() + "|" + caller; + OpenApiStatDaily s = grouped.get(key); + if (s == null) { + s = new OpenApiStatDaily(); + s.setStatDate(e.getTs().toLocalDate()); + s.setTargetCode(e.getTargetCode()); + s.setCallerAppKey(caller); + grouped.put(key, s); + } + s.setTotal(s.getTotal() + 1); + if (isErr(e)) { + s.setErrTotal(s.getErrTotal() + 1); + } + s.setAvgCostMs((int) ((long) s.getAvgCostMs() * (s.getTotal() - 1) + e.getCostMs()) / s.getTotal()); + } + for (OpenApiStatDaily fresh : grouped.values()) { + OpenApiStatDaily existing = statRepository.findFirstByStatDateAndTargetCodeAndCallerAppKey( + fresh.getStatDate(), fresh.getTargetCode(), fresh.getCallerAppKey()); + if (existing != null) { + int total = existing.getTotal() + fresh.getTotal(); + existing.setErrTotal(existing.getErrTotal() + fresh.getErrTotal()); + existing.setAvgCostMs((int) (((long) existing.getAvgCostMs() * existing.getTotal() + + (long) fresh.getAvgCostMs() * fresh.getTotal()) / total)); + existing.setTotal(total); + statRepository.save(existing); + } else { + statRepository.save(fresh); + } + } + } + + private static boolean isErr(OpenApiLog e) { + return e.getStatus() >= 400 || (e.getBackendCode() != null && e.getBackendCode() != 0); + } + + /** target = 规整化 URI 首段；取不到记 unknown（列 NOT NULL，台账一条都不能丢在写入端） */ + private static String targetOf(HttpServletRequest req) { + Object uri = req.getAttribute(SanitizeFilter.ATTR_SANITIZED_URI); + String path = uri instanceof String ? (String) uri : req.getRequestURI(); + if (path == null || path.length() < 2) { + return STAT_CALLER_UNKNOWN; + } + int second = path.indexOf('/', 1); + return second < 0 ? path.substring(1) : path.substring(1, second); + } + + private static String callerOf(HttpServletRequest req) { + Object verdict = req.getAttribute(SignVerifyFilter.ATTR_VERDICT); + if (SignVerifyFilter.VERDICT_ANON.equals(verdict)) { + return CALLER_ANONYMOUS; + } + if (SignVerifyFilter.VERDICT_PASS.equals(verdict)) { + Object appKey = req.getAttribute(SignVerifyFilter.ATTR_AUTH_APPKEY); + if (appKey instanceof String && !((String) appKey).trim().isEmpty()) { + return (String) appKey; + } + } + String header = req.getHeader("X-App-Key"); + return header == null || header.trim().isEmpty() ? null : header.trim(); // 401 无 appKey = NULL + } + + private static String truncate(String s, int max) { + if (s == null) { + return null; + } + return s.length() <= max ? s : s.substring(0, max); + } +} diff --git a/src/main/java/com/yunhai/platform/gateway/repository/OpenApiStatDailyRepository.java b/src/main/java/com/yunhai/platform/gateway/repository/OpenApiStatDailyRepository.java index 4af15a7..1ee92d0 100644 --- a/src/main/java/com/yunhai/platform/gateway/repository/OpenApiStatDailyRepository.java +++ b/src/main/java/com/yunhai/platform/gateway/repository/OpenApiStatDailyRepository.java @@ -3,5 +3,11 @@ package com.yunhai.platform.gateway.repository; import com.yunhai.platform.gateway.entity.OpenApiStatDaily; import org.springframework.data.jpa.repository.JpaRepository; +import java.time.LocalDate; + public interface OpenApiStatDailyRepository extends JpaRepository { + + /** uk_stat 维度定位（T1.5 按日 upsert）：stat_date + target_code + caller_app_key */ + OpenApiStatDaily findFirstByStatDateAndTargetCodeAndCallerAppKey( + LocalDate statDate, String targetCode, String callerAppKey); } diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index ffb912a..94d30d4 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -15,6 +15,12 @@ spring: hibernate: # 表结构由 docs/sql/01_schema.sql 交付，应用不自动建表 ddl-auto: none + properties: + hibernate: + # 台账批量刷盘（T1.5）：一批 JPA batch insert + jdbc: + batch_size: 200 + order_inserts: true redis: # 密钥链（sign:app:{appKey}）走共享 Redis db0——由平台配置层注入 host: # diff --git a/src/test/java/com/yunhai/platform/gateway/log/AccessLogServiceTest.java b/src/test/java/com/yunhai/platform/gateway/log/AccessLogServiceTest.java new file mode 100644 index 0000000..ee2b6d3 --- /dev/null +++ b/src/test/java/com/yunhai/platform/gateway/log/AccessLogServiceTest.java @@ -0,0 +1,107 @@ +package com.yunhai.platform.gateway.log; + +import com.yunhai.platform.gateway.entity.OpenApiLog; +import com.yunhai.platform.gateway.entity.OpenApiStatDaily; +import com.yunhai.platform.gateway.repository.OpenApiLogRepository; +import com.yunhai.platform.gateway.repository.OpenApiStatDailyRepository; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.springframework.test.util.ReflectionTestUtils; + +import java.time.LocalDateTime; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; + +/** + * AccessLogService 三用例（T1.5 TDD）：offer 入队成功路径（请求线程零写库）； + * 满 200 条触发批量 flush（同步触发点测，不起真线程等待——注入同线程 dispatcher， + * 一批 JPA saveAll + 按日 upsert stat_daily）；队列满 offer 失败 → 丢弃计数 +1 且绝不抛异常。 + */ +class AccessLogServiceTest { + + private AccessLogService service(int capacity, OpenApiLogRepository logRepo, + OpenApiStatDailyRepository statRepo) { + AccessLogService svc = new AccessLogService(capacity); + ReflectionTestUtils.setField(svc, "logRepository", logRepo); + ReflectionTestUtils.setField(svc, "statRepository", statRepo); + svc.setFlushDispatcher(Runnable::run); // 同步触发点：不真起线程 + return svc; + } + + private OpenApiLog entry(String target, String caller, int status) { + OpenApiLog e = new OpenApiLog(); + e.setTs(LocalDateTime.now()); + e.setTargetCode(target); + e.setCallerAppKey(caller); + e.setMethod("GET"); + e.setPath("/oapi/x"); + e.setStatus(status); + e.setCostMs(10); + return e; + } + + @Test + void offer入队成功且请求线程零写库() { + OpenApiLogRepository logRepo = mock(OpenApiLogRepository.class); + OpenApiStatDailyRepository statRepo = mock(OpenApiStatDailyRepository.class); + AccessLogService svc = service(AccessLogService.QUEUE_CAPACITY, logRepo, statRepo); + + svc.record(entry("usermsg", "100617", 200)); + + assertThat(svc.queueSize()).as("入队 1 条").isEqualTo(1); + assertThat(svc.droppedCount()).isZero(); + verifyNoInteractions(logRepo, statRepo); // 红线：不得在请求线程直接写库 + } + + @Test + void 满200条触发批量flush并upsert日统计() { + OpenApiLogRepository logRepo = mock(OpenApiLogRepository.class); + OpenApiStatDailyRepository statRepo = mock(OpenApiStatDailyRepository.class); + AccessLogService svc = service(AccessLogService.QUEUE_CAPACITY, logRepo, statRepo); + + for (int i = 0; i < AccessLogService.BATCH_SIZE; i++) { + svc.record(entry("usermsg", "100617", i == 199 ? 500 : 200)); // 末条 HTTP≥400 计 err + } + + ArgumentCaptor> batchCap = ArgumentCaptor.forClass(List.class); + verify(logRepo).saveAll(batchCap.capture()); + assertThat(batchCap.getValue()).as("一批 200 条批量插入").hasSize(200); + assertThat(svc.queueSize()).as("flush 后队列清空").isZero(); + + ArgumentCaptor statCap = ArgumentCaptor.forClass(OpenApiStatDaily.class); + verify(statRepo).save(statCap.capture()); + OpenApiStatDaily stat = statCap.getValue(); + assertThat(stat.getTargetCode()).isEqualTo("usermsg"); + assertThat(stat.getCallerAppKey()).isEqualTo("100617"); + assertThat(stat.getTotal()).isEqualTo(200); + assertThat(stat.getErrTotal()).as("status≥400 计入 err").isEqualTo(1); + assertThat(stat.getAvgCostMs()).isEqualTo(10); + } + + @Test + void 队列满丢弃计数加一且不抛异常() { + OpenApiLogRepository logRepo = mock(OpenApiLogRepository.class); + OpenApiStatDailyRepository statRepo = mock(OpenApiStatDailyRepository.class); + AccessLogService svc = service(2, logRepo, statRepo); // 测试容量：小队列复现背压 + + svc.record(entry("usermsg", "a", 200)); + svc.record(entry("usermsg", "a", 200)); + assertThat(svc.queueSize()).isEqualTo(2); + + assertThatCode(() -> { + svc.record(entry("usermsg", "a", 200)); // 第 3 条 = 队列满 + svc.record(entry("usermsg", "a", 200)); + }).as("绝不阻塞/抛出，反杀业务不可接受").doesNotThrowAnyException(); + + assertThat(svc.droppedCount()).as("累计丢弃数").isEqualTo(2); + assertThat(svc.queueSize()).isEqualTo(2); + verify(logRepo, never()).saveAll(any()); + } +}