本文目录导读：
我来详细介绍Python指标采集的几种常见方法：
基础系统指标采集
CPU和内存使用率
import psutil
import time
# CPU使用率
cpu_percent = psutil.cpu_percent(interval=1) # 1秒采样
cpu_count = psutil.cpu_count()
cpu_freq = psutil.cpu_freq()
# 内存使用
memory = psutil.virtual_memory()
memory_total = memory.total
memory_used = memory.used
memory_percent = memory.percent
# 磁盘使用
disk = psutil.disk_usage('/')
disk_total = disk.total
disk_used = disk.used
disk_percent = disk.percent
print(f"CPU使用率: {cpu_percent}%")
print(f"内存使用率: {memory_percent}%")
print(f"磁盘使用率: {disk_percent}%")
Prometheus指标采集
示例：使用prometheus_client库
from prometheus_client import start_http_server, Gauge, Counter, Histogram
import psutil
import time
import random
# 定义指标
cpu_usage = Gauge('cpu_usage_percent', 'CPU使用率')
memory_usage = Gauge('memory_usage_percent', '内存使用率')
request_count = Counter('http_requests_total', 'HTTP请求总数')
request_duration = Histogram('http_request_duration_seconds', 'HTTP请求耗时')
# 启动Prometheus HTTP服务
start_http_server(8000)
while True:
# 采集系统指标
cpu_usage.set(psutil.cpu_percent(interval=1))
memory_usage.set(psutil.virtual_memory().percent)
# 模拟业务指标
request_count.inc(random.randint(0, 10))
request_duration.observe(random.uniform(0.1, 2.0))
time.sleep(5)
应用自定义指标
使用装饰器简化指标采集
import functools
import time
from collections import defaultdict
class MetricsCollector:
def __init__(self):
self.metrics = defaultdict(dict)
def record_time(self, metric_name):
"""装饰器：记录函数执行时间"""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
start = time.time()
result = func(*args, **kwargs)
duration = time.time() - start
if 'timing' not in self.metrics[metric_name]:
self.metrics[metric_name]['timing'] = []
self.metrics[metric_name]['timing'].append(duration)
return result
return wrapper
return decorator
def record_count(self, metric_name):
"""装饰器：记录函数调用次数"""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
if 'count' not in self.metrics[metric_name]:
self.metrics[metric_name]['count'] = 0
self.metrics[metric_name]['count'] += 1
return func(*args, **kwargs)
return wrapper
return decorator
def get_summary(self, metric_name):
"""获取指标统计摘要"""
data = self.metrics.get(metric_name, {})
if 'timing' in data and data['timing']:
timings = data['timing']
return {
'count': len(timings),
'avg': sum(timings) / len(timings),
'min': min(timings),
'max': max(timings),
'p99': sorted(timings)[int(len(timings) * 0.99)]
}
return data
# 使用示例
collector = MetricsCollector()
@collector.record_time('api_request')
@collector.record_count('api_calls')
def process_api_request():
time.sleep(0.1) # 模拟处理时间
return "OK"
# 调用函数
for _ in range(100):
process_api_request()
# 获取指标
print(collector.get_summary('api_request'))
数据库指标采集
import pymysql
import time
from contextlib import contextmanager
class DatabaseMetrics:
def __init__(self, host, user, password, database):
self.connection_config = {
'host': host,
'user': user,
'password': password,
'database': database
}
@contextmanager
def get_connection(self):
conn = pymysql.connect(**self.connection_config)
try:
yield conn
finally:
conn.close()
def collect_pg_metrics(self):
"""采集MySQL性能指标"""
with self.get_connection() as conn:
with conn.cursor() as cursor:
# 连接数
cursor.execute("SHOW STATUS LIKE 'Threads_connected'")
connections = cursor.fetchone()[1]
# 查询数
cursor.execute("SHOW STATUS LIKE 'Queries'")
queries = cursor.fetchone()[1]
# 慢查询数
cursor.execute("SHOW STATUS LIKE 'Slow_queries'")
slow_queries = cursor.fetchone()[1]
# InnoDB行锁等待
cursor.execute("SHOW STATUS LIKE 'Innodb_row_lock_current_waits'")
row_locks = cursor.fetchone()[1]
return {
'connections': connections,
'total_queries': queries,
'slow_queries': slow_queries,
'row_locks': row_locks
}
# 使用示例
db_metrics = DatabaseMetrics('localhost', 'user', 'password', 'mydb')
metrics = db_metrics.collect_pg_metrics()
print(f"数据库连接数: {metrics['connections']}")
日志指标采集
import re
from collections import Counter
from datetime import datetime
class LogMetricsCollector:
def __init__(self, log_file):
self.log_file = log_file
self.metrics = {
'error_count': 0,
'warning_count': 0,
'request_count': 0,
'response_times': [],
'status_codes': Counter(),
'error_patterns': Counter()
}
def parse_logs(self):
"""解析日志文件并提取指标"""
patterns = {
'error': r'ERROR|Error|error',
'warning': r'WARN|WARNING|warn|warning',
'request': r'GET|POST|PUT|DELETE\s+/\S+\s+HTTP/\d\.\d',
'status_code': r'HTTP/\d\.\d"\s+(\d{3})',
'response_time': r'response_time=(\d+\.?\d*)',
'api_path': r'(GET|POST|PUT|DELETE)\s+(/\S+)'
}
with open(self.log_file, 'r', encoding='utf-8') as f:
for line in f:
# 统计错误和警告
if re.search(patterns['error'], line):
self.metrics['error_count'] += 1
# 提取错误模式
error_match = re.search(r'error:?\s*(.*?)(?:\s|$)', line, re.IGNORECASE)
if error_match:
self.metrics['error_patterns'][error_match.group(1)] += 1
if re.search(patterns['warning'], line):
self.metrics['warning_count'] += 1
# 统计请求
if re.search(patterns['request'], line):
self.metrics['request_count'] += 1
# 统计状态码
status_match = re.search(patterns['status_code'], line)
if status_match:
self.metrics['status_codes'][status_match.group(1)] += 1
# 统计响应时间
time_match = re.search(patterns['response_time'], line)
if time_match:
self.metrics['response_times'].append(float(time_match.group(1)))
return self.metrics
def get_summary(self):
"""获取指标摘要"""
metrics = self.parse_logs()
response_times = metrics['response_times']
return {
'total_errors': metrics['error_count'],
'total_warnings': metrics['warning_count'],
'total_requests': metrics['request_count'],
'avg_response_time': sum(response_times) / len(response_times) if response_times else 0,
'max_response_time': max(response_times) if response_times else 0,
'status_code_distribution': dict(metrics['status_codes']),
'top_errors': metrics['error_patterns'].most_common(5)
}
# 使用示例
collector = LogMetricsCollector('/var/log/app.log')
summary = collector.get_summary()
print(f"错误总数: {summary['total_errors']}")
print(f"平均响应时间: {summary['avg_response_time']:.2f}ms")
完整的监控系统示例
import time
import threading
import json
from datetime import datetime
class MonitoringSystem:
def __init__(self, push_url=None):
self.metrics = {}
self.collectors = []
self.push_url = push_url
def register_collector(self, name, collector_func, interval=10):
"""注册指标采集器"""
self.collectors.append({
'name': name,
'func': collector_func,
'interval': interval,
'last_collected': 0
})
def start(self):
"""启动监控系统"""
def collect_loop():
while True:
for collector in self.collectors:
current_time = time.time()
if current_time - collector['last_collected'] >= collector['interval']:
try:
data = collector['func']()
self.metrics[collector['name']] = {
'data': data,
'timestamp': datetime.now().isoformat()
}
collector['last_collected'] = current_time
if self.push_url:
self._push_metrics(collector['name'], data)
except Exception as e:
print(f"采集器 {collector['name']} 错误: {e}")
time.sleep(1)
threading.Thread(target=collect_loop, daemon=True).start()
def _push_metrics(self, name, data):
"""推送指标到远程服务"""
try:
import requests
payload = {
'name': name,
'data': data,
'timestamp': datetime.now().isoformat()
}
requests.post(self.push_url, json=payload, timeout=5)
except Exception as e:
print(f"推送指标失败: {e}")
def get_metrics(self):
"""获取所有指标"""
return self.metrics
def get_metric(self, name):
"""获取特定指标"""
return self.metrics.get(name)
# 使用示例
monitor = MonitoringSystem(push_url='http://monitor.example.com/api/metrics')
# 注册系统指标采集器
monitor.register_collector('system', lambda: {
'cpu': psutil.cpu_percent(),
'memory': psutil.virtual_memory().percent,
'disk': psutil.disk_usage('/').percent
}, interval=60)
# 注册数据库指标采集器
db_metrics = DatabaseMetrics('localhost', 'user', 'password', 'mydb')
monitor.register_collector('database', db_metrics.collect_pg_metrics, interval=300)
# 启动监控
monitor.start()
# 获取指标
time.sleep(10)
all_metrics = monitor.get_metrics()
print(json.dumps(all_metrics, indent=2))
最佳实践建议
-
采集频率：
- 系统指标：10-60秒
- 业务指标：1-5分钟
- 慢变化指标：5-15分钟
-
存储策略：
- 使用时序数据库（Prometheus、InfluxDB）
- 实现数据聚合和降采样
- 设置合理的保留策略
-
告警机制：
- 设置阈值告警
- 异常检测算法
- 分级告警（INFO/WARNING/CRITICAL）
-
性能优化：
- 异步采集
- 批量处理
- 使用缓冲队列
- 避免阻塞主流程
-
安全考量：
- 指标访问认证
- 敏感信息脱敏
- 使用HTTPS传输
根据你的具体需求选择合适的方案,可以从简单的系统指标开始，逐步扩展到业务指标和日志分析。
