diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/file/service/object/RustfsMetricsBaselineTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/file/service/object/RustfsMetricsBaselineTest.java new file mode 100644 index 00000000..1bd58435 --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/file/service/object/RustfsMetricsBaselineTest.java @@ -0,0 +1,262 @@ +package com.nanri.aiimage.modules.file.service.object; + +import com.nanri.aiimage.config.TransientStorageProperties; +import io.micrometer.core.instrument.Counter; +import io.micrometer.core.instrument.DistributionSummary; +import io.micrometer.core.instrument.MeterRegistry; +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; +import io.minio.GetObjectResponse; +import io.minio.MinioClient; +import io.minio.RemoveObjectArgs; +import io.minio.PutObjectArgs; +import io.minio.StatObjectArgs; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentMatchers; +import org.springframework.beans.factory.ObjectProvider; +import org.springframework.test.util.ReflectionTestUtils; + +import java.io.ByteArrayInputStream; +import java.nio.charset.StandardCharsets; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.doNothing; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** + * Task 61：共享任务链路资源指标基线：线程、连接、队列、GC、Redis、RustFS 和 DB。 + * RustfsObjectStorageService 为上传/读取/删除/stat 记录 Micrometer 指标 + * （aiimage.rustfs.operation.total 计数、duration 计时、payload.bytes 分布、 + * fallback.local.total 计数）与并发信号量上限、超时 deadline、失败窗口熔断。 + * 指标基线：未配置零指标、成功/重试/失败/拒绝各结果分类计数准确、重复操作 + * 计数精确累加、空输入零调用、单元素计数正确、并发超限 rejected、熔断后恢复。 + */ +class RustfsMetricsBaselineTest { + + private TransientStorageProperties properties; + private SimpleMeterRegistry simpleRegistry; + private MinioClient client; + private ObjectProvider meterRegistryProvider; + private ObjectProvider deleteRetryProvider; + private RustfsObjectStorageService service; + + @BeforeEach + @SuppressWarnings("unchecked") + void setUp() { + properties = new TransientStorageProperties(); + simpleRegistry = new SimpleMeterRegistry(); + meterRegistryProvider = mock(ObjectProvider.class); + when(meterRegistryProvider.getIfAvailable()).thenReturn(simpleRegistry); + deleteRetryProvider = mock(ObjectProvider.class); + client = mock(MinioClient.class); + service = new RustfsObjectStorageService(properties, meterRegistryProvider, deleteRetryProvider, + () -> client); + } + + @AfterEach + void tearDown() { + simpleRegistry.clear(); + } + + private static void setConfigured(TransientStorageProperties properties) { + properties.setEndpoint("http://rustfs.local:9000"); + properties.setBucket("bucket"); + properties.setAccessKeyId("ak"); + properties.setAccessKeySecret("sk"); + properties.setRegion("us-east-1"); + } + + private static long counter(MeterRegistry registry, String name, String operation, String result) { + Counter counter = registry.find(name).tags("operation", operation, "result", result).counter(); + return counter == null ? 0L : (long) counter.count(); + } + + private GetObjectResponse readResponse(String content) throws Exception { + GetObjectResponse response = mock(GetObjectResponse.class); + when(response.readAllBytes()).thenReturn(content.getBytes(StandardCharsets.UTF_8)); + return response; + } + + @Test + void test_task_061_rustfs_metrics_normal_default_path() { + // 正常路径：未配置时零调用零指标（基线默认），调用抛可识别异常。 + assertEquals(0L, counter(simpleRegistry, "aiimage.rustfs.operation.total", "upload", "success"), + "未配置零指标"); + assertThrows(IllegalStateException.class, () -> service.uploadText("k", "v"), + "未配置抛可识别异常"); + assertEquals(0L, counter(simpleRegistry, "aiimage.rustfs.operation.total", "upload", "success"), + "失败调用不产生 success 指标"); + } + + @Test + void test_task_061_rustfs_metrics_normal_multiple_items() throws Exception { + // 批量场景：多次操作各结果分类计数准确累加（retry + failure），无丢失无重复。 + setConfigured(properties); + properties.setReadMaxRetries(2); + properties.setUploadMaxRetries(2); + properties.setDeleteMaxRetries(2); + properties.setBaseRetryDelayMillis(0); + properties.setMaxRetryDelayMillis(0); + // 高失败窗口阈值：避免失败计数触发熔断，干扰各结果分类计数断言。 + properties.setFailureWindowThreshold(1000); + + doThrow(new java.io.IOException("rustfs down")).when(client) + .getObject(ArgumentMatchers.any(io.minio.GetObjectArgs.class)); + doThrow(new java.io.IOException("rustfs down")).when(client) + .putObject(ArgumentMatchers.any(PutObjectArgs.class)); + doThrow(new java.io.IOException("rustfs down")).when(client) + .removeObject(ArgumentMatchers.any(RemoveObjectArgs.class)); + + assertThrows(Exception.class, () -> service.uploadText("a", "v")); + assertThrows(Exception.class, () -> service.readObjectAsString("b")); + assertThrows(Exception.class, () -> service.deleteObject("c")); + + assertEquals(1L, counter(simpleRegistry, "aiimage.rustfs.operation.total", "upload", "retry")); + assertEquals(1L, counter(simpleRegistry, "aiimage.rustfs.operation.total", "read", "retry")); + assertEquals(1L, counter(simpleRegistry, "aiimage.rustfs.operation.total", "delete", "retry")); + assertEquals(1L, counter(simpleRegistry, "aiimage.rustfs.operation.total", "upload", "failure")); + assertEquals(1L, counter(simpleRegistry, "aiimage.rustfs.operation.total", "read", "failure")); + assertEquals(1L, counter(simpleRegistry, "aiimage.rustfs.operation.total", "delete", "failure")); + } + + @Test + void test_task_061_rustfs_metrics_normal_repeated_operation_is_idempotent() throws Exception { + // 幂等：同一操作重复执行成功 → 指标按次数精确累加，不重复注册不丢计数。 + setConfigured(properties); + + GetObjectResponse readResponse = readResponse("{}"); + when(client.getObject(ArgumentMatchers.any(io.minio.GetObjectArgs.class))) + .thenReturn(readResponse); + when(client.statObject(ArgumentMatchers.any(StatObjectArgs.class))) + .thenReturn(null); + when(client.putObject(ArgumentMatchers.any(PutObjectArgs.class))) + .thenReturn(mock(io.minio.ObjectWriteResponse.class)); + doNothing().when(client).removeObject(ArgumentMatchers.any(RemoveObjectArgs.class)); + + service.uploadText("k", "v"); + service.uploadText("k", "v"); + service.readObjectAsString("k"); + service.readObjectAsString("k"); + service.deleteObject("k"); + service.deleteObject("k"); + + assertEquals(2L, counter(simpleRegistry, "aiimage.rustfs.operation.total", "upload", "success")); + assertEquals(2L, counter(simpleRegistry, "aiimage.rustfs.operation.total", "read", "success")); + assertEquals(2L, counter(simpleRegistry, "aiimage.rustfs.operation.total", "delete", "success")); + assertEquals(2L, counter(simpleRegistry, "aiimage.rustfs.operation.total", "stat", "success"), + "上传后 stat 可见性校验 2 次"); + } + + @Test + void test_task_061_rustfs_metrics_boundary_empty_input() { + // 空输入：空对象键删除零调用零指标，不创建无效资源。 + setConfigured(properties); + service.deleteObject(""); + service.deleteObject(null); + + assertEquals(0L, counter(simpleRegistry, "aiimage.rustfs.operation.total", "delete", "success")); + } + + @Test + void test_task_061_rustfs_metrics_boundary_single_item() throws Exception { + // 单元素：单次上传记录 success 计数与 payload.bytes 分布，无并发上限触发。 + setConfigured(properties); + + when(client.putObject(ArgumentMatchers.any(PutObjectArgs.class))) + .thenReturn(mock(io.minio.ObjectWriteResponse.class)); + when(client.statObject(ArgumentMatchers.any(StatObjectArgs.class))).thenReturn(null); + + service.uploadText("solo", "hello"); + + assertEquals(1L, counter(simpleRegistry, "aiimage.rustfs.operation.total", "upload", "success")); + DistributionSummary summary = simpleRegistry.find("aiimage.rustfs.payload.bytes").summary(); + assertTrue(summary != null && summary.count() >= 1L, "payload 字节分布有记录"); + assertTrue(summary != null && summary.totalAmount() >= 5L, "记录字节数与内容一致"); + } + + @Test + void test_task_061_rustfs_metrics_boundary_limit_and_overflow() throws Exception { + // 上限/超限：上传并发上限 1 + 获取许可超时 0 → 第二并发请求被拒绝， + // rejected 指标 +1，不发生无界排队。 + setConfigured(properties); + properties.setMaxConcurrentUploads(1); + properties.setAcquirePermitTimeoutMillis(0); + // 信号量在构造时从 properties 固定，须先设置上限再构造受限实例。 + RustfsObjectStorageService limited = new RustfsObjectStorageService( + properties, meterRegistryProvider, deleteRetryProvider, () -> client); + + java.util.concurrent.CountDownLatch entered = new java.util.concurrent.CountDownLatch(1); + org.mockito.Mockito.doAnswer(invocation -> { + entered.countDown(); + Thread.sleep(500); + throw new java.io.IOException("hold"); + }).when(client).putObject(ArgumentMatchers.any(PutObjectArgs.class)); + + Thread first = new Thread(() -> { + try { + limited.uploadText("block", "v"); + } catch (Exception ignored) { + } + }); + first.setDaemon(true); + first.start(); + try { + assertTrue(entered.await(2, java.util.concurrent.TimeUnit.SECONDS), + "首线程已持有许可"); + } catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + } + assertThrows(IllegalStateException.class, () -> limited.uploadText("blocked", "v"), + "并发超限被拒绝"); + assertEquals(1L, counter(simpleRegistry, "aiimage.rustfs.operation.total", "upload", "rejected")); + first.interrupt(); + } + + @Test + void test_task_061_rustfs_metrics_invalid_input_rejected() { + // 非法参数：未配置时上传/读取抛可识别异常且零指标（不污染基线）。 + assertThrows(IllegalStateException.class, () -> service.uploadText("k", "v"), + "未配置抛可识别异常"); + assertThrows(IllegalStateException.class, () -> service.readObjectAsString("k"), + "未配置抛可识别异常"); + assertEquals(0L, counter(simpleRegistry, "aiimage.rustfs.operation.total", "upload", "success")); + assertEquals(0L, counter(simpleRegistry, "aiimage.rustfs.operation.total", "read", "success")); + } + + @Test + void test_task_061_rustfs_metrics_dependency_failure_releases_resources() throws Exception { + // 依赖失败：上传成功但 stat 可见性校验失败 → 失败窗口打开， + // 后续读取被熔断拒绝（rejected 指标）；熔断过期后自动恢复，指标完整。 + setConfigured(properties); + properties.setFailureWindowThreshold(1); + properties.setFailureCooldownMillis(5_000); + properties.setFailureWindowSeconds(10); + properties.setDeleteRetryEnabled(false); + + when(client.putObject(ArgumentMatchers.any(PutObjectArgs.class))) + .thenReturn(mock(io.minio.ObjectWriteResponse.class)); + doThrow(new java.io.IOException("stat down")).when(client) + .statObject(ArgumentMatchers.any(StatObjectArgs.class)); + GetObjectResponse readResponse = readResponse("{}"); + when(client.getObject(ArgumentMatchers.any(io.minio.GetObjectArgs.class))) + .thenReturn(readResponse); + + assertThrows(Exception.class, () -> service.uploadText("k", "v"), "stat 失败上传抛错"); + assertThrows(IllegalStateException.class, () -> service.readObjectAsString("k"), + "熔断打开读取被拒绝"); + assertEquals(1L, counter(simpleRegistry, "aiimage.rustfs.operation.total", "read", "rejected")); + + // 熔断冷却到期后自动恢复。 + ReflectionTestUtils.setField(service, "circuitOpenUntilMillis", 0L); + assertEquals("{}", service.readObjectAsString("k"), "熔断重置后恢复"); + assertEquals(1L, counter(simpleRegistry, "aiimage.rustfs.operation.total", "read", "success")); + Object openUntil = ReflectionTestUtils.getField(service, "circuitOpenUntilMillis"); + assertFalse(openUntil instanceof Long && (Long) openUntil > 0L, "熔断已复位"); + } +}