diff --git a/xuqm-bugcollect-service/docs/BugCollect-API-v1.md b/xuqm-bugcollect-service/docs/BugCollect-API-v1.md index a735b8d..3473e0c 100644 --- a/xuqm-bugcollect-service/docs/BugCollect-API-v1.md +++ b/xuqm-bugcollect-service/docs/BugCollect-API-v1.md @@ -143,6 +143,26 @@ POST /bugcollect/v1/issues/batch Content-Type: application/json ``` +### 摄取过载 + +`issues/batch` 与 `events/batch` 单批最多 30 条。平台触发容量保护时返回： + +```http +HTTP/1.1 429 Too Many Requests +Retry-After: 27 +Content-Type: application/json +``` + +```json +{ + "code": 429, + "message": "BUGCOLLECT_RATE_LIMITED", + "data": null +} +``` + +SDK 应保留本地队列，在 `Retry-After` 指定时间后重试；不得把该响应解释为永久停用。 + #### 请求体 ```json diff --git a/xuqm-bugcollect-service/docs/BugCollect-Backpressure.md b/xuqm-bugcollect-service/docs/BugCollect-Backpressure.md new file mode 100644 index 0000000..e7a2e3d --- /dev/null +++ b/xuqm-bugcollect-service/docs/BugCollect-Backpressure.md @@ -0,0 +1,35 @@ +# BugCollect 摄取防雪崩设计 + +## 目标 + +线上同一故障可能在短时间内由大量设备同时触发。摄取链路必须保护数据库和服务线程， +同时不能让 BugCollect 故障阻断宿主 App。 + +## 权威链路 + +1. SDK 本地按指纹采样和限频，批量上报最多 30 条。 +2. 服务端按 `appKey` 统计固定窗口内的事件数，默认每分钟 6000 条。 +3. 多实例部署使用 Redis Lua 原子计数；Redis 不可用时退回进程内窗口，继续保护当前实例。 +4. 超限返回 HTTP 429、`message=BUGCOLLECT_RATE_LIMITED` 和标准 + `Retry-After` 秒数。 +5. RN/Android SDK 保留原队列，按 `Retry-After` 冷却，最长一小时；冷却期间不重复请求。 +6. `BUGCOLLECT_DISABLED` 仍表示租户关闭服务，SDK 清队列并停用。它与临时限流不能混用。 + +## 配置 + +| 环境变量 | 默认值 | 说明 | +|---|---:|---| +| `BUGCOLLECT_MAX_EVENTS_PER_MINUTE` | `6000` | 单个 appKey 每个窗口允许的事件数，最小 30 | +| `BUGCOLLECT_RATE_WINDOW_SECONDS` | `60` | 固定窗口长度，最小 10 秒 | + +限流属于平台容量保护，不是租户发布配置。租户平台不向开发者暴露“上传网络建议”或 +让租户关闭服务端保护。 + +## 验收 + +- 单批 31 条必须在 Bean Validation 阶段拒绝。 +- Redis 可用时，多实例共享同一 appKey 计数。 +- Redis 不可用时，单实例超过本地额度仍返回 429。 +- 429 必须携带正整数 `Retry-After`。 +- SDK 收到 429 后不得丢队列、立即重试或向宿主业务线程抛未处理异常。 +- 服务关闭返回 403 `BUGCOLLECT_DISABLED`，行为不受限流改动影响。 diff --git a/xuqm-bugcollect-service/src/main/java/com/xuqm/bugcollect/controller/GlobalExceptionHandler.java b/xuqm-bugcollect-service/src/main/java/com/xuqm/bugcollect/controller/GlobalExceptionHandler.java index d3e2e10..86a746c 100644 --- a/xuqm-bugcollect-service/src/main/java/com/xuqm/bugcollect/controller/GlobalExceptionHandler.java +++ b/xuqm-bugcollect-service/src/main/java/com/xuqm/bugcollect/controller/GlobalExceptionHandler.java @@ -3,6 +3,7 @@ package com.xuqm.bugcollect.controller; import com.xuqm.common.exception.BusinessException; import com.xuqm.common.model.ApiResponse; import com.xuqm.bugcollect.service.BugCollectDisabledException; +import com.xuqm.bugcollect.service.IngestionRateLimitedException; import jakarta.servlet.http.HttpServletRequest; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -30,6 +31,13 @@ public class GlobalExceptionHandler { .body(ApiResponse.error(403, BugCollectDisabledException.ERROR_CODE)); } + @ExceptionHandler(IngestionRateLimitedException.class) + public ResponseEntity> handle(IngestionRateLimitedException ex) { + return ResponseEntity.status(HttpStatus.TOO_MANY_REQUESTS) + .header("Retry-After", Long.toString(ex.retryAfterSeconds())) + .body(ApiResponse.error(429, IngestionRateLimitedException.ERROR_CODE)); + } + @ExceptionHandler(BusinessException.class) public ResponseEntity> handle(BusinessException ex, HttpServletRequest request) { if (ex.getCode() >= 500) { diff --git a/xuqm-bugcollect-service/src/main/java/com/xuqm/bugcollect/controller/LogController.java b/xuqm-bugcollect-service/src/main/java/com/xuqm/bugcollect/controller/LogController.java index b3aee06..cb772e2 100644 --- a/xuqm-bugcollect-service/src/main/java/com/xuqm/bugcollect/controller/LogController.java +++ b/xuqm-bugcollect-service/src/main/java/com/xuqm/bugcollect/controller/LogController.java @@ -7,6 +7,7 @@ import com.xuqm.bugcollect.service.LogService; import com.xuqm.bugcollect.service.SourcemapService; import com.xuqm.bugcollect.service.WebhookService; import com.xuqm.bugcollect.service.BugCollectAvailabilityService; +import com.xuqm.bugcollect.service.IngestionRateLimiter; import jakarta.validation.Valid; import org.springframework.data.domain.Page; import org.springframework.security.access.AccessDeniedException; @@ -25,20 +26,28 @@ public class LogController { private final SourcemapService sourcemapService; private final WebhookService webhookService; private final BugCollectAvailabilityService availabilityService; + private final IngestionRateLimiter ingestionRateLimiter; public LogController(LogService logService, SourcemapService sourcemapService, WebhookService webhookService, - BugCollectAvailabilityService availabilityService) { + BugCollectAvailabilityService availabilityService, + IngestionRateLimiter ingestionRateLimiter) { this.logService = logService; this.sourcemapService = sourcemapService; this.webhookService = webhookService; this.availabilityService = availabilityService; + this.ingestionRateLimiter = ingestionRateLimiter; } // ── Ingestion ────────────────────────────────────────────────────────────── @PostMapping("/issues/batch") public ApiResponse ingestIssues(@Valid @RequestBody IssueBatchRequest request) { + request.events().stream() + .collect(java.util.stream.Collectors.groupingBy( + IssueBatchRequest.IssueEventItem::appKey, + java.util.stream.Collectors.counting())) + .forEach((appKey, count) -> ingestionRateLimiter.acquire(appKey, count.intValue())); request.events().stream() .map(item -> item.appKey() + "\n" + item.platform()) .distinct() @@ -52,6 +61,11 @@ public class LogController { @PostMapping("/events/batch") public ApiResponse ingestEvents(@Valid @RequestBody EventBatchRequest request) { + request.events().stream() + .collect(java.util.stream.Collectors.groupingBy( + EventBatchRequest.EventItem::appKey, + java.util.stream.Collectors.counting())) + .forEach((appKey, count) -> ingestionRateLimiter.acquire(appKey, count.intValue())); request.events().stream() .map(item -> item.appKey() + "\n" + item.platform()) .distinct() diff --git a/xuqm-bugcollect-service/src/main/java/com/xuqm/bugcollect/dto/EventBatchRequest.java b/xuqm-bugcollect-service/src/main/java/com/xuqm/bugcollect/dto/EventBatchRequest.java index 3c50350..a5bc399 100644 --- a/xuqm-bugcollect-service/src/main/java/com/xuqm/bugcollect/dto/EventBatchRequest.java +++ b/xuqm-bugcollect-service/src/main/java/com/xuqm/bugcollect/dto/EventBatchRequest.java @@ -3,12 +3,13 @@ package com.xuqm.bugcollect.dto; import com.fasterxml.jackson.annotation.JsonProperty; import jakarta.validation.constraints.NotBlank; import jakarta.validation.constraints.NotEmpty; +import jakarta.validation.constraints.Size; import java.util.List; public record EventBatchRequest( @JsonProperty("sentAt") String sentAt, @JsonProperty("sdk") IssueBatchRequest.SdkInfo sdk, - @JsonProperty("events") @NotEmpty List events + @JsonProperty("events") @NotEmpty @Size(max = 30) List events ) { public record EventItem( @JsonProperty("eventId") String eventId, diff --git a/xuqm-bugcollect-service/src/main/java/com/xuqm/bugcollect/dto/IssueBatchRequest.java b/xuqm-bugcollect-service/src/main/java/com/xuqm/bugcollect/dto/IssueBatchRequest.java index 0c77215..cf21357 100644 --- a/xuqm-bugcollect-service/src/main/java/com/xuqm/bugcollect/dto/IssueBatchRequest.java +++ b/xuqm-bugcollect-service/src/main/java/com/xuqm/bugcollect/dto/IssueBatchRequest.java @@ -3,13 +3,14 @@ package com.xuqm.bugcollect.dto; import com.fasterxml.jackson.annotation.JsonProperty; import jakarta.validation.constraints.NotBlank; import jakarta.validation.constraints.NotEmpty; +import jakarta.validation.constraints.Size; import java.util.List; import java.util.Map; public record IssueBatchRequest( @JsonProperty("sentAt") String sentAt, @JsonProperty("sdk") SdkInfo sdk, - @JsonProperty("events") @NotEmpty List events + @JsonProperty("events") @NotEmpty @Size(max = 30) List events ) { public record SdkInfo( String name, diff --git a/xuqm-bugcollect-service/src/main/java/com/xuqm/bugcollect/service/IngestionRateLimitedException.java b/xuqm-bugcollect-service/src/main/java/com/xuqm/bugcollect/service/IngestionRateLimitedException.java new file mode 100644 index 0000000..e8b83d4 --- /dev/null +++ b/xuqm-bugcollect-service/src/main/java/com/xuqm/bugcollect/service/IngestionRateLimitedException.java @@ -0,0 +1,23 @@ +package com.xuqm.bugcollect.service; + +/** + * BugCollect 摄取保护触发时返回给 SDK 的稳定信号。 + * + *

客户端收到该信号后应保留本地队列，并至少等待 {@link #retryAfterSeconds()} + * 后再尝试，不能把限流当作永久停用。

+ */ +public final class IngestionRateLimitedException extends RuntimeException { + + public static final String ERROR_CODE = "BUGCOLLECT_RATE_LIMITED"; + + private final long retryAfterSeconds; + + public IngestionRateLimitedException(long retryAfterSeconds) { + super(ERROR_CODE); + this.retryAfterSeconds = Math.max(1, retryAfterSeconds); + } + + public long retryAfterSeconds() { + return retryAfterSeconds; + } +} diff --git a/xuqm-bugcollect-service/src/main/java/com/xuqm/bugcollect/service/IngestionRateLimiter.java b/xuqm-bugcollect-service/src/main/java/com/xuqm/bugcollect/service/IngestionRateLimiter.java new file mode 100644 index 0000000..00862e1 --- /dev/null +++ b/xuqm-bugcollect-service/src/main/java/com/xuqm/bugcollect/service/IngestionRateLimiter.java @@ -0,0 +1,102 @@ +package com.xuqm.bugcollect.service; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.data.redis.core.script.DefaultRedisScript; +import org.springframework.stereotype.Service; + +import java.time.Clock; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +/** + * 按 appKey 统计摄取事件数，防止线上同一错误集中爆发时压垮数据库。 + * + *

Redis Lua 保证多实例计数原子；Redis 暂不可用时退回进程内固定窗口， + * 继续提供单实例保护，不把基础设施异常传播给 App。

+ */ +@Service +public class IngestionRateLimiter { + + private static final Logger log = LoggerFactory.getLogger(IngestionRateLimiter.class); + private static final DefaultRedisScript LIMIT_SCRIPT = new DefaultRedisScript<>(""" + local current = redis.call('INCRBY', KEYS[1], ARGV[1]) + if current == tonumber(ARGV[1]) then + redis.call('PEXPIRE', KEYS[1], ARGV[3]) + end + if current > tonumber(ARGV[2]) then + return redis.call('PTTL', KEYS[1]) + end + return 0 + """, Long.class); + + private final StringRedisTemplate redis; + private final long maxEventsPerWindow; + private final long windowMillis; + private final Clock clock; + private final Map localWindows = new ConcurrentHashMap<>(); + + public IngestionRateLimiter( + StringRedisTemplate redis, + @Value("${bugcollect-service.ingestion.max-events-per-minute:6000}") long maxEventsPerMinute, + @Value("${bugcollect-service.ingestion.window-seconds:60}") long windowSeconds) { + this(redis, maxEventsPerMinute, windowSeconds, Clock.systemUTC()); + } + + IngestionRateLimiter( + StringRedisTemplate redis, + long maxEventsPerMinute, + long windowSeconds, + Clock clock) { + this.redis = redis; + this.maxEventsPerWindow = Math.max(30, maxEventsPerMinute); + this.windowMillis = Math.max(10, windowSeconds) * 1_000L; + this.clock = clock; + } + + public void acquire(String appKey, int eventCount) { + if (appKey == null || appKey.isBlank() || eventCount <= 0) return; + String normalized = appKey.trim(); + try { + Long retryAfterMillis = redis.execute( + LIMIT_SCRIPT, + List.of("xuqm:bugcollect:ingestion:" + normalized), + Integer.toString(eventCount), + Long.toString(maxEventsPerWindow), + Long.toString(windowMillis)); + if (retryAfterMillis != null && retryAfterMillis > 0) { + throw new IngestionRateLimitedException(toSeconds(retryAfterMillis)); + } + return; + } catch (IngestionRateLimitedException ex) { + throw ex; + } catch (RuntimeException ex) { + log.warn("BugCollect Redis 限流不可用，退回进程内保护: appKey={} cause={}", + normalized, ex.getClass().getSimpleName()); + } + acquireLocally(normalized, eventCount); + } + + private void acquireLocally(String appKey, int eventCount) { + long now = clock.millis(); + LocalWindow current = localWindows.compute(appKey, (key, previous) -> { + if (previous == null || now >= previous.endsAtMillis()) { + return new LocalWindow(eventCount, now + windowMillis); + } + return new LocalWindow(previous.count() + eventCount, previous.endsAtMillis()); + }); + if (current.count() > maxEventsPerWindow) { + throw new IngestionRateLimitedException(toSeconds(current.endsAtMillis() - now)); + } + } + + private static long toSeconds(long millis) { + return Math.max(1, (millis + 999L) / 1_000L); + } + + private record LocalWindow(long count, long endsAtMillis) { + } +} diff --git a/xuqm-bugcollect-service/src/main/resources/application.yml b/xuqm-bugcollect-service/src/main/resources/application.yml index 1f0a6d4..135328b 100644 --- a/xuqm-bugcollect-service/src/main/resources/application.yml +++ b/xuqm-bugcollect-service/src/main/resources/application.yml @@ -5,9 +5,9 @@ spring: application: name: xuqm-bugcollect-service datasource: - url: jdbc:mysql://39.107.53.187:3306/xuqm_bugcollect?useUnicode=true&characterEncoding=utf8&useSSL=true&serverTimezone=GMT%2B8&allowPublicKeyRetrieval=true - username: xuqm - password: Xuqm@2026 + url: ${BUGCOLLECT_DB_URL:jdbc:mysql://127.0.0.1:3306/xuqm_bugcollect?useUnicode=true&characterEncoding=utf8&useSSL=true&serverTimezone=GMT%2B8&allowPublicKeyRetrieval=true} + username: ${BUGCOLLECT_DB_USERNAME:xuqm} + password: ${BUGCOLLECT_DB_PASSWORD:} driver-class-name: com.mysql.cj.jdbc.Driver hikari: minimum-idle: 5 @@ -25,10 +25,10 @@ spring: format_sql: true data: redis: - host: redisdev.xuqinmin.com - port: 6379 - password: xuqinmin1022 - database: 2 + host: ${BUGCOLLECT_REDIS_HOST:127.0.0.1} + port: ${BUGCOLLECT_REDIS_PORT:6379} + password: ${BUGCOLLECT_REDIS_PASSWORD:} + database: ${BUGCOLLECT_REDIS_DATABASE:2} timeout: 10s lettuce: pool: @@ -58,6 +58,9 @@ sdk: bugcollect-service: tenant-sdk-config-url: ${TENANT_SDK_CONFIG_URL:http://127.0.0.1:8081/api/sdk/config} availability-cache-seconds: ${BUGCOLLECT_AVAILABILITY_CACHE_SECONDS:30} + ingestion: + max-events-per-minute: ${BUGCOLLECT_MAX_EVENTS_PER_MINUTE:6000} + window-seconds: ${BUGCOLLECT_RATE_WINDOW_SECONDS:60} sourcemap: storage-dir: ${LOG_SOURCEMAP_DIR:/data/bugcollect-service/sourcemaps} webhook: diff --git a/xuqm-bugcollect-service/src/test/java/com/xuqm/bugcollect/controller/GlobalExceptionHandlerTest.java b/xuqm-bugcollect-service/src/test/java/com/xuqm/bugcollect/controller/GlobalExceptionHandlerTest.java index a79329a..d811854 100644 --- a/xuqm-bugcollect-service/src/test/java/com/xuqm/bugcollect/controller/GlobalExceptionHandlerTest.java +++ b/xuqm-bugcollect-service/src/test/java/com/xuqm/bugcollect/controller/GlobalExceptionHandlerTest.java @@ -1,6 +1,7 @@ package com.xuqm.bugcollect.controller; import com.xuqm.bugcollect.service.BugCollectDisabledException; +import com.xuqm.bugcollect.service.IngestionRateLimitedException; import org.junit.jupiter.api.Test; import static org.assertj.core.api.Assertions.assertThat; @@ -15,4 +16,14 @@ class GlobalExceptionHandlerTest { assertThat(response.getBody()).isNotNull(); assertThat(response.getBody().message()).isEqualTo("BUGCOLLECT_DISABLED"); } + + @Test + void exposesRetryAfterForIngestionBackpressure() { + var response = new GlobalExceptionHandler().handle(new IngestionRateLimitedException(27)); + + assertThat(response.getStatusCode().value()).isEqualTo(429); + assertThat(response.getHeaders().getFirst("Retry-After")).isEqualTo("27"); + assertThat(response.getBody()).isNotNull(); + assertThat(response.getBody().message()).isEqualTo("BUGCOLLECT_RATE_LIMITED"); + } } diff --git a/xuqm-bugcollect-service/src/test/java/com/xuqm/bugcollect/service/IngestionRateLimiterTest.java b/xuqm-bugcollect-service/src/test/java/com/xuqm/bugcollect/service/IngestionRateLimiterTest.java new file mode 100644 index 0000000..83658c1 --- /dev/null +++ b/xuqm-bugcollect-service/src/test/java/com/xuqm/bugcollect/service/IngestionRateLimiterTest.java @@ -0,0 +1,46 @@ +package com.xuqm.bugcollect.service; + +import org.junit.jupiter.api.Test; +import org.springframework.data.redis.core.StringRedisTemplate; + +import java.time.Clock; +import java.time.Instant; +import java.time.ZoneOffset; + +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +class IngestionRateLimiterTest { + + @Test + void usesDistributedRetryWindowReturnedByRedis() { + StringRedisTemplate redis = mock(StringRedisTemplate.class); + when(redis.execute(any(), any(), any(Object[].class))) + .thenReturn(0L, 27_000L); + IngestionRateLimiter limiter = new IngestionRateLimiter( + redis, 6_000, 60, Clock.systemUTC()); + + limiter.acquire("app-key", 30); + assertThatThrownBy(() -> limiter.acquire("app-key", 30)) + .isInstanceOf(IngestionRateLimitedException.class) + .satisfies(error -> org.assertj.core.api.Assertions.assertThat( + ((IngestionRateLimitedException) error).retryAfterSeconds()).isEqualTo(27)); + } + + @Test + void fallsBackToLocalWindowWhenRedisIsUnavailable() { + StringRedisTemplate redis = mock(StringRedisTemplate.class); + when(redis.execute(any(), any(), any(Object[].class))) + .thenThrow(new IllegalStateException("redis unavailable")); + Clock clock = Clock.fixed(Instant.ofEpochMilli(10_000), ZoneOffset.UTC); + IngestionRateLimiter limiter = new IngestionRateLimiter(redis, 30, 60, clock); + + limiter.acquire("app-key", 20); + assertThatThrownBy(() -> limiter.acquire("app-key", 11)) + .isInstanceOf(IngestionRateLimitedException.class) + .satisfies(error -> org.assertj.core.api.Assertions.assertThat( + ((IngestionRateLimitedException) error).retryAfterSeconds()).isEqualTo(60)); + } +}