Python指标采集怎么做

wen python案例 2026-07-11 208

本文目录导读：

Python指标采集怎么做

基础系统指标采集
Prometheus指标采集
应用自定义指标
数据库指标采集
日志指标采集
完整的监控系统示例
最佳实践建议

我来详细介绍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))