优化
This commit is contained in:
+103
@@ -0,0 +1,103 @@
|
||||
package com.baiyue.eventbridge.metrics;
|
||||
|
||||
import com.baiyue.eventbridge.mapper.BridgeInboxMapper;
|
||||
import io.micrometer.core.instrument.MeterRegistry;
|
||||
import io.micrometer.core.instrument.MultiGauge;
|
||||
import io.micrometer.core.instrument.Tags;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.scheduling.annotation.Scheduled;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
/** Periodically samples indexed Inbox aggregates; Prometheus scrapes only the cached gauges. */
|
||||
@Component
|
||||
@Slf4j
|
||||
public class EventBridgeInboxMetricsSampler {
|
||||
public static final String BACKLOG_METRIC = "baiyue.event.bridge.inbox.backlog";
|
||||
public static final String OLDEST_AGE_METRIC = "baiyue.event.bridge.inbox.oldest.age.seconds";
|
||||
public static final String DEAD_METRIC = "baiyue.event.bridge.inbox.dead";
|
||||
|
||||
private final BridgeInboxMapper inboxMapper;
|
||||
private final MultiGauge backlog;
|
||||
private final MultiGauge oldestAge;
|
||||
private final MultiGauge dead;
|
||||
|
||||
public EventBridgeInboxMetricsSampler(BridgeInboxMapper inboxMapper, MeterRegistry registry) {
|
||||
this.inboxMapper = inboxMapper;
|
||||
this.backlog = MultiGauge.builder(BACKLOG_METRIC)
|
||||
.description("Events waiting for downstream acknowledgement").register(registry);
|
||||
this.oldestAge = MultiGauge.builder(OLDEST_AGE_METRIC)
|
||||
.description("Age in seconds of the oldest pending Inbox event").register(registry);
|
||||
this.dead = MultiGauge.builder(DEAD_METRIC)
|
||||
.description("Current number of DEAD Inbox events").register(registry);
|
||||
}
|
||||
|
||||
@Scheduled(initialDelayString = "${baiyue.event-bridge.metrics.initial-delay:5000}",
|
||||
fixedDelayString = "${baiyue.event-bridge.metrics.sample-delay:30000}")
|
||||
public void sample() {
|
||||
final List<InboxMetricsSnapshot> snapshots;
|
||||
try {
|
||||
snapshots = inboxMapper.observabilitySnapshots();
|
||||
} catch (RuntimeException exception) {
|
||||
log.warn("Event Bridge Inbox 指标采样失败，保留上次成功快照", exception);
|
||||
return;
|
||||
}
|
||||
Map<MetricKey, Aggregate> aggregates = aggregate(snapshots);
|
||||
backlog.register(rows(aggregates, MetricValue.BACKLOG), true);
|
||||
oldestAge.register(rows(aggregates, MetricValue.OLDEST_AGE), true);
|
||||
dead.register(rows(aggregates, MetricValue.DEAD), true);
|
||||
}
|
||||
|
||||
private Map<MetricKey, Aggregate> aggregate(List<InboxMetricsSnapshot> snapshots) {
|
||||
Map<MetricKey, Aggregate> result = new LinkedHashMap<>();
|
||||
if (snapshots == null) {
|
||||
return result;
|
||||
}
|
||||
for (InboxMetricsSnapshot snapshot : snapshots) {
|
||||
MetricKey key = new MetricKey(EventBridgeMetrics.targetRegion(snapshot.getTargetRegion()),
|
||||
EventBridgeMetrics.eventType(snapshot.getEventType()));
|
||||
result.computeIfAbsent(key, ignored -> new Aggregate()).add(snapshot);
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
private List<MultiGauge.Row<?>> rows(Map<MetricKey, Aggregate> aggregates, MetricValue value) {
|
||||
List<MultiGauge.Row<?>> rows = new ArrayList<>();
|
||||
aggregates.forEach((key, aggregate) -> rows.add(MultiGauge.Row.of(
|
||||
Tags.of("target_region", key.targetRegion, "event_type", key.eventType),
|
||||
value.value(aggregate))));
|
||||
return rows;
|
||||
}
|
||||
|
||||
private record MetricKey(String targetRegion, String eventType) { }
|
||||
|
||||
private static final class Aggregate {
|
||||
private long backlog;
|
||||
private double oldestAge;
|
||||
private long dead;
|
||||
|
||||
void add(InboxMetricsSnapshot snapshot) {
|
||||
backlog += snapshot.getBacklogCount();
|
||||
oldestAge = Math.max(oldestAge, snapshot.getOldestPendingAgeSeconds());
|
||||
dead += snapshot.getDeadCount();
|
||||
}
|
||||
}
|
||||
|
||||
private enum MetricValue {
|
||||
BACKLOG {
|
||||
@Override Number value(Aggregate aggregate) { return aggregate.backlog; }
|
||||
},
|
||||
OLDEST_AGE {
|
||||
@Override Number value(Aggregate aggregate) { return aggregate.oldestAge; }
|
||||
},
|
||||
DEAD {
|
||||
@Override Number value(Aggregate aggregate) { return aggregate.dead; }
|
||||
};
|
||||
|
||||
abstract Number value(Aggregate aggregate);
|
||||
}
|
||||
}
|
||||
+121
@@ -0,0 +1,121 @@
|
||||
package com.baiyue.eventbridge.metrics;
|
||||
|
||||
import com.baiyue.common.event.EventEnvelope;
|
||||
import com.baiyue.common.event.EventTypes;
|
||||
import com.baiyue.eventbridge.domain.BridgeInbox;
|
||||
import io.micrometer.core.instrument.MeterRegistry;
|
||||
import io.micrometer.core.instrument.Timer;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.time.Clock;
|
||||
import java.time.Duration;
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.Locale;
|
||||
import java.util.Set;
|
||||
|
||||
/** Low-cardinality transport metrics for the cross-region event bridge. */
|
||||
@Component
|
||||
@Slf4j
|
||||
public class EventBridgeMetrics {
|
||||
public static final String RECEIVE_METRIC = "baiyue.event.bridge.receive";
|
||||
public static final String TRANSPORT_METRIC = "baiyue.event.bridge.transport";
|
||||
public static final String DELIVERY_LATENCY_METRIC = "baiyue.event.bridge.delivery.latency";
|
||||
|
||||
private static final Set<String> KNOWN_EVENT_TYPES = Set.of(
|
||||
EventTypes.ORDER_APPROVED,
|
||||
EventTypes.ORDER_CANCELLED,
|
||||
EventTypes.FIRST_LEG_SHIPMENT_DEPARTED,
|
||||
EventTypes.FIRST_LEG_LOADING_PLAN_COMPLETED,
|
||||
EventTypes.FIRST_LEG_CONTAINER_SHIPPING_NODE_RECORDED,
|
||||
EventTypes.OVERSEAS_CONTAINER_UNLOADED,
|
||||
EventTypes.IAM_USER_STATUS_CHANGED,
|
||||
EventTypes.IAM_TENANT_STATUS_CHANGED,
|
||||
EventTypes.IAM_PERMISSION_VERSION_CHANGED,
|
||||
EventTypes.LAST_MILE_HANDOFF_ACCEPTED,
|
||||
EventTypes.LAST_MILE_HANDOFF_REJECTED,
|
||||
EventTypes.LAST_MILE_MILESTONE_CHANGED,
|
||||
EventTypes.LAST_MILE_COMPLETED,
|
||||
EventTypes.TRACKING_MILESTONE_PROJECTED);
|
||||
|
||||
private final MeterRegistry registry;
|
||||
private final Clock clock;
|
||||
|
||||
public EventBridgeMetrics(MeterRegistry registry) {
|
||||
this(registry, Clock.systemDefaultZone());
|
||||
}
|
||||
|
||||
EventBridgeMetrics(MeterRegistry registry, Clock clock) {
|
||||
this.registry = registry;
|
||||
this.clock = clock;
|
||||
}
|
||||
|
||||
public void recordReceive(EventEnvelope envelope, ReceiveResult result) {
|
||||
try {
|
||||
registry.counter(RECEIVE_METRIC,
|
||||
"target_region", targetRegion(envelope.targetRegion().name()),
|
||||
"event_type", eventType(envelope.eventType()),
|
||||
"result", result.name()).increment();
|
||||
} catch (RuntimeException exception) {
|
||||
log.warn("Event Bridge 接收指标记录失败，不影响事件事务", exception);
|
||||
}
|
||||
}
|
||||
|
||||
public void recordTransport(BridgeInbox event, TransportResult result) {
|
||||
try {
|
||||
String targetRegion = targetRegion(event.getTargetRegion());
|
||||
String eventType = eventType(event.getEventType());
|
||||
registry.counter(TRANSPORT_METRIC,
|
||||
"target_region", targetRegion,
|
||||
"event_type", eventType,
|
||||
"result", result.name()).increment();
|
||||
if (result.acknowledged() && event.getReceivedAt() != null) {
|
||||
Duration latency = Duration.between(event.getReceivedAt(), LocalDateTime.now(clock));
|
||||
if (latency.isNegative()) {
|
||||
latency = Duration.ZERO;
|
||||
}
|
||||
Timer.builder(DELIVERY_LATENCY_METRIC)
|
||||
.description("Time from Event Bridge receipt to downstream acknowledgement")
|
||||
.serviceLevelObjectives(Duration.ofSeconds(60), Duration.ofMinutes(15))
|
||||
.tags("target_region", targetRegion, "event_type", eventType)
|
||||
.register(registry).record(latency);
|
||||
}
|
||||
} catch (RuntimeException exception) {
|
||||
log.warn("Event Bridge 传输指标记录失败，不影响可靠性状态机", exception);
|
||||
}
|
||||
}
|
||||
|
||||
public static String targetRegion(String value) {
|
||||
String normalized = value == null ? "" : value.trim().toUpperCase(Locale.ROOT);
|
||||
return "CN".equals(normalized) || "OVERSEAS".equals(normalized) ? normalized : "OTHER";
|
||||
}
|
||||
|
||||
public static String eventType(String value) {
|
||||
return value != null && KNOWN_EVENT_TYPES.contains(value) ? value : "OTHER";
|
||||
}
|
||||
|
||||
public enum ReceiveResult {
|
||||
RECEIVED,
|
||||
DUPLICATE,
|
||||
CONFLICT
|
||||
}
|
||||
|
||||
public enum TransportResult {
|
||||
DELIVERED(true),
|
||||
RETRY(false),
|
||||
DEAD(false),
|
||||
ACKNOWLEDGED_LEASE_LOST(true),
|
||||
FAILED_LEASE_LOST(false),
|
||||
MANUAL_REPLAY(false);
|
||||
|
||||
private final boolean acknowledged;
|
||||
|
||||
TransportResult(boolean acknowledged) {
|
||||
this.acknowledged = acknowledged;
|
||||
}
|
||||
|
||||
boolean acknowledged() {
|
||||
return acknowledged;
|
||||
}
|
||||
}
|
||||
}
|
||||
+50
@@ -0,0 +1,50 @@
|
||||
package com.baiyue.eventbridge.metrics;
|
||||
|
||||
/** Result row of the bounded, scheduled Inbox observability aggregation. */
|
||||
public class InboxMetricsSnapshot {
|
||||
private String targetRegion;
|
||||
private String eventType;
|
||||
private long backlogCount;
|
||||
private double oldestPendingAgeSeconds;
|
||||
private long deadCount;
|
||||
|
||||
public String getTargetRegion() {
|
||||
return targetRegion;
|
||||
}
|
||||
|
||||
public void setTargetRegion(String targetRegion) {
|
||||
this.targetRegion = targetRegion;
|
||||
}
|
||||
|
||||
public String getEventType() {
|
||||
return eventType;
|
||||
}
|
||||
|
||||
public void setEventType(String eventType) {
|
||||
this.eventType = eventType;
|
||||
}
|
||||
|
||||
public long getBacklogCount() {
|
||||
return backlogCount;
|
||||
}
|
||||
|
||||
public void setBacklogCount(long backlogCount) {
|
||||
this.backlogCount = backlogCount;
|
||||
}
|
||||
|
||||
public double getOldestPendingAgeSeconds() {
|
||||
return oldestPendingAgeSeconds;
|
||||
}
|
||||
|
||||
public void setOldestPendingAgeSeconds(double oldestPendingAgeSeconds) {
|
||||
this.oldestPendingAgeSeconds = oldestPendingAgeSeconds;
|
||||
}
|
||||
|
||||
public long getDeadCount() {
|
||||
return deadCount;
|
||||
}
|
||||
|
||||
public void setDeadCount(long deadCount) {
|
||||
this.deadCount = deadCount;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user