未记录事实核验 · 状态：维护中
记录仅说明核验时间与维护状态，不保证适用于所有环境；使用前请留意正文中的版本、适用条件与风险提示。
warning 适用边界与维护说明
本文适合按正文顺序、在单实例环境中学习RED/USE指标组织、Python指标导出、结构化日志、本地日志分析与模拟告警的组合实践；本次仅作结构审阅，未执行代码或进行外部事实核验。当前代码已通过Counter导入别名、复用全局指标收集器及错误率查询聚合修正原有名称覆盖、单实例重复注册和标签匹配问题，但RED计算仍可能保留停止请求后的过期数据，磁盘饱和度缺少采样间隔依据，部分告警记录值与触发指标不符，模型准确率未接入告警评估，指标恢复后也不会自动解除告警。请求、活跃连接数、模型准确率及通知发送包含模拟或占位实现，部分USE错误计数固定为零；正文未提供Prometheus抓取、Grafana数据源接入、ELK部署或APM链路追踪配置，因此学习目标中的部署表述不代表已交付完整生产方案，健康摘要也不能证明自动恢复或SLA保障。
系统监控 - 可观测性工程实践
abstract 📋 学习目标
- 掌握监控指标体系设计（RED、USE方法）
- 实施Prometheus+Grafana监控栈
- 构建日志聚合与分析系统
- 制定告警策略与事件管理流程
- 部署APM应用性能监控
- 实现完整监控系统案例
监控指标体系设计
RED方法论（面向服务监控）
RED方法关注三个关键维度：Rate（请求速率）、Errors（错误率）、Duration（响应时间）。
from dataclasses import dataclass
from typing import Dict, List, Optional
from datetime import datetime, timedelta
import time
import threading
from collections import deque, defaultdict
import statistics
import json
@dataclass
class REDMetrics:
"""RED监控指标"""
timestamp: datetime
service_name: str
endpoint: str
# Rate: 请求速率
request_rate: float # 每秒请求数
# Errors: 错误率
error_rate: float # 错误请求比例，取值0到1；展示为百分比时乘以100
# Duration: 响应时间
avg_duration: float # 平均响应时间
p95_duration: float # 95%分位响应时间
p99_duration: float # 99%分位响应时间
class REDMonitor:
"""RED方法监控实现"""
def __init__(self, window_size: int = 300): # 5分钟窗口
self.window_size = window_size
self.metrics_buffer = deque()
self.lock = threading.Lock()
def record_request(self, service: str, endpoint: str,
duration: float, is_error: bool):
"""记录请求指标"""
with self.lock:
self.metrics_buffer.append({
'timestamp': datetime.now(),
'service': service,
'endpoint': endpoint,
'duration': duration,
'is_error': is_error
})
# 清理超出时间窗口的数据
cutoff = datetime.now() - timedelta(seconds=self.window_size)
while (self.metrics_buffer and
self.metrics_buffer[0]['timestamp'] < cutoff):
self.metrics_buffer.popleft()
def calculate_red_metrics(self, service: str, endpoint: str) -> REDMetrics:
"""计算RED指标"""
with self.lock:
# 过滤相关请求
relevant_requests = [
req for req in self.metrics_buffer
if req['service'] == service and req['endpoint'] == endpoint
]
if not relevant_requests:
return REDMetrics(
timestamp=datetime.now(),
service_name=service,
endpoint=endpoint,
request_rate=0.0,
error_rate=0.0,
avg_duration=0.0,
p95_duration=0.0,
p99_duration=0.0
)
# 计算请求速率 (requests per second)
time_span = self.window_size
request_rate = len(relevant_requests) / time_span
# 计算错误率
error_count = sum(1 for req in relevant_requests if req['is_error'])
error_rate = error_count / len(relevant_requests)
# 计算响应时间统计
durations = [req['duration'] for req in relevant_requests]
avg_duration = statistics.mean(durations)
sorted_durations = sorted(durations)
p95_index = int(len(sorted_durations) * 0.95)
p99_index = int(len(sorted_durations) * 0.99)
p95_duration = sorted_durations[p95_index]
p99_duration = sorted_durations[p99_index]
return REDMetrics(
timestamp=datetime.now(),
service_name=service,
endpoint=endpoint,
request_rate=request_rate,
error_rate=error_rate,
avg_duration=avg_duration,
p95_duration=p95_duration,
p99_duration=p99_duration
)
USE方法论（面向基础设施监控）
USE方法关注：Utilization（使用率）、Saturation（饱和度）、Errors（错误数）。
import psutil
import subprocess
import os
@dataclass
class USEMetrics:
"""USE监控指标"""
timestamp: datetime
resource_name: str
# Utilization: 资源使用率
utilization: float # 0-100%
# Saturation: 资源饱和度
saturation: float # 队列长度或等待时间
# Errors: 错误计数
error_count: int # 错误事件数量
class USEMonitor:
"""USE方法监控实现"""
def __init__(self):
self.previous_disk_io = psutil.disk_io_counters()
self.previous_network_io = psutil.net_io_counters()
def collect_cpu_metrics(self) -> USEMetrics:
"""CPU指标收集"""
# Utilization: CPU使用率
cpu_percent = psutil.cpu_percent(interval=1)
# Saturation: 负载平均值
if hasattr(os, 'getloadavg'):
load_avg = os.getloadavg()[0] # 1分钟负载
cpu_count = psutil.cpu_count()
saturation = (load_avg / cpu_count) * 100
else:
saturation = 0.0
# Errors: 上下文切换异常（简化指标）
error_count = 0 # 需要系统级监控工具获取
return USEMetrics(
timestamp=datetime.now(),
resource_name='CPU',
utilization=cpu_percent,
saturation=saturation,
error_count=error_count
)
def collect_memory_metrics(self) -> USEMetrics:
"""内存指标收集"""
memory = psutil.virtual_memory()
# Utilization: 内存使用率
utilization = memory.percent
# Saturation: 交换分区使用情况
swap = psutil.swap_memory()
saturation = swap.percent
# Errors: 内存错误（简化）
error_count = 0
return USEMetrics(
timestamp=datetime.now(),
resource_name='Memory',
utilization=utilization,
saturation=saturation,
error_count=error_count
)
def collect_disk_metrics(self) -> USEMetrics:
"""磁盘指标收集"""
# Utilization: 磁盘使用率
disk_usage = psutil.disk_usage('/')
utilization = disk_usage.percent
# Saturation: 磁盘IO等待
current_disk_io = psutil.disk_io_counters()
if self.previous_disk_io:
io_time_diff = (current_disk_io.read_time + current_disk_io.write_time) - \
(self.previous_disk_io.read_time + self.previous_disk_io.write_time)
saturation = min(io_time_diff / 10, 100) # 简化计算
else:
saturation = 0.0
self.previous_disk_io = current_disk_io
return USEMetrics(
timestamp=datetime.now(),
resource_name='Disk',
utilization=utilization,
saturation=saturation,
error_count=0
)
Prometheus监控栈实现
Prometheus指标导出
from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time
import random
from functools import wraps
class PrometheusMetrics:
"""Prometheus指标收集器"""
def __init__(self):
# 计数器：累计值
self.request_count = Counter(
'http_requests_total',
'HTTP请求总数',
['method', 'endpoint', 'status']
)
# 直方图：响应时间分布
self.request_duration = Histogram(
'http_request_duration_seconds',
'HTTP请求响应时间',
['method', 'endpoint']
)
# 计量器：瞬时值
self.active_connections = Gauge(
'http_active_connections',
'当前活跃连接数'
)
self.system_memory_usage = Gauge(
'system_memory_usage_percent',
'系统内存使用率'
)
self.model_accuracy = Gauge(
'ml_model_accuracy',
'ML模型准确率',
['model_name', 'version']
)
def record_request(self, method: str, endpoint: str,
status_code: int, duration: float):
"""记录HTTP请求指标"""
self.request_count.labels(
method=method,
endpoint=endpoint,
status=str(status_code)
).inc()
self.request_duration.labels(
method=method,
endpoint=endpoint
).observe(duration)
def update_system_metrics(self):
"""更新系统指标"""
memory = psutil.virtual_memory()
self.system_memory_usage.set(memory.percent)
# 模拟活跃连接数
self.active_connections.set(random.randint(10, 100))
def update_model_metrics(self, model_name: str, version: str, accuracy: float):
"""更新模型指标"""
self.model_accuracy.labels(
model_name=model_name,
version=version
).set(accuracy)
# 全局指标实例
metrics = PrometheusMetrics()
def prometheus_monitor(endpoint: str):
"""Prometheus监控装饰器"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
start_time = time.time()
status_code = 200
try:
result = func(*args, **kwargs)
return result
except Exception as e:
status_code = 500
raise
finally:
duration = time.time() - start_time
metrics.record_request('POST', endpoint, status_code, duration)
return wrapper
return decorator
# 启动Prometheus HTTP服务器
def start_metrics_server(port: int = 8000):
"""启动指标服务器"""
start_http_server(port)
print(f"Prometheus指标服务器启动在端口 {port}")
# 后台更新系统指标
def background_metrics_update():
"""后台更新系统指标"""
import threading
def update_loop():
while True:
metrics.update_system_metrics()
# 模拟模型指标更新
metrics.update_model_metrics('user_recommender', 'v1.2', 0.85 + random.uniform(-0.05, 0.05))
time.sleep(15) # 每15秒更新一次
thread = threading.Thread(target=update_loop, daemon=True)
thread.start()
Grafana仪表板配置
{
"dashboard": {
"title": "应用监控仪表板",
"panels": [
{
"title": "请求速率",
"type": "graph",
"targets": [
{
"expr": "rate(http_requests_total[5m])",
"legendFormat": "{{method}} {{endpoint}}"
}
]
},
{
"title": "错误率",
"type": "stat",
"targets": [
{
"expr": "sum by (method, endpoint) (rate(http_requests_total{status=~\"5..\"}[5m])) / sum by (method, endpoint) (rate(http_requests_total[5m])) * 100"
}
],
"thresholds": [
{"color": "green", "value": 0},
{"color": "yellow", "value": 1},
{"color": "red", "value": 5}
]
},
{
"title": "响应时间P95",
"type": "graph",
"targets": [
{
"expr": "histogram_quantile(0.95, rate(http_request_duration_seconds_bucket[5m]))"
}
]
},
{
"title": "系统资源",
"type": "graph",
"targets": [
{
"expr": "system_memory_usage_percent",
"legendFormat": "内存使用率"
}
]
}
]
}
}
日志聚合与分析（ELK栈）
结构化日志记录
import logging
import json
from datetime import datetime
from typing import Dict, Any
import traceback
import sys
class StructuredLogger:
"""结构化日志记录器"""
def __init__(self, service_name: str, version: str = "1.0"):
self.service_name = service_name
self.version = version
# 配置日志格式
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
# 控制台处理器
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setFormatter(formatter)
# 文件处理器
file_handler = logging.FileHandler(f'{service_name}.log')
file_handler.setFormatter(formatter)
# 创建logger
self.logger = logging.getLogger(service_name)
self.logger.setLevel(logging.INFO)
self.logger.addHandler(console_handler)
self.logger.addHandler(file_handler)
def _create_log_entry(self, level: str, message: str, **kwargs) -> Dict[str, Any]:
"""创建结构化日志条目"""
return {
"timestamp": datetime.now().isoformat(),
"service": self.service_name,
"version": self.version,
"level": level,
"message": message,
"context": kwargs
}
def info(self, message: str, **kwargs):
"""信息日志"""
log_entry = self._create_log_entry("INFO", message, **kwargs)
self.logger.info(json.dumps(log_entry))
def warning(self, message: str, **kwargs):
"""警告日志"""
log_entry = self._create_log_entry("WARNING", message, **kwargs)
self.logger.warning(json.dumps(log_entry))
def error(self, message: str, exception: Exception = None, **kwargs):
"""错误日志"""
context = kwargs
if exception:
context.update({
"exception_type": type(exception).__name__,
"exception_message": str(exception),
"traceback": traceback.format_exc()
})
log_entry = self._create_log_entry("ERROR", message, **context)
self.logger.error(json.dumps(log_entry))
def audit(self, action: str, user_id: str, resource: str, **kwargs):
"""审计日志"""
context = {
"action": action,
"user_id": user_id,
"resource": resource,
**kwargs
}
log_entry = self._create_log_entry("AUDIT", f"User {user_id} {action} {resource}", **context)
self.logger.info(json.dumps(log_entry))
# 使用示例
app_logger = StructuredLogger("recommendation_service", "v2.1")
@prometheus_monitor('/api/recommend')
def recommend_items(user_id: str, category: str):
"""推荐服务示例"""
start_time = time.time()
try:
app_logger.info(
"开始处理推荐请求",
user_id=user_id,
category=category,
request_id=f"req_{int(time.time())}"
)
# 模拟推荐逻辑
time.sleep(random.uniform(0.1, 0.3))
# 模拟偶发错误
if random.random() < 0.05:
raise ValueError("推荐模型暂时不可用")
recommendations = [f"item_{i}" for i in range(5)]
app_logger.info(
"推荐请求成功完成",
user_id=user_id,
category=category,
recommendation_count=len(recommendations),
processing_time=time.time() - start_time
)
# 审计日志
app_logger.audit(
action="GET_RECOMMENDATIONS",
user_id=user_id,
resource=f"category/{category}",
item_count=len(recommendations)
)
return {"recommendations": recommendations, "status": "success"}
except Exception as e:
app_logger.error(
"推荐请求处理失败",
user_id=user_id,
category=category,
processing_time=time.time() - start_time,
exception=e
)
raise
日志分析器
import re
from collections import Counter as CollectionsCounter, defaultdict
from typing import List, Dict
import pandas as pd
class LogAnalyzer:
"""日志分析器"""
def __init__(self, log_file: str):
self.log_file = log_file
self.parsed_logs = []
def parse_logs(self) -> List[Dict]:
"""解析日志文件"""
with open(self.log_file, 'r') as f:
for line in f:
try:
# 尝试解析JSON格式日志
if line.strip().endswith('}'):
json_part = line.split(' - ', 3)[-1]
log_data = json.loads(json_part)
self.parsed_logs.append(log_data)
except:
continue
return self.parsed_logs
def analyze_error_patterns(self) -> Dict[str, int]:
"""分析错误模式"""
error_logs = [log for log in self.parsed_logs if log.get('level') == 'ERROR']
error_messages = [log.get('message', '') for log in error_logs]
return CollectionsCounter(error_messages)
def analyze_performance_trends(self) -> pd.DataFrame:
"""分析性能趋势"""
perf_data = []
for log in self.parsed_logs:
if 'processing_time' in log.get('context', {}):
perf_data.append({
'timestamp': log['timestamp'],
'processing_time': log['context']['processing_time'],
'service': log['service']
})
return pd.DataFrame(perf_data)
def generate_report(self) -> Dict[str, Any]:
"""生成分析报告"""
total_logs = len(self.parsed_logs)
error_count = len([log for log in self.parsed_logs if log.get('level') == 'ERROR'])
warning_count = len([log for log in self.parsed_logs if log.get('level') == 'WARNING'])
return {
'total_logs': total_logs,
'error_count': error_count,
'warning_count': warning_count,
'error_rate': error_count / total_logs if total_logs > 0 else 0,
'top_errors': self.analyze_error_patterns(),
'performance_summary': self.analyze_performance_trends().describe().to_dict() if not self.analyze_performance_trends().empty else {}
}
告警策略与事件管理
告警规则引擎
from enum import Enum
from typing import Callable, List
import smtplib
from email.mime.text import MIMEText
class AlertLevel(Enum):
INFO = "info"
WARNING = "warning"
CRITICAL = "critical"
@dataclass
class AlertRule:
"""告警规则定义"""
name: str
condition: Callable[[Dict], bool]
level: AlertLevel
message: str
cooldown_minutes: int = 5
@dataclass
class Alert:
"""告警事件"""
rule_name: str
level: AlertLevel
message: str
timestamp: datetime
metric_value: float
resolved: bool = False
class AlertManager:
"""告警管理器"""
def __init__(self):
self.rules: List[AlertRule] = []
self.active_alerts: List[Alert] = []
self.alert_history: List[Alert] = []
self.last_alert_times: Dict[str, datetime] = {}
def add_rule(self, rule: AlertRule):
"""添加告警规则"""
self.rules.append(rule)
def evaluate_metrics(self, metrics: Dict[str, float]):
"""评估指标并触发告警"""
current_time = datetime.now()
for rule in self.rules:
# 检查冷却期
last_alert = self.last_alert_times.get(rule.name)
if last_alert:
cooldown_delta = timedelta(minutes=rule.cooldown_minutes)
if current_time - last_alert < cooldown_delta:
continue
# 评估条件
if rule.condition(metrics):
alert = Alert(
rule_name=rule.name,
level=rule.level,
message=rule.message,
timestamp=current_time,
metric_value=list(metrics.values())[0] # 简化
)
self.trigger_alert(alert)
self.last_alert_times[rule.name] = current_time
def trigger_alert(self, alert: Alert):
"""触发告警"""
self.active_alerts.append(alert)
self.alert_history.append(alert)
# 发送通知
self.send_notification(alert)
print(f"🚨 告警触发: [{alert.level.value.upper()}] {alert.message}")
def send_notification(self, alert: Alert):
"""发送通知"""
# 这里可以集成多种通知方式：邮件、Slack、短信等
notification_message = f"""
告警级别: {alert.level.value.upper()}
告警规则: {alert.rule_name}
告警消息: {alert.message}
触发时间: {alert.timestamp}
指标值: {alert.metric_value}
"""
# 根据告警级别选择通知渠道
if alert.level == AlertLevel.CRITICAL:
self._send_email(alert, notification_message)
self._send_slack_message(alert, notification_message)
elif alert.level == AlertLevel.WARNING:
self._send_slack_message(alert, notification_message)
def _send_email(self, alert: Alert, message: str):
"""发送邮件通知"""
# 实际实现需要配置SMTP服务器
print(f"📧 发送邮件告警: {alert.message}")
def _send_slack_message(self, alert: Alert, message: str):
"""发送Slack消息"""
# 实际实现需要Slack Webhook
print(f"💬 发送Slack告警: {alert.message}")
def resolve_alert(self, rule_name: str):
"""解决告警"""
for alert in self.active_alerts:
if alert.rule_name == rule_name and not alert.resolved:
alert.resolved = True
print(f"✅ 告警已解决: {alert.message}")
def get_active_alerts(self) -> List[Alert]:
"""获取活跃告警"""
return [alert for alert in self.active_alerts if not alert.resolved]
# 预定义告警规则
def create_standard_alert_rules() -> List[AlertRule]:
"""创建标准告警规则"""
return [
AlertRule(
name="high_error_rate",
condition=lambda m: m.get('error_rate', 0) > 0.05, # 5%错误率
level=AlertLevel.CRITICAL,
message="错误率超过5%阈值",
cooldown_minutes=3
),
AlertRule(
name="slow_response_time",
condition=lambda m: m.get('p95_response_time', 0) > 2.0, # 2秒响应时间
level=AlertLevel.WARNING,
message="P95响应时间超过2秒",
cooldown_minutes=5
),
AlertRule(
name="high_memory_usage",
condition=lambda m: m.get('memory_usage', 0) > 85, # 85%内存使用
level=AlertLevel.WARNING,
message="内存使用率超过85%",
cooldown_minutes=10
),
AlertRule(
name="low_model_accuracy",
condition=lambda m: m.get('model_accuracy', 1.0) < 0.7, # 70%准确率
level=AlertLevel.CRITICAL,
message="模型准确率低于70%阈值",
cooldown_minutes=15
)
]
完整监控系统实施案例
监控系统集成
class ComprehensiveMonitoringSystem:
"""综合监控系统"""
def __init__(self, service_name: str):
self.service_name = service_name
# 初始化各监控组件
self.red_monitor = REDMonitor()
self.use_monitor = USEMonitor()
self.prometheus_metrics = metrics # 本文单实例示例复用已注册的全局指标收集器
self.logger = StructuredLogger(service_name)
self.alert_manager = AlertManager()
# 添加标准告警规则
for rule in create_standard_alert_rules():
self.alert_manager.add_rule(rule)
# 启动后台任务
self._start_background_tasks()
def _start_background_tasks(self):
"""启动后台监控任务"""
import threading
def monitoring_loop():
while True:
try:
# 收集系统指标
cpu_metrics = self.use_monitor.collect_cpu_metrics()
memory_metrics = self.use_monitor.collect_memory_metrics()
disk_metrics = self.use_monitor.collect_disk_metrics()
# 更新Prometheus指标
self.prometheus_metrics.update_system_metrics()
# 准备告警评估指标
alert_metrics = {
'cpu_usage': cpu_metrics.utilization,
'memory_usage': memory_metrics.utilization,
'disk_usage': disk_metrics.utilization
}
# 评估告警
self.alert_manager.evaluate_metrics(alert_metrics)
# 记录系统状态
self.logger.info(
"系统监控检查",
cpu_usage=cpu_metrics.utilization,
memory_usage=memory_metrics.utilization,
disk_usage=disk_metrics.utilization
)
except Exception as e:
self.logger.error("监控系统异常", exception=e)
time.sleep(30) # 每30秒检查一次
monitor_thread = threading.Thread(target=monitoring_loop, daemon=True)
monitor_thread.start()
def monitor_request(self, endpoint: str):
"""请求监控装饰器"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
start_time = time.time()
is_error = False
try:
result = func(*args, **kwargs)
return result
except Exception as e:
is_error = True
self.logger.error(f"请求处理失败: {endpoint}", exception=e)
raise
finally:
duration = time.time() - start_time
# 记录RED指标
self.red_monitor.record_request(
self.service_name, endpoint, duration, is_error
)
# 记录Prometheus指标
status_code = 500 if is_error else 200
self.prometheus_metrics.record_request(
'POST', endpoint, status_code, duration
)
# 计算并评估RED指标
red_metrics = self.red_monitor.calculate_red_metrics(
self.service_name, endpoint
)
alert_metrics = {
'error_rate': red_metrics.error_rate,
'p95_response_time': red_metrics.p95_duration,
'request_rate': red_metrics.request_rate
}
self.alert_manager.evaluate_metrics(alert_metrics)
return wrapper
return decorator
def get_system_health_summary(self) -> Dict[str, Any]:
"""获取系统健康摘要"""
# 收集当前指标
cpu_metrics = self.use_monitor.collect_cpu_metrics()
memory_metrics = self.use_monitor.collect_memory_metrics()
# 获取活跃告警
active_alerts = self.alert_manager.get_active_alerts()
return {
'service_name': self.service_name,
'timestamp': datetime.now().isoformat(),
'system_resources': {
'cpu_utilization': cpu_metrics.utilization,
'memory_utilization': memory_metrics.utilization,
'cpu_saturation': cpu_metrics.saturation,
'memory_saturation': memory_metrics.saturation
},
'active_alerts': len(active_alerts),
'alert_details': [
{
'rule': alert.rule_name,
'level': alert.level.value,
'message': alert.message,
'timestamp': alert.timestamp.isoformat()
}
for alert in active_alerts
],
'health_status': 'critical' if any(alert.level == AlertLevel.CRITICAL for alert in active_alerts)
else 'warning' if active_alerts
else 'healthy'
}
# 实施监控系统
print("🚀 启动综合监控系统")
monitoring_system = ComprehensiveMonitoringSystem("ml_recommendation_service")
# 启动Prometheus指标服务器
start_metrics_server(8000)
background_metrics_update()
# 被监控的服务函数
@monitoring_system.monitor_request('/api/recommend')
def recommendation_service(user_id: str, category: str = "general"):
"""推荐服务"""
# 模拟处理时间
processing_time = random.uniform(0.1, 1.5)
time.sleep(processing_time)
# 模拟偶发错误和性能问题
if random.random() < 0.03: # 3%错误率
raise Exception("推荐模型服务异常")
if random.random() < 0.1: # 10%概率慢请求
time.sleep(2.5) # 触发慢响应告警
recommendations = [f"item_{i}" for i in range(random.randint(1, 10))]
return {
"user_id": user_id,
"category": category,
"recommendations": recommendations,
"model_version": "v2.1"
}
# 模拟服务调用
print("\n📊 模拟服务调用...")
for i in range(20):
try:
result = recommendation_service(f"user_{i}", "electronics")
if i % 5 == 0:
print(f"✅ 推荐请求成功: 用户user_{i}")
except Exception as e:
print(f"❌ 推荐请求失败: user_{i} - {str(e)}")
time.sleep(0.5)
# 生成监控报告
print("\n📈 系统健康摘要:")
health_summary = monitoring_system.get_system_health_summary()
print(f"服务状态: {health_summary['health_status']}")
print(f"活跃告警数: {health_summary['active_alerts']}")
print(f"CPU使用率: {health_summary['system_resources']['cpu_utilization']:.1f}%")
print(f"内存使用率: {health_summary['system_resources']['memory_utilization']:.1f}%")
if health_summary['alert_details']:
print("\n⚠️ 活跃告警:")
for alert in health_summary['alert_details']:
print(f" • [{alert['level'].upper()}] {alert['message']}")
关键知识连接
前置技能基础
- 模型部署 - 被监控的服务对象
- Flask Web开发 - Web服务监控基础
监控工具栈
- Prometheus - 指标收集与存储
- Grafana - 监控仪表板与可视化
- ELK Stack - 日志聚合与分析
- APM工具 - 应用性能监控
可观测性三大支柱
- 指标监控 - 量化系统状态
- 日志分析 - 记录系统行为
- 链路追踪 - 分布式系统调用链
业务价值实现
- SLA保障 - 服务质量承诺
- 故障预防 - 主动发现问题
- 性能优化 - 基于数据驱动优化
- 成本控制 - 资源使用效率
创建时间：2024-06-29
最后更新：2024-12-10
学习时长：120分钟
难度等级：⭐⭐⭐（中级）
#系统监控 #Prometheus #Grafana #可观测性 #APM #告警策略 #日志分析 #RED方法 #USE方法