task-61: Rustfs 对象存储指标基线验收测试（计数/计时/分布/熔断/拒绝）
This commit is contained in:
+262
@@ -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<MeterRegistry> meterRegistryProvider;
|
||||||
|
private ObjectProvider<RustfsDeleteRetryService> 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, "熔断已复位");
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user