15.4 与监控报警系统集成: 聚合所有任务报警

老马啸西风2025/9/6大约 16 分钟

在分布式调度平台的运维管理中，监控和报警系统是确保系统稳定运行、及时发现和处理问题的关键基础设施。通过与主流监控报警系统的深度集成，调度平台能够将任务执行过程中的各种异常情况、性能指标、资源使用等信息统一聚合和展示，形成完整的监控告警闭环。这种集成不仅提升了问题发现的及时性和准确性，还为运维人员提供了统一的告警管理和处理入口，显著提高了运维效率和系统可靠性。本文将深入探讨分布式调度平台与监控报警系统集成的核心理念、技术实现以及最佳实践。

监控报警集成的核心价值

理解监控报警系统集成在分布式调度平台中的重要意义是构建高质量运维体系的基础。

集成挑战分析

在分布式调度平台中实施监控报警系统集成面临诸多挑战：

数据整合挑战：

多源数据：来自不同组件和系统的监控数据
格式差异：不同监控系统采用的数据格式和协议
实时性要求：监控数据需要实时收集和处理
数据质量：确保监控数据的准确性和完整性

告警管理挑战：

告警风暴：大量告警同时触发造成信息过载
误报处理：减少和处理误报提高告警质量
根因定位：快速定位问题的根本原因
处理跟踪：跟踪告警的处理状态和进度

系统兼容挑战：

协议适配：适配不同监控系统的通信协议
API集成：集成各种监控系统的API接口
数据转换：转换不同系统的数据格式
版本兼容：处理不同版本间的兼容性问题

性能优化挑战：

采集性能：高效采集大量监控数据
传输效率：优化监控数据的传输效率
存储优化：优化监控数据的存储和查询
查询性能：提升监控数据的查询性能

核心价值体现

监控报警集成带来的核心价值：

统一监控：

视图统一：提供统一的监控视图和仪表板
数据整合：整合各组件的监控数据
指标标准化：建立标准化的监控指标体系
历史追溯：提供完整的监控历史数据

智能告警：

告警聚合：聚合相关告警减少信息过载
智能分析：基于规则和机器学习分析告警
根因定位：快速定位问题的根本原因
自动处理：自动处理简单和常见的告警

运维提效：

快速响应：快速发现和响应系统异常
减少人工：减少人工监控和告警处理
决策支持：为运维决策提供数据支持
成本优化：优化运维资源配置和成本

质量保障：

问题预防：通过监控预警预防问题发生
质量评估：评估系统运行质量和稳定性
持续改进：基于监控数据持续改进系统
合规审计：满足运维管理的合规要求

集成架构设计

设计合理的监控报警系统集成架构。

整体架构

构建端到端的集成架构：

数据流向：

指标采集：从调度平台各组件采集监控指标
数据传输：将监控数据传输到监控系统
数据存储：在监控系统中存储和索引数据
告警处理：基于规则处理和生成告警
通知分发：将告警通知分发给相关人员
处理跟踪：跟踪告警的处理状态和结果

核心组件：

指标采集器：负责采集调度平台的监控指标
数据传输器：负责将数据传输到监控系统
监控存储：存储监控数据和告警信息
告警引擎：处理告警规则和生成告警
通知中心：分发告警通知给相关人员
展示界面：提供监控数据和告警的展示界面

集成模式：

Push模式：调度平台主动推送监控数据
Pull模式：监控系统主动拉取监控数据
混合模式：结合Push和Pull两种模式
事件驱动：基于事件触发监控数据收集

数据模型设计

设计统一的监控数据模型：

指标定义：

metrics:
  - name: "task_execution_duration_seconds"
    type: "histogram"
    description: "任务执行耗时分布"
    labels: ["task_type", "status", "worker_node"]
    
  - name: "task_execution_success_rate"
    type: "gauge"
    description: "任务执行成功率"
    labels: ["task_type", "time_window"]
    
  - name: "scheduler_queue_size"
    type: "gauge"
    description: "调度队列大小"
    labels: ["queue_type"]
    
  - name: "worker_node_cpu_usage"
    type: "gauge"
    description: "Worker节点CPU使用率"
    labels: ["node_name", "instance"]
    
  - name: "task_failure_count"
    type: "counter"
    description: "任务失败次数"
    labels: ["task_type", "error_type"]

告警规则：

alertRules:
  - name: "HighTaskFailureRate"
    expr: "rate(task_failure_count[5m]) > 0.1"
    for: "2m"
    labels:
      severity: "warning"
    annotations:
      summary: "任务失败率过高"
      description: "过去5分钟任务失败率超过10%"
      
  - name: "SchedulerQueueBacklog"
    expr: "scheduler_queue_size > 1000"
    for: "5m"
    labels:
      severity: "critical"
    annotations:
      summary: "调度队列积压严重"
      description: "调度队列积压超过1000个任务"
      
  - name: "WorkerNodeDown"
    expr: "up{job='scheduler-worker'} == 0"
    for: "1m"
    labels:
      severity: "critical"
    annotations:
      summary: "Worker节点宕机"
      description: "Worker节点不可达"

事件结构：

{
  "eventId": "evt-20250906-123456",
  "eventType": "task.execution.failed",
  "timestamp": "2025-09-06T10:30:00Z",
  "source": "scheduler-worker-01",
  "severity": "ERROR",
  "payload": {
    "taskId": "task-12345",
    "taskName": "data-processing-job",
    "executionId": "exec-67890",
    "errorType": "TimeoutException",
    "errorMessage": "任务执行超时",
    "duration": 3600,
    "workerNode": "worker-01"
  },
  "metadata": {
    "traceId": "trace-abcdef-123456",
    "version": "1.0"
  }
}

采集策略

设计高效的监控数据采集策略：

采集频率：

高频采集：关键指标每秒采集一次
中频采集：一般指标每分钟采集一次
低频采集：统计指标每小时采集一次
事件触发：基于事件触发数据采集

采集方式：

主动上报：组件主动上报监控数据
被动拉取：监控系统定期拉取监控数据
混合采集：结合主动和被动两种方式
批量处理：批量处理和传输监控数据

数据处理：

实时处理：实时处理和分析监控数据
聚合计算：对原始数据进行聚合计算
异常检测：检测数据中的异常情况
压缩存储：压缩存储历史监控数据

Prometheus集成实现

实现与Prometheus监控系统的集成。

Prometheus架构理解

理解Prometheus的核心架构和特性：

核心组件：

Prometheus Server：主服务，负责数据采集、存储和查询
Client Libraries：客户端库，用于在应用中暴露指标
Pushgateway：用于推送短期任务的指标
Alertmanager：处理告警通知和分发
Service Discovery：服务发现机制

关键特性：

多维数据模型：基于时间序列的多维数据模型
Pull模式：主要采用Pull模式采集数据
强大的查询语言：PromQL用于数据查询和分析
告警规则：基于规则的告警处理机制

指标暴露

在调度平台中暴露监控指标：

Java应用集成：

@Component
public class SchedulerMetrics {
    
    private final Counter taskExecutionCounter;
    private final Timer taskExecutionTimer;
    private final Gauge queueSizeGauge;
    private final Counter taskFailureCounter;
    
    public SchedulerMetrics(MeterRegistry meterRegistry) {
        this.taskExecutionCounter = Counter.builder("task_execution_total")
                .description("Total number of task executions")
                .tag("type", "scheduler")
                .register(meterRegistry);
                
        this.taskExecutionTimer = Timer.builder("task_execution_duration_seconds")
                .description("Task execution duration")
                .tag("type", "scheduler")
                .publishPercentileHistogram()
                .register(meterRegistry);
                
        this.queueSizeGauge = Gauge.builder("scheduler_queue_size")
                .description("Current scheduler queue size")
                .tag("type", "task")
                .register(meterRegistry, this, SchedulerMetrics::getQueueSize);
                
        this.taskFailureCounter = Counter.builder("task_failure_total")
                .description("Total number of task failures")
                .tag("type", "scheduler")
                .register(meterRegistry);
    }
    
    public void recordTaskExecution(String taskType, String status, long durationMs) {
        taskExecutionCounter.tag("task_type", taskType).tag("status", status).increment();
        
        Timer.Sample sample = Timer.start();
        sample.stop(Timer.builder("task_execution_duration_seconds")
                .tag("task_type", taskType)
                .tag("status", status)
                .register(Metrics.globalRegistry));
    }
    
    public void recordTaskFailure(String taskType, String errorType) {
        taskFailureCounter.tag("task_type", taskType).tag("error_type", errorType).increment();
    }
    
    public void setQueueSize(int size) {
        // 更新队列大小指标
    }
    
    private static double getQueueSize(SchedulerMetrics metrics) {
        // 获取当前队列大小
        return 0.0;
    }
}

Spring Boot集成：

@Configuration
@EnablePrometheusEndpoint
public class PrometheusConfig {
    
    @Bean
    public CollectorRegistry collectorRegistry() {
        return new CollectorRegistry();
    }
    
    @Bean
    public ServletRegistrationBean prometheusEndpoint() {
        PrometheusScrapingServlet servlet = new PrometheusScrapingServlet(collectorRegistry());
        return new ServletRegistrationBean(servlet, "/prometheus");
    }
    
    @Bean
    public TimedAspect timedAspect(MeterRegistry meterRegistry) {
        return new TimedAspect(meterRegistry);
    }
}

Docker集成：

FROM openjdk:11-jre-slim

# 添加JMX Exporter
ADD https://repo1.maven.org/maven2/io/prometheus/jmx/jmx_prometheus_javaagent/0.16.1/jmx_prometheus_javaagent-0.16.1.jar /opt/jmx_exporter.jar
ADD jmx-exporter-config.yaml /opt/jmx-exporter-config.yaml

# 设置JVM参数
ENV JAVA_OPTS="-javaagent:/opt/jmx_exporter.jar=8080:/opt/jmx-exporter-config.yaml"

COPY target/scheduler-platform.jar app.jar
ENTRYPOINT ["sh", "-c", "java $JAVA_OPTS -jar /app.jar"]

配置管理

配置Prometheus监控系统：

Prometheus配置文件：

global:
  scrape_interval: 15s
  evaluation_interval: 15s
  
rule_files:
  - "scheduler_rules.yml"
  
alerting:
  alertmanagers:
    - static_configs:
        - targets:
            - alertmanager:9093

scrape_configs:
  - job_name: 'scheduler-master'
    static_configs:
      - targets: ['scheduler-master:8080']
    metrics_path: '/actuator/prometheus'
    
  - job_name: 'scheduler-worker'
    static_configs:
      - targets: ['scheduler-worker-1:8080', 'scheduler-worker-2:8080']
    metrics_path: '/actuator/prometheus'
    
  - job_name: 'scheduler-api'
    static_configs:
      - targets: ['scheduler-api:8080']
    metrics_path: '/actuator/prometheus'

告警规则配置：

groups:
- name: scheduler-alerts
  rules:
  - alert: HighTaskFailureRate
    expr: rate(task_failure_total[5m]) > 0.05
    for: 2m
    labels:
      severity: warning
    annotations:
      summary: "任务失败率过高"
      description: "过去5分钟任务失败率超过5%，当前值为 {{ $value }}"
      
  - alert: SchedulerQueueBacklog
    expr: scheduler_queue_size > 1000
    for: 5m
    labels:
      severity: critical
    annotations:
      summary: "调度队列积压严重"
      description: "调度队列积压超过1000个任务，当前值为 {{ $value }}"
      
  - alert: WorkerNodeDown
    expr: up{job=~"scheduler-worker"} == 0
    for: 1m
    labels:
      severity: critical
    annotations:
      summary: "Worker节点宕机"
      description: "Worker节点 {{ $labels.instance }} 不可达"
      
  - alert: HighCPUUsage
    expr: rate(process_cpu_seconds_total[1m]) > 0.8
    for: 2m
    labels:
      severity: warning
    annotations:
      summary: "CPU使用率过高"
      description: "CPU使用率超过80%，当前值为 {{ $value }}"

Grafana集成实现

实现与Grafana可视化平台的集成。

Grafana架构理解

理解Grafana的核心架构和特性：

核心概念：

DataSource：数据源，连接到监控数据库
Dashboard：仪表板，展示监控数据的可视化界面
Panel：面板，仪表板中的具体图表组件
Variable：变量，用于动态过滤和查询数据
Alert：告警，基于数据的告警规则

关键特性：

丰富的可视化：支持多种图表类型和可视化方式
灵活的查询：支持复杂的数据查询和分析
强大的告警：支持基于查询结果的告警功能
插件生态：丰富的插件扩展生态系统

仪表板设计

设计调度平台监控仪表板：

概览仪表板：

{
  "dashboard": {
    "title": "调度平台概览",
    "panels": [
      {
        "id": 1,
        "title": "任务执行统计",
        "type": "graph",
        "datasource": "Prometheus",
        "targets": [
          {
            "expr": "rate(task_execution_total[5m])",
            "legendFormat": "{{status}}"
          }
        ]
      },
      {
        "id": 2,
        "title": "任务执行成功率",
        "type": "gauge",
        "datasource": "Prometheus",
        "targets": [
          {
            "expr": "task_execution_success_rate",
            "legendFormat": "成功率"
          }
        ]
      },
      {
        "id": 3,
        "title": "调度队列积压",
        "type": "stat",
        "datasource": "Prometheus",
        "targets": [
          {
            "expr": "scheduler_queue_size",
            "legendFormat": "队列大小"
          }
        ]
      },
      {
        "id": 4,
        "title": "Worker节点状态",
        "type": "table",
        "datasource": "Prometheus",
        "targets": [
          {
            "expr": "up{job=~\"scheduler-worker\"}",
            "format": "table"
          }
        ]
      }
    ]
  }
}

详细仪表板：

{
  "dashboard": {
    "title": "任务执行详情",
    "templating": {
      "list": [
        {
          "name": "task_type",
          "type": "query",
          "datasource": "Prometheus",
          "label": "任务类型",
          "query": "label_values(task_execution_duration_seconds, task_type)"
        }
      ]
    },
    "panels": [
      {
        "id": 1,
        "title": "任务执行耗时分布",
        "type": "heatmap",
        "datasource": "Prometheus",
        "targets": [
          {
            "expr": "rate(task_execution_duration_seconds_bucket{task_type=\"$task_type\"}[5m])",
            "format": "heatmap",
            "legendFormat": "{{le}}"
          }
        ]
      },
      {
        "id": 2,
        "title": "任务执行趋势",
        "type": "graph",
        "datasource": "Prometheus",
        "targets": [
          {
            "expr": "rate(task_execution_total{task_type=\"$task_type\"}[5m])",
            "legendFormat": "{{status}}"
          }
        ]
      },
      {
        "id": 3,
        "title": "失败任务分析",
        "type": "table",
        "datasource": "Prometheus",
        "targets": [
          {
            "expr": "topk(10, task_failure_total{task_type=\"$task_type\"})",
            "format": "table"
          }
        ]
      }
    ]
  }
}

变量配置

配置动态查询变量：

变量定义：

{
  "templating": {
    "list": [
      {
        "name": "datasource",
        "type": "datasource",
        "label": "数据源",
        "query": "prometheus"
      },
      {
        "name": "job",
        "type": "query",
        "label": "服务",
        "query": "label_values(up, job)",
        "refresh": 1
      },
      {
        "name": "instance",
        "type": "query",
        "label": "实例",
        "query": "label_values(up{job=\"$job\"}, instance)",
        "refresh": 1,
        "hide": 0
      },
      {
        "name": "task_type",
        "type": "query",
        "label": "任务类型",
        "query": "label_values(task_execution_total, task_type)",
        "refresh": 1
      },
      {
        "name": "time_range",
        "type": "interval",
        "label": "时间范围",
        "options": [
          {"text": "最近5分钟", "value": "5m"},
          {"text": "最近1小时", "value": "1h"},
          {"text": "最近24小时", "value": "24h"}
        ],
        "current": {"text": "最近1小时", "value": "1h"}
      }
    ]
  }
}

变量使用：

{
  "panels": [
    {
      "title": "CPU使用率",
      "targets": [
        {
          "expr": "rate(process_cpu_seconds_total{job=\"$job\", instance=\"$instance\"}[$time_range])",
          "legendFormat": "CPU使用率"
        }
      ]
    },
    {
      "title": "内存使用",
      "targets": [
        {
          "expr": "process_resident_memory_bytes{job=\"$job\", instance=\"$instance\"}",
          "legendFormat": "内存使用"
        }
      ]
    }
  ]
}

告警管理实现

实现完善的告警管理体系。

告警规则设计

设计灵活的告警规则：

基础告警规则：

# 任务相关告警
- alert: TaskExecutionTimeout
  expr: task_execution_duration_seconds > 3600
  for: 1m
  labels:
    severity: warning
    category: task
  annotations:
    summary: "任务执行超时"
    description: "任务 {{ $labels.task_name }} 执行时间超过1小时"

- alert: HighTaskRetryRate
  expr: rate(task_retry_total[5m]) > 0.1
  for: 2m
  labels:
    severity: warning
    category: task
  annotations:
    summary: "任务重试率过高"
    description: "任务重试率超过10%，可能存在配置问题"

# 系统相关告警
- alert: MasterNodeDown
  expr: up{job="scheduler-master"} == 0
  for: 1m
  labels:
    severity: critical
    category: system
  annotations:
    summary: "调度主节点宕机"
    description: "调度主节点 {{ $labels.instance }} 不可达"

- alert: LowWorkerNodeCount
  expr: count(up{job="scheduler-worker"}) < 3
  for: 5m
  labels:
    severity: warning
    category: system
  annotations:
    summary: "Worker节点数量不足"
    description: "Worker节点数量少于3个，当前为 {{ $value }} 个"

智能告警规则：

# 基于机器学习的异常检测
- alert: AbnormalTaskExecutionPattern
  expr: >
    task_execution_duration_seconds > 
    (avg_over_time(task_execution_duration_seconds[1h]) * 2)
  for: 5m
  labels:
    severity: warning
    category: anomaly
  annotations:
    summary: "任务执行模式异常"
    description: "任务 {{ $labels.task_name }} 执行时间异常，超过历史平均值的2倍"

# 基于趋势的预测告警
- alert: IncreasingFailureRate
  expr: >
    deriv(rate(task_failure_total[10m])[30m:]) > 0.001
  for: 10m
  labels:
    severity: warning
    category: trend
  annotations:
    summary: "失败率呈上升趋势"
    description: "任务失败率呈上升趋势，需要关注"

告警处理流程

设计告警处理流程：

告警分组：

route:
  group_by: ['alertname', 'cluster']
  group_wait: 30s
  group_interval: 5m
  repeat_interval: 3h
  receiver: 'default-receiver'
  
  routes:
    - match:
        severity: critical
      receiver: 'critical-alerts'
      group_wait: 10s
      group_interval: 1m
      repeat_interval: 1h
      
    - match:
        category: task
      receiver: 'task-team'
      group_by: ['alertname', 'task_type']
      
    - match_re:
        service: ^(scheduler|worker)
      receiver: 'platform-team'

告警抑制：

inhibit_rules:
  - source_match:
      severity: 'critical'
      category: 'system'
    target_match:
      severity: 'warning'
      category: 'task'
    equal: ['cluster', 'job']
    
  - source_match:
      alertname: 'MasterNodeDown'
    target_match:
      alertname: 'WorkerNodeDown'
    equal: ['cluster']

通知配置

配置多样化的通知方式：

接收器配置：

receivers:
  - name: 'default-receiver'
    email_configs:
      - to: 'ops-team@example.com'
        send_resolved: true
        
  - name: 'critical-alerts'
    webhook_configs:
      - url: 'http://wechat-alert-gateway:8080/webhook'
        send_resolved: true
    email_configs:
      - to: 'oncall@example.com'
        send_resolved: true
        
  - name: 'task-team'
    slack_configs:
      - api_url: 'https://hooks.slack.com/services/XXX/YYY/ZZZ'
        channel: '#task-alerts'
        send_resolved: true
        
  - name: 'platform-team'
    pagerduty_configs:
      - service_key: 'XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX'
        send_resolved: true

自定义模板：

{{ define "__subject" }}[{{ .Status | toUpper }}{{ if eq .Status "firing" }}:{{ .Alerts.Firing | len }}{{ end }}] {{ .GroupLabels.SortedPairs.Values | join " " }} {{ if gt (len .CommonLabels) (len .GroupLabels) }}({{ with .CommonLabels.Remove .GroupLabels.Names }}{{ .Values | join " " }}{{ end }}){{ end }}{{ end }}

{{ define "__text_alert_list" }}{{ range . }}
Labels:
{{ range .Labels.SortedPairs }} - {{ .Name }} = {{ .Value }}
{{ end }}
Annotations:
{{ range .Annotations.SortedPairs }} - {{ .Name }} = {{ .Value }}
{{ end }}
Source: {{ .GeneratorURL }}
{{ end }}{{ end }}

{{ define "email.default.subject" }}{{ template "__subject" . }}{{ end }}

{{ define "email.default.html" }}
<!DOCTYPE html>
<html>
<head>
    <meta charset="UTF-8">
    <title>调度平台告警通知</title>
</head>
<body>
    <h2>{{ template "__subject" . }}</h2>
    <p><strong>告警时间:</strong> {{ .Alerts.Firing | len }} 个告警触发</p>
    <p><strong>恢复时间:</strong> {{ .Alerts.Resolved | len }} 个告警恢复</p>
    
    {{ if gt (len .Alerts.Firing) 0 }}
    <h3>触发的告警:</h3>
    <ul>
    {{ range .Alerts.Firing }}
        <li>
            <strong>{{ .Annotations.summary }}</strong><br>
            {{ .Annotations.description }}<br>
            <em>严重级别: {{ .Labels.severity }}</em>
        </li>
    {{ end }}
    </ul>
    {{ end }}
    
    {{ if gt (len .Alerts.Resolved) 0 }}
    <h3>恢复的告警:</h3>
    <ul>
    {{ range .Alerts.Resolved }}
        <li>
            <strong>{{ .Annotations.summary }}</strong><br>
            {{ .Annotations.description }}
        </li>
    {{ end }}
    </ul>
    {{ end }}
</body>
</html>
{{ end }}

监控数据聚合

实现监控数据的统一聚合。

数据聚合策略

设计数据聚合策略：

时间聚合：

@Service
public class MetricsAggregationService {
    
    @Autowired
    private MeterRegistry meterRegistry;
    
    // 每分钟聚合一次
    @Scheduled(fixedRate = 60000)
    public void aggregateTaskMetrics() {
        // 聚合任务执行指标
        aggregateTaskExecutionMetrics();
        
        // 聚合资源使用指标
        aggregateResourceUsageMetrics();
        
        // 聚合错误统计指标
        aggregateErrorMetrics();
    }
    
    private void aggregateTaskExecutionMetrics() {
        // 获取原始指标数据
        List<TaskExecutionRecord> records = taskExecutionRepository
            .findLastMinuteRecords();
            
        // 按任务类型聚合
        Map<String, TaskMetrics> aggregatedMetrics = records.stream()
            .collect(Collectors.groupingBy(
                TaskExecutionRecord::getTaskType,
                Collectors.collectingAndThen(
                    Collectors.toList(),
                    this::calculateTaskMetrics
                )
            ));
            
        // 更新聚合指标
        aggregatedMetrics.forEach((taskType, metrics) -> {
            Gauge.builder("task_execution_aggregated_success_rate")
                .tag("task_type", taskType)
                .register(meterRegistry, metrics, TaskMetrics::getSuccessRate);
                
            Gauge.builder("task_execution_aggregated_avg_duration")
                .tag("task_type", taskType)
                .register(meterRegistry, metrics, TaskMetrics::getAvgDuration);
        });
    }
    
    private TaskMetrics calculateTaskMetrics(List<TaskExecutionRecord> records) {
        long total = records.size();
        long success = records.stream()
            .filter(r -> "SUCCESS".equals(r.getStatus()))
            .count();
            
        double avgDuration = records.stream()
            .mapToLong(TaskExecutionRecord::getDuration)
            .average()
            .orElse(0.0);
            
        return new TaskMetrics(total, success, avgDuration);
    }
}

空间聚合：

@Component
public class SpatialMetricsAggregator {
    
    public AggregatedMetrics aggregateByCluster(List<NodeMetrics> nodeMetrics) {
        return nodeMetrics.stream()
            .collect(Collectors.groupingBy(
                NodeMetrics::getCluster,
                Collectors.collectingAndThen(
                    Collectors.toList(),
                    this::aggregateClusterMetrics
                )
            ));
    }
    
    public AggregatedMetrics aggregateByService(List<NodeMetrics> nodeMetrics) {
        return nodeMetrics.stream()
            .collect(Collectors.groupingBy(
                NodeMetrics::getService,
                Collectors.collectingAndThen(
                    Collectors.toList(),
                    this::aggregateServiceMetrics
                )
            ));
    }
    
    private AggregatedMetrics aggregateClusterMetrics(List<NodeMetrics> metrics) {
        double avgCpu = metrics.stream()
            .mapToDouble(NodeMetrics::getCpuUsage)
            .average()
            .orElse(0.0);
            
        double totalMemory = metrics.stream()
            .mapToDouble(NodeMetrics::getMemoryUsage)
            .sum();
            
        return new AggregatedMetrics(avgCpu, totalMemory, metrics.size());
    }
}

跨系统聚合

实现跨系统的监控数据聚合：

统一指标接口：

@RestController
@RequestMapping("/api/v1/metrics")
public class UnifiedMetricsController {
    
    @Autowired
    private MetricsAggregationService aggregationService;
    
    @GetMapping("/overview")
    public ResponseEntity<SystemOverviewMetrics> getSystemOverview() {
        SystemOverviewMetrics overview = aggregationService.getSystemOverview();
        return ResponseEntity.ok(overview);
    }
    
    @GetMapping("/tasks")
    public ResponseEntity<TaskMetricsSummary> getTaskMetrics(
            @RequestParam(required = false) String taskType,
            @RequestParam(defaultValue = "1h") String timeRange) {
        TaskMetricsSummary metrics = aggregationService.getTaskMetrics(taskType, timeRange);
        return ResponseEntity.ok(metrics);
    }
    
    @GetMapping("/resources")
    public ResponseEntity<ResourceMetrics> getResourceMetrics(
            @RequestParam(required = false) String node,
            @RequestParam(defaultValue = "30m") String timeRange) {
        ResourceMetrics metrics = aggregationService.getResourceMetrics(node, timeRange);
        return ResponseEntity.ok(metrics);
    }
    
    @GetMapping("/alerts")
    public ResponseEntity<AlertSummary> getAlertSummary(
            @RequestParam(required = false) String severity,
            @RequestParam(defaultValue = "24h") String timeRange) {
        AlertSummary alerts = aggregationService.getAlertSummary(severity, timeRange);
        return ResponseEntity.ok(alerts);
    }
}

指标标准化：

@Component
public class MetricsStandardizer {
    
    public StandardizedMetrics standardize(MetricData rawData) {
        StandardizedMetrics metrics = new StandardizedMetrics();
        
        // 标准化时间戳
        metrics.setTimestamp(standardizeTimestamp(rawData.getTimestamp()));
        
        // 标准化指标名称
        metrics.setMetricName(standardizeMetricName(rawData.getName()));
        
        // 标准化标签
        metrics.setLabels(standardizeLabels(rawData.getLabels()));
        
        // 标准化值
        metrics.setValue(standardizeValue(rawData.getValue()));
        
        // 添加元数据
        metrics.setSource(rawData.getSource());
        metrics.setVersion("1.0");
        
        return metrics;
    }
    
    private String standardizeMetricName(String name) {
        // 转换为snake_case
        return name.replaceAll("([a-z])([A-Z])", "$1_$2").toLowerCase();
    }
    
    private Map<String, String> standardizeLabels(Map<String, String> labels) {
        Map<String, String> standardized = new HashMap<>();
        
        // 标准化常见标签
        standardized.put("service", standardizeServiceName(labels.get("service")));
        standardized.put("instance", standardizeInstanceName(labels.get("instance")));
        standardized.put("cluster", standardizeClusterName(labels.get("cluster")));
        
        // 保留其他标签
        labels.forEach((key, value) -> {
            if (!standardized.containsKey(key)) {
                standardized.put(key, value);
            }
        });
        
        return standardized;
    }
}