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 snapshots; try { snapshots = inboxMapper.observabilitySnapshots(); } catch (RuntimeException exception) { log.warn("Event Bridge Inbox 指标采样失败,保留上次成功快照", exception); return; } Map 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 aggregate(List snapshots) { Map 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> rows(Map aggregates, MetricValue value) { List> 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); } }