SpringBoot告警闭环之Prometheus+AlertManager详解

├── b-0001
│   ├── chunks       # 原始数据点
│   ├── index        # 倒排索引
│   └── meta.json    # 元数据

public List<Long> selectSeries(Map<String, Map<String, List<Long>>> postings, 
                              List<LabelMatcher> matchers) {
    // 初始化结果集
    Set<Long> result = new HashSet<>();
    boolean firstMatcher = true;
    for (LabelMatcher matcher : matchers) {
        Set<Long> matches = new HashSet<>();
        // 遍历标签名称
        for (String name : postings.keySet()) {
            if (!matcher.matchesName(name)) {
                continue;
            }
            // 遍历标签值
            Map<String, List<Long>> values = postings.get(name);
            for (String value : values.keySet()) {
                if (matcher.matchesValue(value)) {
                    matches.addAll(values.get(value));
                }
            }
        }
        // 应用集合操作
        if (firstMatcher) {
            result.addAll(matches);
            firstMatcher = false;
        } else {
            result.retainAll(matches);
        }
        if (result.isEmpty()) {
            break;
        }
    }
    return new ArrayList<>(result);
}

// 订单支付延迟监控完整实现
@Service
public class PaymentMonitoringService {
    private final Timer paymentProcessingTimer;
    private final Counter paymentSuccessCounter;
    private final Counter paymentFailureCounter;
    private final DistributionSummary paymentAmountSummary;
    // 支付延迟百分位数配置
    private static final double[] PERCENTILES = {0.5, 0.75, 0.95, 0.99};
    // SLO配置（单位：毫秒）
    private static final Duration[] SLO_BOUNDARIES = {
        Duration.ofMillis(100),
        Duration.ofMillis(500),
        Duration.ofMillis(1000)
    };
    @Autowired
    public PaymentMonitoringService(MeterRegistry registry) {
        // 初始化支付处理计时器
        paymentProcessingTimer = Timer.builder("payment.process.time")
            .description("支付处理耗时分布")
            .publishPercentiles(PERCENTILES)  // 发布百分位数指标
            .publishPercentileHistogram(true) // 启用百分位直方图
            .serviceLevelObjectives(SLO_BOUNDARIES) // 设置SLO边界
            .register(registry);
        // 初始化支付成功计数器
        paymentSuccessCounter = Counter.builder("payment.success.count")
            .description("成功支付次数")
            .tag("payment_method", "all") // 按支付方式打标
            .register(registry);
        // 初始化支付失败计数器
        paymentFailureCounter = Counter.builder("payment.failure.count")
            .description("失败支付次数")
            .tag("error_type", "all") // 按错误类型打标
            .register(registry);
        // 初始化支付金额分布摘要
        paymentAmountSummary = DistributionSummary.builder("payment.amount.summary")
            .description("支付金额分布")
            .baseUnit("USD") // 设置基准单位
            .register(registry);
    }
    /**
     * 记录支付处理时间
     * @param paymentProcess 支付处理逻辑
     * @param amount 支付金额
     * @return 支付处理结果
     */
    public <T> T recordPaymentExecution(Supplier<T> paymentProcess, BigDecimal amount) {
        // 记录支付开始时间
        long startTime = System.nanoTime();
        try {
            // 执行支付处理
            T result = paymentProcess.get();
            // 记录成功指标
            paymentSuccessCounter.increment();
            paymentAmountSummary.record(amount.doubleValue());
            return result;
        } catch (Exception e) {
            // 记录失败指标
            paymentFailureCounter.increment();
            throw e;
        } finally {
            // 记录处理耗时
            long duration = System.nanoTime() - startTime;
            paymentProcessingTimer.record(duration, TimeUnit.NANOSECONDS);
        }
    }
    /**
     * 获取当前支付性能指标快照
     * @return 指标快照信息
     */
    public PaymentMetricsSnapshot getCurrentMetricsSnapshot() {
        return new PaymentMetricsSnapshot(
            paymentProcessingTimer.count(),
            paymentProcessingTimer.totalTime(TimeUnit.MILLISECONDS),
            paymentSuccessCounter.count(),
            paymentFailureCounter.count(),
            paymentAmountSummary.totalAmount(),
            paymentAmountSummary.count()
        );
    }
    // 指标快照数据传输对象
    @Data
    @AllArgsConstructor
    public static class PaymentMetricsSnapshot {
        private long totalProcessedCount;
        private double totalProcessingTimeMs;
        private long successCount;
        private long failureCount;
        private double totalAmountProcessed;
        private long totalAmountRecords;
        public double getSuccessRate() {
            return totalProcessedCount > 0 ? 
                (double) successCount / totalProcessedCount * 100 : 0;
        }
        public double getAverageProcessingTime() {
            return totalProcessedCount > 0 ? 
                totalProcessingTimeMs / totalProcessedCount : 0;
        }
    }
}

public class AlertInhibitionEngine {
    private final List<InhibitionRule> rules;
    private final AlertStore alertStore;
    // 抑制规则内部类
    @Data
    @AllArgsConstructor
    public static class InhibitionRule {
        private Map<String, String> sourceMatchers;
        private Map<String, String> targetMatchers;
        private List<String> equalLabels;
    }
    /**
     * 检查告警是否被抑制
     */
    public boolean isInhibited(Alert alert) {
        for (InhibitionRule rule : rules) {
            if (matchesRule(alert, rule)) {
                return true;
            }
        }
        return false;
    }
    private boolean matchesRule(Alert alert, InhibitionRule rule) {
        // 检查是否匹配目标条件
        if (!matchesLabels(alert.getLabels(), rule.getTargetMatchers())) {
            return false;
        }
        // 查找匹配源条件的活跃告警
        return alertStore.getActiveAlerts().stream()
            .filter(sourceAlert -> matchesLabels(sourceAlert.getLabels(), rule.getSourceMatchers()))
            .anyMatch(sourceAlert -> hasEqualLabels(sourceAlert, alert, rule.getEqualLabels()));
    }
    private boolean matchesLabels(Map<String, String> alertLabels, 
                                 Map<String, String> matchers) {
        return matchers.entrySet().stream()
            .allMatch(entry -> alertLabels.getOrDefault(entry.getKey(), "")
                .matches(entry.getValue()));
    }
    private boolean hasEqualLabels(Alert source, Alert target, List<String> labelKeys) {
        return labelKeys.stream()
            .allMatch(key -> Objects.equals(
                source.getLabels().get(key),
                target.getLabels().get(key)));
    }
}

public class AlertThrottlingManager {
    private final DistributedCache cache;
    private final long initialInterval; // 初始间隔（毫秒）
    private final long maxInterval;     // 最大间隔（毫秒）
    public boolean shouldNotify(String alertKey, int alertCount) {
        long calculatedInterval = initialInterval * (long) Math.pow(2, alertCount - 1);
        long waitTime = Math.min(calculatedInterval, maxInterval);
        // 使用分布式锁确保集群环境下的正确性
        return cache.acquireLock(alertKey, waitTime);
    }
    // 释放通知锁，允许下次通知
    public void releaseNotification(String alertKey) {
        cache.releaseLock(alertKey);
    }
}

// VictoriaMetrics Java客户端配置
@Configuration
public class VictoriaMetricsConfig {
    @Value("${victoriametrics.host:localhost}")
    private String vmHost;
    @Value("${victoriametrics.port:8428}")
    private int vmPort;
    @Bean
    public MeterRegistry meterRegistry() {
        VictoriaMetricsConfig config = new VictoriaMetricsConfig() {
            @Override
            public String uri() {
                return "http://" + vmHost + ":" + vmPort + "/api/v1/import/prometheus";
            }
            @Override
            public Duration step() {
                return Duration.ofSeconds(15);
            }
        };
        return new VictoriaMeterRegistry(config, Clock.SYSTEM);
    }
    // 自定义指标推送器
    @Bean
    public VictoriaMetricsPusher metricsPusher(MeterRegistry meterRegistry) {
        VictoriaMetricsPusher pusher = new VictoriaMetricsPusher(config, meterRegistry);
        pusher.start();
        return pusher;
    }
}

// 查询优化策略实现
public class QueryOptimizer {
    // 查询缓存实现
    private final Cache<String, QueryResult> queryCache = 
        Caffeine.newBuilder()
            .maximumSize(10000)
            .expireAfterWrite(5, TimeUnit.MINUTES)
            .build();
    // 并行查询执行器
    public QueryResult executeParallelQuery(String query, long start, long end) {
        // 拆分时间范围为多个区间
        List<TimeRange> ranges = splitTimeRange(start, end, 24 * 3600 * 1000); // 按天拆分
        // 并行执行子查询
        List<CompletableFuture<QueryResult>> futures = ranges.stream()
            .map(range -> CompletableFuture.supplyAsync(() -> 
                executeSubQuery(query, range.start, range.end), queryExecutor))
            .collect(Collectors.toList());
        // 合并结果
        return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
            .thenApply(v -> futures.stream()
                .map(CompletableFuture::join)
                .reduce(QueryResult::merge)
                .orElseThrow(() -> new RuntimeException("Merge failed")))
            .join();
    }
    // 查询重写优化
    public String rewriteQuery(String originalQuery) {
        // 应用启发式重写规则
        String rewritten = originalQuery;
        // 规则1: 将高基数标签过滤提前
        rewritten = rewritten.replaceAll(
            "sum\\(([^}]+})\\) by \\(([^)]+)\\)", 
            "sum($1) by ($2)"
        );
        // 规则2: 避免不必要的排序
        if (rewritten.contains("sort(") && !rewritten.contains("topk") && !rewritten.contains("bottomk")) {
            rewritten = rewritten.replace("sort(", "(");
        }
        return rewritten;
    }
}

// 自动修复规则引擎
@Service
public class AutoRemediationEngine {
    private final Map<String, RemediationRule> rules;
    private final ScriptExecutor scriptExecutor;
    private final AlertManagerClient alertManagerClient;
    // 修复规则定义
    @Data
    public static class RemediationRule {
        private String name;
        private String alertPattern;
        private String conditionScript;
        private String remediationScript;
        private int maxAttempts;
        private Duration cooldownPeriod;
    }
    /**
     * 处理传入告警
     */
    @Async
    public void processAlert(Alert alert) {
        rules.values().stream()
            .filter(rule -> matchesPattern(alert, rule.getAlertPattern()))
            .filter(rule -> evaluateCondition(alert, rule.getConditionScript()))
            .forEach(rule -> executeRemediation(alert, rule));
    }
    /**
     * 执行修复操作
     */
    private void executeRemediation(Alert alert, RemediationRule rule) {
        try {
            // 检查冷却期
            if (isInCooldown(alert, rule)) {
                return;
            }
            // 执行修复脚本
            RemediationResult result = scriptExecutor.execute(
                rule.getRemediationScript(), 
                createExecutionContext(alert)
            );
            if (result.isSuccess()) {
                // 修复成功，静默告警
                alertManagerClient.silenceAlert(alert, rule.getCooldownPeriod());
                log.info("Remediation successful for alert: {}", alert.getLabels());
            } else {
                // 修复失败，记录重试
                recordRemediationAttempt(alert, rule, result);
            }
        } catch (Exception e) {
            log.error("Remediation failed for alert: {}", alert.getLabels(), e);
        }
    }
    // 创建脚本执行上下文
    private Map<String, Object> createExecutionContext(Alert alert) {
        Map<String, Object> context = new HashMap<>();
        context.put("alert", alert);
        context.put("labels", alert.getLabels());
        context.put("annotations", alert.getAnnotations());
        context.put("startTime", alert.getStartsAt());
        context.put("currentTime", Instant.now());
        return context;
    }
}

# 内存溢出修复模板
- name: memory_overflow_remediation
  description: Java应用内存溢出自动处理
  conditions:
    - alertname: "JVMOutOfMemoryError"
    - severity: "critical"
  steps:
    - action: "thread_dump"
      target: "application"
      args:
        output: "/tmp/threaddump_{{timestamp}}"
    - action: "heap_dump"
      target: "application"  
      args:
        output: "/tmp/heapdump_{{timestamp}}"
    - action: "restart"
      target: "application"
      args:
        delay: "30s"
        gracePeriod: "60s"
  coolingPeriod: "1h"
# 数据库连接池耗尽修复模板  
- name: db_connection_pool_exhausted
  description: 数据库连接池耗尽自动扩容
  conditions:
    - alertname: "DBConnectionPoolExhausted"
    - severity: "warning"
  steps:
    - action: "scale_connection_pool"
      target: "database"
      args:
        maxSize: "{{currentSize * 1.5}}"
        increment: "10"
    - action: "execute_sql"
      target: "database"
      args:
        sql: "KILL QUERY PROCESSLIST WHERE Time > 300 AND State = 'Sleep'"
  coolingPeriod: "30m"

// 高可用监控代理
public class HighAvailabilityAgent {
    private final List<PrometheusClient> prometheusClients;
    private final HealthChecker healthChecker;
    private final LoadBalancer loadBalancer;
    /**
     * 高可用查询方法
     */
    public QueryResult haQuery(String query, long timeoutMs) {
        List<PrometheusClient> healthyClients = prometheusClients.stream()
            .filter(healthChecker::isHealthy)
            .collect(Collectors.toList());
        if (healthyClients.isEmpty()) {
            throw new MonitoringException("No healthy Prometheus instances available");
        }
        // 使用负载均衡策略选择实例
        PrometheusClient selectedClient = loadBalancer.select(healthyClients);
        try {
            return selectedClient.executeQuery(query, timeoutMs);
        } catch (Exception e) {
            // 标记实例不健康并重试
            healthChecker.markUnhealthy(selectedClient);
            return haQuery(query, timeoutMs); // 递归重试
        }
    }
    // 负载均衡策略
    private PrometheusClient selectClient(List<PrometheusClient> clients) {
        // 基于最少活跃请求的负载均衡
        return clients.stream()
            .min(Comparator.comparingInt(PrometheusClient::getActiveRequests))
            .orElseThrow(() -> new IllegalStateException("No clients available"));
    }
}

# 监控系统自监控规则
groups:
- name: monitoring-system
  rules:
  - alert: PrometheusScrapeFailing
    expr: up{job=~"prometheus.*"} == 0
    for: 5m
    labels:
      severity: critical
      component: monitoring
    annotations:
      summary: "Prometheus实例抓取失败"
      description: "Prometheus实例 {{ $labels.instance }} 无法抓取指标"
  - alert: AlertManagerNotConnected
    expr: alertmanager_cluster_health_score != 1
    for: 2m
    labels:
      severity: warning
      component: alerting
    annotations:
      summary: "AlertManager集群健康状态异常"
      description: "AlertManager集群健康得分为 {{ $value }}"
  - alert: VictoriaMetricsHighMemoryUsage
    expr: process_resident_memory_bytes{job=~"victoriametrics.*"} / 
          machine_memory_bytes > 0.8
    for: 10m
    labels:
      severity: warning
      component: storage
    annotations:
      summary: "VictoriaMetrics内存使用率过高"
      description: "VictoriaMetrics实例 {{ $labels.instance }} 内存使用率超过80%"

// 智能异常检测服务
@Service
public class AnomalyDetectionService {
    private final TimeSeriesAnalyzer timeSeriesAnalyzer;
    private final ModelTrainer modelTrainer;
    private final Map<String, DetectionModel> models;
    /**
     * 训练检测模型
     */
    public void trainModel(String metricName, List<DataPoint> historicalData) {
        DetectionModel model = modelTrainer.train(historicalData, createTrainingConfig());
        models.put(metricName, model);
        // 保存模型到持久化存储
        saveModel(metricName, model);
    }
    /**
     * 检测异常点
     */
    public AnomalyDetectionResult detectAnomalies(String metricName, List<DataPoint> currentData) {
        DetectionModel model = models.get(metricName);
        if (model == null) {
            throw new ModelNotTrainedException(metricName);
        }
        return model.detect(currentData);
    }
    // 异常检测配置
    private TrainingConfig createTrainingConfig() {
        return TrainingConfig.builder()
            .algorithm(Algorithm.ISOLATION_FOREST)
            .sensitivity(0.8)
            .windowSize(1440) // 24小时数据点
            .seasonality(Seasonality.DAILY)
            .build();
    }
}

// 容量预测服务
@Service 
public class CapacityForecastingService {
    private final TimeSeriesForecaster forecaster;
    private final ResourceUsageCollector usageCollector;
    /**
     * 生成容量预测报告
     */
    public CapacityForecast generateForecast(String resourceType, int forecastDays) {
        // 收集历史使用数据
        List<UsageData> historicalData = usageCollector.collectHistoricalData(
            resourceType, 
            LocalDate.now().minusDays(90), // 90天历史数据
            LocalDate.now()
        );
        // 训练预测模型
        ForecastModel model = forecaster.trainModel(historicalData);
        // 生成预测
        return model.forecast(forecastDays);
    }
    /**
     * 获取扩容建议
     */
    public ScalingRecommendation getScalingRecommendation(String resourceType) {
        CapacityForecast forecast = generateForecast(resourceType, 30);
        double currentCapacity = getCurrentCapacity(resourceType);
        double predictedPeak = forecast.getPeakUsage();
        double safetyMargin = 0.3; // 30%安全边际
        if (predictedPeak > currentCapacity * (1 - safetyMargin)) {
            return new ScalingRecommendation(
                resourceType,
                ScalingDirection.SCALE_UP,
                calculateRequiredCapacity(predictedPeak, safetyMargin),
                forecast.getPeakDate()
            );
        }
        return ScalingRecommendation.noAction(resourceType);
    }
}