lhq/FT546-WCS
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(wcs): add correlated logs and product metrics

Browse Source
This commit is contained in:
zgw
2026-08-11 11:33:55 +08:00
parent 434cbe57b9
commit f9ebbcf51c
6 changed files with 369 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
ruoyi-wcs/pom.xml
View File

@@ -23,6 +23,10 @@
<groupId>org.springframework</groupId>
<artifactId>spring-tx</artifactId>
</dependency>
<dependency>
<groupId>io.micrometer</groupId>
<artifactId>micrometer-core</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>

76
ruoyi-wcs/src/main/java/com/ruoyi/wcs/observability/CorrelationContext.java Normal file
View File

@@ -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<String, String> 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;
}
}
}

171
ruoyi-wcs/src/main/java/com/ruoyi/wcs/observability/MicrometerWcsTelemetry.java Normal file
View File

@@ -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<String, AtomicLong> gauges =
new ConcurrentHashMap<String, AtomicLong>();
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;
}
}

21
ruoyi-wcs/src/main/java/com/ruoyi/wcs/observability/WcsMetricNames.java Normal file
View File

@@ -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() {
}
}

30
ruoyi-wcs/src/main/java/com/ruoyi/wcs/observability/WcsTelemetry.java Normal file
View File

@@ -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);
}

67
ruoyi-wcs/src/test/java/com/ruoyi/wcs/observability/WcsObservabilityTest.java Normal file
View File

@@ -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());
}
}