diff --git a/ruoyi-wcs/pom.xml b/ruoyi-wcs/pom.xml index 1f22ff3..52f39d2 100644 --- a/ruoyi-wcs/pom.xml +++ b/ruoyi-wcs/pom.xml @@ -23,6 +23,10 @@ org.springframework spring-tx + + io.micrometer + micrometer-core + org.springframework.boot spring-boot-starter-test diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/observability/CorrelationContext.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/observability/CorrelationContext.java new file mode 100644 index 0000000..4873fc9 --- /dev/null +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/observability/CorrelationContext.java @@ -0,0 +1,76 @@ +package com.ruoyi.wcs.observability; + +import org.slf4j.MDC; + +import java.util.Map; +import java.util.Objects; + +public final class CorrelationContext implements AutoCloseable { + private final Map previousContext; + private boolean closed; + + private CorrelationContext(Fields fields) { + this.previousContext = MDC.getCopyOfContextMap(); + put("traceId", fields.traceId); + put("externalRequestId", fields.externalRequestId); + put("orderId", fields.orderId); + put("workflowId", fields.workflowId); + put("stepId", fields.stepId); + put("deviceId", fields.deviceId); + put("commandKey", fields.commandKey); + put("resourceId", fields.resourceId); + put("eventId", fields.eventId); + put("recoveryCaseId", fields.recoveryCaseId); + } + + public static CorrelationContext open(Fields fields) { + return new CorrelationContext(Objects.requireNonNull(fields, "fields")); + } + + @Override + public void close() { + if (closed) { + return; + } + MDC.clear(); + if (previousContext != null && !previousContext.isEmpty()) { + MDC.setContextMap(previousContext); + } + closed = true; + } + + private static void put(String key, String value) { + if (value != null && !value.trim().isEmpty()) { + MDC.put(key, value.trim()); + } + } + + public static final class Fields { + private final String traceId; + private final String externalRequestId; + private final String orderId; + private final String workflowId; + private final String stepId; + private final String deviceId; + private final String commandKey; + private final String resourceId; + private final String eventId; + private final String recoveryCaseId; + + public Fields(String traceId, String externalRequestId, String orderId, + String workflowId, String stepId, String deviceId, + String commandKey, String resourceId, String eventId, + String recoveryCaseId) { + this.traceId = traceId; + this.externalRequestId = externalRequestId; + this.orderId = orderId; + this.workflowId = workflowId; + this.stepId = stepId; + this.deviceId = deviceId; + this.commandKey = commandKey; + this.resourceId = resourceId; + this.eventId = eventId; + this.recoveryCaseId = recoveryCaseId; + } + } +} diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/observability/MicrometerWcsTelemetry.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/observability/MicrometerWcsTelemetry.java new file mode 100644 index 0000000..2438e21 --- /dev/null +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/observability/MicrometerWcsTelemetry.java @@ -0,0 +1,171 @@ +package com.ruoyi.wcs.observability; + +import io.micrometer.core.instrument.Counter; +import io.micrometer.core.instrument.Gauge; +import io.micrometer.core.instrument.MeterRegistry; +import io.micrometer.core.instrument.Tags; +import io.micrometer.core.instrument.Timer; + +import java.time.Duration; +import java.util.Objects; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.atomic.AtomicLong; + +public final class MicrometerWcsTelemetry implements WcsTelemetry { + private static final int MAX_TAG_LENGTH = 64; + + private final MeterRegistry registry; + private final String warehouse; + private final ConcurrentMap gauges = + new ConcurrentHashMap(); + + public MicrometerWcsTelemetry(MeterRegistry registry, String warehouse) { + this.registry = Objects.requireNonNull(registry, "registry"); + this.warehouse = tag(warehouse, "warehouse"); + } + + @Override + public void setTaskBacklog(long backlog, String orderType) { + if (backlog < 0L) { + throw new IllegalArgumentException("backlog must not be negative"); + } + String type = tag(orderType, "orderType"); + longGauge(WcsMetricNames.TASK_BACKLOG, + Tags.of("warehouse", warehouse, "orderType", type)).set(backlog); + } + + @Override + public void recordDispatch(Duration duration, String orderType, + String outcome) { + timer(WcsMetricNames.DISPATCH_DURATION, duration, + Tags.of("warehouse", warehouse, + "orderType", tag(orderType, "orderType"), + "outcome", tag(outcome, "outcome"))); + } + + @Override + public void recordTrafficWait(Duration duration, String outcome) { + timer(WcsMetricNames.TRAFFIC_WAIT_DURATION, duration, + Tags.of("warehouse", warehouse, + "outcome", tag(outcome, "outcome"))); + } + + @Override + public void recordCommandAcknowledgement(Duration duration, + String commandType, + String outcome) { + commandTimer(WcsMetricNames.COMMAND_ACK_DURATION, duration, + commandType, outcome); + } + + @Override + public void recordCommandCompletion(Duration duration, String commandType, + String outcome) { + commandTimer(WcsMetricNames.COMMAND_COMPLETE_DURATION, duration, + commandType, outcome); + } + + @Override + public void setDeviceUtilization(double utilization, String deviceType) { + doubleGauge(WcsMetricNames.DEVICE_UTILIZATION, utilization, + deviceTags(deviceType)); + } + + @Override + public void setDeviceBattery(double battery, String deviceType) { + doubleGauge(WcsMetricNames.DEVICE_BATTERY, battery, + deviceTags(deviceType)); + } + + @Override + public void recordLiftWait(Duration duration, String outcome) { + timer(WcsMetricNames.LIFT_WAIT_DURATION, duration, + Tags.of("warehouse", warehouse, + "outcome", tag(outcome, "outcome"))); + } + + @Override + public void incrementOutboxRetry(String outboxType, String outcome) { + Counter.builder(WcsMetricNames.OUTBOX_RETRY_COUNT) + .tags("warehouse", warehouse, + "outboxType", tag(outboxType, "outboxType"), + "outcome", tag(outcome, "outcome")) + .register(registry).increment(); + } + + @Override + public void incrementDeviceFault(String deviceType, String outcome) { + Counter.builder(WcsMetricNames.DEVICE_FAULT_COUNT) + .tags("warehouse", warehouse, + "deviceType", tag(deviceType, "deviceType"), + "outcome", tag(outcome, "outcome")) + .register(registry).increment(); + } + + @Override + public void recordRecovery(Duration duration, String recoveryLevel, + String outcome) { + timer(WcsMetricNames.RECOVERY_DURATION, duration, + Tags.of("warehouse", warehouse, + "recoveryLevel", tag(recoveryLevel, "recoveryLevel"), + "outcome", tag(outcome, "outcome"))); + } + + private void commandTimer(String name, Duration duration, + String commandType, String outcome) { + timer(name, duration, Tags.of("warehouse", warehouse, + "commandType", tag(commandType, "commandType"), + "outcome", tag(outcome, "outcome"))); + } + + private Tags deviceTags(String deviceType) { + return Tags.of("warehouse", warehouse, + "deviceType", tag(deviceType, "deviceType")); + } + + private void timer(String name, Duration duration, Tags tags) { + Duration measured = Objects.requireNonNull(duration, "duration"); + if (measured.isNegative()) { + throw new IllegalArgumentException("duration must not be negative"); + } + Timer.builder(name).tags(tags).register(registry).record(measured); + } + + private AtomicLong longGauge(String name, Tags tags) { + String key = name + '|' + tags.toString(); + return gauges.computeIfAbsent(key, ignored -> { + AtomicLong value = new AtomicLong(); + Gauge.builder(name, value, AtomicLong::doubleValue) + .tags(tags).register(registry); + return value; + }); + } + + private void doubleGauge(String name, double measurement, Tags tags) { + if (!Double.isFinite(measurement) || measurement < 0.0D) { + throw new IllegalArgumentException( + "gauge measurement must be finite and non-negative"); + } + String key = name + '|' + tags.toString(); + AtomicLong value = gauges.computeIfAbsent(key, ignored -> { + AtomicLong bits = new AtomicLong(); + Gauge.builder(name, bits, + holder -> Double.longBitsToDouble(holder.get())) + .tags(tags).register(registry); + return bits; + }); + value.set(Double.doubleToRawLongBits(measurement)); + } + + private static String tag(String value, String name) { + if (value == null || value.trim().isEmpty()) { + throw new IllegalArgumentException(name + " must not be blank"); + } + String result = value.trim(); + if (result.length() > MAX_TAG_LENGTH) { + throw new IllegalArgumentException(name + " exceeds bounded length"); + } + return result; + } +} diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/observability/WcsMetricNames.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/observability/WcsMetricNames.java new file mode 100644 index 0000000..b8a3420 --- /dev/null +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/observability/WcsMetricNames.java @@ -0,0 +1,21 @@ +package com.ruoyi.wcs.observability; + +public final class WcsMetricNames { + public static final String TASK_BACKLOG = "wcs.task.backlog"; + public static final String DISPATCH_DURATION = "wcs.dispatch.duration"; + public static final String TRAFFIC_WAIT_DURATION = + "wcs.traffic.wait.duration"; + public static final String COMMAND_ACK_DURATION = + "wcs.command.ack.duration"; + public static final String COMMAND_COMPLETE_DURATION = + "wcs.command.complete.duration"; + public static final String DEVICE_UTILIZATION = "wcs.device.utilization"; + public static final String DEVICE_BATTERY = "wcs.device.battery"; + public static final String LIFT_WAIT_DURATION = "wcs.lift.wait.duration"; + public static final String OUTBOX_RETRY_COUNT = "wcs.outbox.retry.count"; + public static final String DEVICE_FAULT_COUNT = "wcs.device.fault.count"; + public static final String RECOVERY_DURATION = "wcs.recovery.duration"; + + private WcsMetricNames() { + } +} diff --git a/ruoyi-wcs/src/main/java/com/ruoyi/wcs/observability/WcsTelemetry.java b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/observability/WcsTelemetry.java new file mode 100644 index 0000000..9b66ba1 --- /dev/null +++ b/ruoyi-wcs/src/main/java/com/ruoyi/wcs/observability/WcsTelemetry.java @@ -0,0 +1,30 @@ +package com.ruoyi.wcs.observability; + +import java.time.Duration; + +public interface WcsTelemetry { + void setTaskBacklog(long backlog, String orderType); + + void recordDispatch(Duration duration, String orderType, String outcome); + + void recordTrafficWait(Duration duration, String outcome); + + void recordCommandAcknowledgement(Duration duration, String commandType, + String outcome); + + void recordCommandCompletion(Duration duration, String commandType, + String outcome); + + void setDeviceUtilization(double utilization, String deviceType); + + void setDeviceBattery(double battery, String deviceType); + + void recordLiftWait(Duration duration, String outcome); + + void incrementOutboxRetry(String outboxType, String outcome); + + void incrementDeviceFault(String deviceType, String outcome); + + void recordRecovery(Duration duration, String recoveryLevel, + String outcome); +} diff --git a/ruoyi-wcs/src/test/java/com/ruoyi/wcs/observability/WcsObservabilityTest.java b/ruoyi-wcs/src/test/java/com/ruoyi/wcs/observability/WcsObservabilityTest.java new file mode 100644 index 0000000..d4914bc --- /dev/null +++ b/ruoyi-wcs/src/test/java/com/ruoyi/wcs/observability/WcsObservabilityTest.java @@ -0,0 +1,67 @@ +package com.ruoyi.wcs.observability; + +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.slf4j.MDC; + +import java.time.Duration; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; + +class WcsObservabilityTest { + + @AfterEach + void clearMdc() { + MDC.clear(); + } + + @Test + void writesAllCorrelationFieldsAndRestoresMdcAfterTheOperation() { + MDC.put("requestScope", "existing"); + CorrelationContext.Fields fields = new CorrelationContext.Fields( + "TRACE-1", "REQ-1", "ORDER-1", "WF-1", "STEP-1", + "VEH-1", "CMD-1", "NODE:A", "EVENT-1", "RC-1"); + + try (CorrelationContext ignored = CorrelationContext.open(fields)) { + assertEquals("TRACE-1", MDC.get("traceId")); + assertEquals("ORDER-1", MDC.get("orderId")); + assertEquals("CMD-1", MDC.get("commandKey")); + assertEquals("RC-1", MDC.get("recoveryCaseId")); + assertEquals("existing", MDC.get("requestScope")); + } + + assertNull(MDC.get("orderId")); + assertEquals("existing", MDC.get("requestScope")); + } + + @Test + void recordsRequiredMetricsUsingOnlyBoundedOperationalLabels() { + SimpleMeterRegistry registry = new SimpleMeterRegistry(); + WcsTelemetry telemetry = new MicrometerWcsTelemetry(registry, "FT546"); + + telemetry.recordDispatch(Duration.ofMillis(80), "INBOUND", "GRANTED"); + telemetry.incrementOutboxRetry("DEVICE_COMMAND", "NOT_SENT"); + telemetry.setTaskBacklog(12L, "INBOUND"); + telemetry.recordCommandAcknowledgement(Duration.ofMillis(30), + "MOVE", "ACKNOWLEDGED"); + telemetry.recordRecovery(Duration.ofSeconds(2), "DEVICE", "CONFIRMED"); + + assertEquals(1L, registry.counter(WcsMetricNames.OUTBOX_RETRY_COUNT, + "warehouse", "FT546", "outboxType", "DEVICE_COMMAND", + "outcome", "NOT_SENT").count(), 0.0D); + assertEquals(1L, registry.timer(WcsMetricNames.DISPATCH_DURATION, + "warehouse", "FT546", "orderType", "INBOUND", + "outcome", "GRANTED").count()); + assertEquals(12.0D, registry.get(WcsMetricNames.TASK_BACKLOG) + .tag("warehouse", "FT546").tag("orderType", "INBOUND") + .gauge().value(), 0.0D); + assertEquals(1L, registry.timer(WcsMetricNames.COMMAND_ACK_DURATION, + "warehouse", "FT546", "commandType", "MOVE", + "outcome", "ACKNOWLEDGED").count()); + assertEquals(1L, registry.timer(WcsMetricNames.RECOVERY_DURATION, + "warehouse", "FT546", "recoveryLevel", "DEVICE", + "outcome", "CONFIRMED").count()); + } +}