Python自动化运维中服务器性能监控与告警详解

import requests
import sys

def check_http_status(url, expected_code=200):
    try:
        response = requests.get(url, timeout=10)
        if response.status_code != expected_code:
            send_alert(f"HTTP状态异常：{url} 返回 {response.status_code}")
    except Exception as e:
        send_alert(f"服务不可达：{url}，错误：{str(e)}")

def send_alert(message):
    # 集成Webhook（如钉钉、企业微信）
    webhook_url = "https://oapi.dingtalk.com/robot/send?access_token=xxx"
    headers = {'Content-Type': 'application/json'}
    data = {"msgtype": "text", "text": {"content": message}}
    requests.post(webhook_url, json=data, headers=headers)

# 调用示例
check_http_status("http://example.com/api/health")

import re
from collections import defaultdict

def analyze_logs(log_path, pattern=r'ERROR: (.*)'):
    error_counts = defaultdict(int)
    with open(log_path, 'r') as f:
        for line in f:
            match = re.search(pattern, line)
            if match:
                error_type = match.group(1)
                error_counts[error_type] += 1
    # 触发阈值告警
    for error, count in error_counts.items():
        if count > 10:
            send_alert(f"错误类型 {error} 在日志中出现 {count} 次")

# 示例：监控Nginx错误日志
analyze_logs('/var/log/nginx/error.log')

import docker
client = docker.from_env()
for container in client.containers.list():
    stats = container.stats(stream=False)
    print(f"容器 {container.name} CPU使用率：{stats['cpu_percent']}%")

if disk.percent > 90:
    os.system("find /var/log -name '*.log' -mtime +7 -exec rm {} \;")