Prometheus 自动化巡检脚本集

#!/usr/bin/env python3
"""
批量服务拨测脚本
通过 Prometheus API 查询 Blackbox Exporter 探测结果
"""

import requests
import json
from datetime import datetime
from typing import List, Dict, Optional
from dataclasses import dataclass, asdict
import smtplib
from email.mime.text import MIMEText
import sys

@dataclass
class ProbeResult:
    """探测结果"""
    instance: str
    module: str          # http_2xx, tcp_connect, icmp 等
    success: bool
    status_code: Optional[int]
    duration: float      # 探测耗时（秒）
    error: Optional[str]
    last_error: Optional[str]
    ssl_cert_expiry_days: Optional[float]

class PrometheusProber:
    """Prometheus 拨测器"""

    def __init__(self, prometheus_url: str, timeout: int = 30):
        self.prometheus_url = prometheus_url.rstrip('/')
        self.timeout = timeout
        self.session = requests.Session()
        self.session.headers.update({'Content-Type': 'application/json'})

    def query(self, promql: str) -> List[Dict]:
        """执行 PromQL 即时查询"""
        resp = self.session.get(
            f"{self.prometheus_url}/api/v1/query",
            params={'query': promql},
            timeout=self.timeout
        )
        resp.raise_for_status()
        data = resp.json()

        if data['status'] != 'success':
            raise Exception(f"Prometheus 查询失败: {data.get('error', 'unknown')}")

        return data['data']['result']

    def query_range(self, promql: str, start: str, end: str, step: str = '60s') -> List[Dict]:
        """执行 PromQL 范围查询"""
        resp = self.session.get(
            f"{self.prometheus_url}/api/v1/query_range",
            params={
                'query': promql,
                'start': start,
                'end': end,
                'step': step
            },
            timeout=self.timeout
        )
        resp.raise_for_status()
        data = resp.json()
        return data['data']['result']

    def probe_all_services(self) -> List[ProbeResult]:
        """探测所有注册的服务"""
        results = []

        # 查询所有 Blackbox 探测结果
        probe_results = self.query(
            'probe_success * 1'
        )

        # 查询探测耗时
        duration_results = self.query('probe_duration_seconds')

        # 查询 HTTP 状态码
        status_results = self.query(
            'probe_http_status_code'
        )

        # 查询 SSL 证书到期天数
        ssl_results = self.query(
            'probe_ssl_earliest_cert_expiry - time()'
        )

        # 查询最近错误信息
        error_results = self.query('probe_last_error')

        # 合并结果
        duration_map = {r['metric']['instance']: float(r['value'][1]) for r in duration_results}
        status_map = {r['metric']['instance']: int(float(r['value'][1])) for r in status_results if r['value'][1] != 'NaN'}
        ssl_map = {r['metric']['instance']: float(r['value'][1]) / 86400 for r in ssl_results if r['value'][1] != 'NaN'}
        error_map = {r['metric']['instance']: r['value'][1] for r in error_results if r['value'][1] != ''}

        for item in probe_results:
            metrics = item['metric']
            instance = metrics.get('instance', 'unknown')
            module = metrics.get('job', 'unknown')
            success = item['value'][1] == '1'

            results.append(ProbeResult(
                instance=instance,
                module=module,
                success=success,
                status_code=status_map.get(instance),
                duration=duration_map.get(instance, 0),
                error=None if success else error_map.get(instance),
                last_error=error_map.get(instance) if not success else None,
                ssl_cert_expiry_days=ssl_map.get(instance)
            ))

        return results

    def check_endpoints(self, endpoints: List[str]) -> List[ProbeResult]:
        """检查指定端点列表"""
        results = self.probe_all_services()
        return [r for r in results if r.instance in endpoints]

class EndpointChecker:
    """端点检查器"""

    def __init__(self, prober: PrometheusProber):
        self.prober = prober

    def check_all(self) -> Dict:
        """执行全面检查"""
        results = self.prober.probe_all_services()

        total = len(results)
        success = sum(1 for r in results if r.success)
        failed = [r for r in results if not r.success]
        slow = [r for r in results if r.success and r.duration > 2.0]
        ssl_warning = [r for r in results if r.ssl_cert_expiry_days and r.ssl_cert_expiry_days < 30]

        report = {
            'timestamp': datetime.now().isoformat(),
            'summary': {
                'total': total,
                'success': success,
                'failed': len(failed),
                'success_rate': f"{success/total*100:.1f}%" if total > 0 else 'N/A',
                'slow_responses': len(slow),
                'ssl_expiring': len(ssl_warning),
            },
            'failed': [asdict(r) for r in failed],
            'slow': [{'instance': r.instance, 'duration': f"{r.duration:.2f}s"} for r in slow],
            'ssl_expiring': [{
                'instance': r.instance,
                'days_left': f"{r.ssl_cert_expiry_days:.0f}"
            } for r in ssl_warning],
        }

        return report

    def print_report(self, report: Dict):
        """打印报告"""
        print("\n" + "=" * 60)
        print("服务拨测报告")
        print("=" * 60)
        print(f"时间: {report['timestamp']}")

        s = report['summary']
        print(f"\n--- 概览 ---")
        print(f"  总端点数:    {s['total']}")
        print(f"  成功:        {s['success']} ({s['success_rate']})")
        print(f"  失败:        {s['failed']}")

        if s['failed'] > 0:
            print(f"\n--- 失败端点 ---")
            for item in report['failed']:
                print(f"  ✗ {item['instance']:<40} | {item['module']}")
                if item['error']:
                    print(f"    错误: {item['error'][:80]}")

        if report['slow']:
            print(f"\n--- 慢响应 (>2s) ---")
            for item in report['slow']:
                print(f"  ⚠ {item['instance']:<40} | {item['duration']}")

        if report['ssl_expiring']:
            print(f"\n--- SSL 证书即将过期 (<30天) ---")
            for item in report['ssl_expiring']:
                print(f"  ⚠ {item['instance']:<40} | {item['days_left']} 天")

        # 退出码
        if s['failed'] > 0:
            print(f"\n❌ 有 {s['failed']} 个端点不可用")
            return 1
        else:
            print(f"\n✅ 所有端点正常")
            return 0

def main():
    prometheus_url = sys.argv[1] if len(sys.argv) > 1 else 'http://prometheus:9090'

    prober = PrometheusProber(prometheus_url)
    checker = EndpointChecker(prober)

    report = checker.check_all()
    exit_code = checker.print_report(report)

    # 保存 JSON 报告
    report_file = f"/tmp/probe_report_{datetime.now():%Y%m%d_%H%M%S}.json"
    with open(report_file, 'w') as f:
        json.dump(report, f, indent=2, ensure_ascii=False, default=str)
    print(f"\n报告已保存: {report_file}")

    sys.exit(exit_code)

if __name__ == '__main__':
    main()

#!/usr/bin/env python3
"""直接 HTTP/TCP 拨测脚本"""

import requests
import socket
import ssl
from datetime import datetime, timezone
from urllib.parse import urlparse
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List, Dict
import json

class DirectProber:
    """直接探测器"""

    def __init__(self, timeout: int = 10):
        self.timeout = timeout

    def http_probe(self, url: str, expected_status: int = 200) -> Dict:
        """HTTP 探测"""
        result = {
            'type': 'http',
            'target': url,
            'timestamp': datetime.now(timezone.utc).isoformat(),
        }

        try:
            resp = requests.get(
                url,
                timeout=self.timeout,
                allow_redirects=True,
                verify=True,
                headers={'User-Agent': 'MonitoringProbe/1.0'}
            )
            result['success'] = resp.status_code == expected_status
            result['status_code'] = resp.status_code
            result['response_time'] = resp.elapsed.total_seconds()
            result['content_length'] = len(resp.content)
            result['ssl_verified'] = True

            # 检查 SSL 证书
            if url.startswith('https://'):
                parsed = urlparse(url)
                cert_info = self.check_ssl_cert(parsed.hostname, parsed.port or 443)
                result['ssl_info'] = cert_info

        except requests.exceptions.SSLError as e:
            result['success'] = False
            result['error'] = f'SSL Error: {e}'
        except requests.exceptions.ConnectionError as e:
            result['success'] = False
            result['error'] = f'Connection Error: {e}'
        except requests.exceptions.Timeout:
            result['success'] = False
            result['error'] = f'Timeout after {self.timeout}s'
        except Exception as e:
            result['success'] = False
            result['error'] = str(e)

        return result

    def tcp_probe(self, host: str, port: int) -> Dict:
        """TCP 端口探测"""
        result = {
            'type': 'tcp',
            'target': f'{host}:{port}',
            'timestamp': datetime.now(timezone.utc).isoformat(),
        }

        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(self.timeout)
            start = datetime.now()
            sock.connect((host, port))
            duration = (datetime.now() - start).total_seconds()
            sock.close()

            result['success'] = True
            result['response_time'] = duration
        except socket.timeout:
            result['success'] = False
            result['error'] = f'Timeout after {self.timeout}s'
        except ConnectionRefusedError:
            result['success'] = False
            result['error'] = 'Connection refused'
        except Exception as e:
            result['success'] = False
            result['error'] = str(e)

        return result

    def check_ssl_cert(self, hostname: str, port: int = 443) -> Dict:
        """检查 SSL 证书"""
        context = ssl.create_default_context()
        result = {}

        try:
            with socket.create_connection((hostname, port), timeout=self.timeout) as sock:
                with context.wrap_socket(sock, server_hostname=hostname) as ssock:
                    cert = ssock.getpeercert()

                    result['issuer'] = dict(x[0] for x in cert['issuer'])
                    result['subject'] = dict(x[0] for x in cert['subject'])
                    result['not_before'] = cert['notBefore']
                    result['not_after'] = cert['notAfter']

                    # 计算剩余天数
                    expiry_date = datetime.strptime(cert['notAfter'], '%b %d %H:%M:%S %Y %Z')
                    days_left = (expiry_date - datetime.now()).days
                    result['days_until_expiry'] = days_left
                    result['expiring_soon'] = days_left < 30

        except Exception as e:
            result['error'] = str(e)

        return result

    def batch_probe(self, targets: List[str], max_workers: int = 10) -> List[Dict]:
        """批量并行探测"""
        results = []

        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            futures = []
            for target in targets:
                if target.startswith('http://') or target.startswith('https://'):
                    futures.append(executor.submit(self.http_probe, target))
                else:
                    # 解析 host:port
                    parts = target.split(':')
                    host = parts[0]
                    port = int(parts[1]) if len(parts) > 1 else 80
                    futures.append(executor.submit(self.tcp_probe, host, port))

            for future in as_completed(futures):
                results.append(future.result())

        return results

# === 配置 ===
TARGETS = [
    'https://api.example.com/health',
    'https://www.example.com',
    'https://admin.example.com',
    'db.internal:5432',
    'redis.internal:6379',
    'rabbitmq.internal:5672',
    'https://grafana.example.com/api/health',
    'https://prometheus.example.com/-/healthy',
]

# === 执行 ===
prober = DirectProber(timeout=10)
results = prober.batch_probe(TARGETS, max_workers=5)

# 输出结果
print("\n=== 批量拨测结果 ===\n")
success_count = 0
for r in results:
    status = "✓" if r['success'] else "✗"
    time_str = f"{r.get('response_time', 0):.3f}s" if r.get('response_time') else 'N/A'
    print(f"  {status} {r['target']:<45} {time_str:>8}")
    if not r['success']:
        print(f"    └─ {r.get('error', 'unknown error')}")
    if 'ssl_info' in r and r['ssl_info'].get('days_until_expiry') is not None:
        days = r['ssl_info']['days_until_expiry']
        marker = "⚠" if days < 30 else " "
        print(f"    └─{marker} SSL 证书剩余 {days} 天")

print(f"\n总计: {len(results)}  成功: {sum(1 for r in results if r['success'])}  "
      f"失败: {sum(1 for r in results if not r['success'])}")

#!/usr/bin/env python3
"""
SSL 证书到期检查脚本
批量检查多域名证书有效期，提前告警
"""

import ssl
import socket
from datetime import datetime, timedelta
from typing import List, Dict
from concurrent.futures import ThreadPoolExecutor, as_completed
import json
import smtplib
from email.mime.text import MIMEText
import sys

class SSLCertChecker:
    """SSL 证书检查器"""

    def __init__(self, warning_days: int = 30, critical_days: int = 7, timeout: int = 10):
        self.warning_days = warning_days
        self.critical_days = critical_days
        self.timeout = timeout

    def check_cert(self, hostname: str, port: int = 443) -> Dict:
        """检查单个域名的 SSL 证书"""
        result = {
            'hostname': hostname,
            'port': port,
            'checked_at': datetime.now().isoformat(),
        }

        context = ssl.create_default_context()

        try:
            with socket.create_connection((hostname, port), timeout=self.timeout) as sock:
                with context.wrap_socket(sock, server_hostname=hostname) as ssock:
                    cert = ssock.getpeercert()

                    # 解析证书信息
                    result['issuer'] = dict(x[0] for x in cert['issuer'])
                    result['subject'] = dict(x[0] for x in cert['subject'])
                    result['not_before'] = cert['notBefore']
                    result['not_after'] = cert['notAfter']

                    # 获取 SAN（Subject Alternative Names）
                    san_list = []
                    for key, value in cert.get('subjectAltName', []):
                        san_list.append(value)
                    result['san'] = san_list

                    # 计算剩余天数
                    expiry_date = datetime.strptime(
                        cert['notAfter'], '%b %d %H:%M:%S %Y %Z'
                    )
                    days_left = (expiry_date - datetime.now()).days
                    result['days_left'] = days_left
                    result['expiry_date'] = expiry_date.strftime('%Y-%m-%d')

                    # 状态分级
                    if days_left <= self.critical_days:
                        result['status'] = 'CRITICAL'
                    elif days_left <= self.warning_days:
                        result['status'] = 'WARNING'
                    else:
                        result['status'] = 'OK'

                    result['success'] = True

        except ssl.SSLCertVerificationError as e:
            result['success'] = False
            result['status'] = 'ERROR'
            result['error'] = f'证书验证失败: {e}'
        except socket.timeout:
            result['success'] = False
            result['status'] = 'ERROR'
            result['error'] = f'连接超时 ({self.timeout}s)'
        except Exception as e:
            result['success'] = False
            result['status'] = 'ERROR'
            result['error'] = str(e)

        return result

    def batch_check(self, domains: List[str], max_workers: int = 10) -> List[Dict]:
        """批量检查"""
        results = []

        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            futures = {
                executor.submit(self.check_cert, domain): domain
                for domain in domains
            }

            for future in as_completed(futures):
                domain = futures[future]
                try:
                    result = future.result()
                    results.append(result)
                except Exception as e:
                    results.append({
                        'hostname': domain,
                        'success': False,
                        'status': 'ERROR',
                        'error': str(e)
                    })

        # 按紧急程度排序
        status_order = {'CRITICAL': 0, 'WARNING': 1, 'ERROR': 2, 'OK': 3}
        results.sort(key=lambda x: (
            status_order.get(x.get('status', 'ERROR'), 99),
            x.get('days_left', 9999)
        ))

        return results

    def print_report(self, results: List[Dict]):
        """打印报告"""
        print("\n" + "=" * 70)
        print("SSL 证书到期检查报告")
        print("=" * 70)
        print(f"检查时间: {datetime.now():%Y-%m-%d %H:%M:%S}")
        print(f"告警阈值: WARNING < {self.warning_days}天, CRITICAL < {self.critical_days}天")
        print("=" * 70)

        # 统计
        stats = {'OK': 0, 'WARNING': 0, 'CRITICAL': 0, 'ERROR': 0}
        for r in results:
            stats[r.get('status', 'ERROR')] = stats.get(r.get('status', 'ERROR'), 0) + 1

        print(f"\n总计: {len(results)}  "
              f"OK: {stats['OK']}  "
              f"WARNING: {stats['WARNING']}  "
              f"CRITICAL: {stats['CRITICAL']}  "
              f"ERROR: {stats['ERROR']}")

        # 详细列表
        print(f"\n{'域名':<35} {'状态':<10} {'剩余天数':<10} {'到期日期':<15} {'颁发者'}")
        print("-" * 90)

        for r in results:
            hostname = r['hostname'][:34]
            status = r.get('status', 'ERROR')

            if status == 'OK':
                marker = '  '
            elif status == 'WARNING':
                marker = '⚠ '
            elif status == 'CRITICAL':
                marker = '🔴'
            else:
                marker = '✗ '

            days = str(r.get('days_left', 'N/A'))
            expiry = r.get('expiry_date', 'N/A')
            issuer = r.get('issuer', {}).get('organizationName', 'N/A')[:25]

            print(f"{marker} {hostname:<33} {status:<10} {days:<10} {expiry:<15} {issuer}")

            if not r.get('success'):
                print(f"   └─ 错误: {r.get('error', 'unknown')[:70]}")

        # 退出码
        if stats['CRITICAL'] > 0 or stats['ERROR'] > 0:
            return 2
        elif stats['WARNING'] > 0:
            return 1
        return 0

    def send_alert_email(self, results: List[Dict], recipients: List[str],
                         smtp_server: str = 'localhost'):
        """发送告警邮件"""
        issues = [r for r in results if r.get('status') in ('CRITICAL', 'WARNING', 'ERROR')]
        if not issues:
            return

        subject = f"[SSL告警] {len(issues)} 个证书需要关注"
        body = "以下 SSL 证书即将过期或存在问题:\n\n"

        for r in issues:
            body += f"域名: {r['hostname']}\n"
            body += f"状态: {r.get('status', 'ERROR')}\n"
            if r.get('days_left') is not None:
                body += f"剩余天数: {r['days_left']}\n"
                body += f"到期日期: {r.get('expiry_date', 'N/A')}\n"
            if not r.get('success'):
                body += f"错误: {r.get('error', 'N/A')}\n"
            body += "\n"

        msg = MIMEText(body)
        msg['Subject'] = subject
        msg['From'] = 'ssl-monitor@example.com'
        msg['To'] = ', '.join(recipients)

        try:
            with smtplib.SMTP(smtp_server) as server:
                server.sendmail(msg['From'], recipients, msg.as_string())
            print(f"告警邮件已发送至 {len(recipients)} 个收件人")
        except Exception as e:
            print(f"邮件发送失败: {e}")

# === 配置 ===
DOMAINS = [
    'www.example.com',
    'api.example.com',
    'admin.example.com',
    'grafana.example.com',
    'prometheus.example.com',
    'alertmanager.example.com',
    'registry.example.com',
    'git.example.com',
    'jenkins.example.com',
    'vault.example.com',
]

# === 执行 ===
checker = SSLCertChecker(warning_days=30, critical_days=7)
results = checker.batch_check(DOMAINS)
exit_code = checker.print_report(results)

# 保存结果
report_file = f"/tmp/ssl_check_{datetime.now():%Y%m%d}.json"
with open(report_file, 'w') as f:
    json.dump(results, f, indent=2, ensure_ascii=False, default=str)
print(f"\n详细报告: {report_file}")

# 发送告警邮件（仅有问题时）
# checker.send_alert_email(results, ['ops@example.com'])

sys.exit(exit_code)

#!/usr/bin/env python3
"""
Prometheus 配置漂移检测
检查 Prometheus 配置文件、规则文件、Target 状态
"""

import requests
import yaml
import json
from datetime import datetime
from typing import List, Dict, Optional
import hashlib
import os

class PrometheusConfigChecker:
    """Prometheus 配置检查器"""

    def __init__(self, prometheus_url: str, config_path: str = '/etc/prometheus/prometheus.yml'):
        self.prometheus_url = prometheus_url.rstrip('/')
        self.config_path = config_path
        self.session = requests.Session()

    def check_config_loaded(self) -> Dict:
        """检查 Prometheus 加载的配置"""
        result = {'check': 'config_loaded', 'passed': True}

        try:
            resp = self.session.get(
                f"{self.prometheus_url}/api/v1/status/config",
                timeout=10
            )
            data = resp.json()

            if data['status'] == 'success':
                loaded_config = yaml.safe_load(data['data']['yaml'])
                result['scrape_configs_count'] = len(loaded_config.get('scrape_configs', []))
                result['rule_files'] = loaded_config.get('rule_files', [])

                # 与磁盘配置比较
                if os.path.exists(self.config_path):
                    with open(self.config_path) as f:
                        disk_config = yaml.safe_load(f)

                    disk_scrape_count = len(disk_config.get('scrape_configs', []))
                    if disk_scrape_count != result['scrape_configs_count']:
                        result['passed'] = False
                        result['error'] = (
                            f"配置不一致: 磁盘={disk_scrape_count}个scrape_config, "
                            f"运行中={result['scrape_configs_count']}个"
                        )
            else:
                result['passed'] = False
                result['error'] = '无法获取运行配置'

        except Exception as e:
            result['passed'] = False
            result['error'] = str(e)

        return result

    def check_targets_health(self) -> Dict:
        """检查所有 Target 的健康状态"""
        result = {'check': 'targets_health', 'passed': True, 'details': []}

        try:
            resp = self.session.get(
                f"{self.prometheus_url}/api/v1/targets",
                params={'state': 'active'},
                timeout=10
            )
            data = resp.json()

            active_targets = data['data']['activeTargets']
            total = len(active_targets)
            up = sum(1 for t in active_targets if t['health'] == 'up')
            down = [t for t in active_targets if t['health'] == 'down']

            result['total_targets'] = total
            result['up_targets'] = up
            result['down_targets'] = len(down)

            if down:
                result['passed'] = len(down) / total < 0.1  # 超过10% down 则告警
                result['details'] = [{
                    'job': t['labels'].get('job', ''),
                    'instance': t['labels'].get('instance', ''),
                    'last_error': t.get('lastError', ''),
                    'last_scrape': t.get('lastScrape', ''),
                } for t in down]

        except Exception as e:
            result['passed'] = False
            result['error'] = str(e)

        return result

    def check_rules_loaded(self) -> Dict:
        """检查告警规则加载状态"""
        result = {'check': 'rules_loaded', 'passed': True, 'details': {}}

        try:
            resp = self.session.get(
                f"{self.prometheus_url}/api/v1/rules",
                timeout=10
            )
            data = resp.json()

            groups = data['data']['groups']
            total_rules = 0
            alerting_rules = 0
            recording_rules = 0
            firing_alerts = 0

            for group in groups:
                for rule in group['rules']:
                    total_rules += 1
                    if rule['type'] == 'alerting':
                        alerting_rules += 1
                        for alert in rule.get('alerts', []):
                            if alert['state'] == 'firing':
                                firing_alerts += 1
                    elif rule['type'] == 'recording':
                        recording_rules += 1

            result['total_rules'] = total_rules
            result['alerting_rules'] = alerting_rules
            result['recording_rules'] = recording_rules
            result['firing_alerts'] = firing_alerts
            result['groups'] = len(groups)

        except Exception as e:
            result['passed'] = False
            result['error'] = str(e)

        return result

    def check_flags(self) -> Dict:
        """检查 Prometheus 启动参数"""
        result = {'check': 'flags', 'passed': True, 'issues': []}

        try:
            resp = self.session.get(
                f"{self.prometheus_url}/api/v1/status/flags",
                timeout=10
            )
            data = resp.json()
            flags = data['data']

            # 检查关键配置
            checks = [
                ('retention_time', '15d', lambda v: v >= '15d'),
                ('storage.tsdb.path', '/data/prometheus', lambda v: True),
                ('web.enable-lifecycle', 'true', lambda v: v == 'true'),
            ]

            for flag_name, expected, validator in checks:
                actual = flags.get(flag_name, 'not set')
                if not validator(actual):
                    result['passed'] = False
                    result['issues'].append(f"{flag_name}: expected={expected}, actual={actual}")

            result['flags'] = flags

        except Exception as e:
            result['passed'] = False
            result['error'] = str(e)

        return result

    def check_tsdb_stats(self) -> Dict:
        """检查 TSDB 统计信息"""
        result = {'check': 'tsdb_stats', 'passed': True}

        try:
            resp = self.session.get(
                f"{self.prometheus_url}/api/v1/status/tsdb",
                timeout=10
            )
            data = resp.json()
            stats = data['data']

            result['head_series'] = stats.get('headStats', {}).get('numSeries', 0)
            result['head_chunks'] = stats.get('headStats', {}).get('numLabelPairs', 0)
            result['block_count'] = len(stats.get('blocks', []))

            # 检查序列数是否异常增长
            if result['head_series'] > 5_000_000:
                result['passed'] = False
                result['warning'] = f"序列数过多: {result['head_series']:,}（可能存在高基数问题）"

        except Exception as e:
            result['passed'] = False
            result['error'] = str(e)

        return result

    def run_all_checks(self) -> List[Dict]:
        """运行所有检查"""
        return [
            self.check_config_loaded(),
            self.check_targets_health(),
            self.check_rules_loaded(),
            self.check_flags(),
            self.check_tsdb_stats(),
        ]

# === 执行 ===
checker = PrometheusConfigChecker('http://prometheus:9090')
results = checker.run_all_checks()

print("\n" + "=" * 60)
print("Prometheus 配置巡检报告")
print("=" * 60)

all_passed = True
for r in results:
    status = "✓ PASS" if r['passed'] else "✗ FAIL"
    print(f"\n[{status}] {r['check']}")

    if r['check'] == 'targets_health':
        print(f"  Targets: {r.get('up_targets', 0)}/{r.get('total_targets', 0)} UP")
        if r.get('down_targets', 0) > 0:
            print(f"  Down: {r['down_targets']}")
            for d in r.get('details', [])[:5]:
                print(f"    - {d['job']}/{d['instance']}: {d['last_error'][:60]}")

    elif r['check'] == 'rules_loaded':
        print(f"  规则: {r.get('total_rules', 0)} (告警={r.get('alerting_rules', 0)}, "
              f"记录={r.get('recording_rules', 0)})")
        print(f"  正在告警: {r.get('firing_alerts', 0)}")

    elif r['check'] == 'tsdb_stats':
        print(f"  序列数: {r.get('head_series', 0):,}")
        print(f"  Block数: {r.get('block_count', 0)}")

    if not r['passed']:
        all_passed = False
        if 'error' in r:
            print(f"  错误: {r['error']}")
        if 'warning' in r:
            print(f"  警告: {r['warning']}")
        if 'issues' in r:
            for issue in r['issues']:
                print(f"  - {issue}")

print("\n" + "=" * 60)
if all_passed:
    print("✅ 所有检查通过")
else:
    print("❌ 存在检查失败项，请处理")

#!/usr/bin/env bash
#
# prometheus_rules_check.sh - Prometheus 告警规则语法验证
#
set -euo pipefail

RULES_DIR="${1:-/etc/prometheus/rules}"
PROMTOOL="${PROMTOOL:-promtool}"
ERRORS=0
CHECKED=0

echo "========================================"
echo "Prometheus 规则文件检查"
echo "========================================"
echo "目录: ${RULES_DIR}"
echo ""

# 检查 promtool 是否可用
if ! command -v "${PROMTOOL}" &>/dev/null; then
    echo "✗ promtool 未安装"
    exit 1
fi

# 检查所有规则文件
for rule_file in "${RULES_DIR}"/*.yml "${RULES_DIR}"/*.yaml; do
    [[ -f "$rule_file" ]] || continue

    CHECKED=$((CHECKED + 1))
    filename=$(basename "$rule_file")

    if "${PROMTOOL}" check rules "$rule_file" 2>/dev/null; then
        echo "  ✓ ${filename}"

        # 统计规则数
        rule_count=$(yq e '.groups[].rules | length' "$rule_file" 2>/dev/null | paste -sd+ | bc || echo "?")
        echo "    规则数: ${rule_count}"
    else
        echo "  ✗ ${filename}"
        "${PROMTOOL}" check rules "$rule_file" 2>&1 | sed 's/^/    /'
        ERRORS=$((ERRORS + 1))
    fi
done

echo ""
echo "========================================"
echo "检查完成: ${CHECKED} 个文件, ${ERRORS} 个错误"

if [[ ${ERRORS} -gt 0 ]]; then
    exit 1
fi

echo "✅ 所有规则文件语法正确"

#!/usr/bin/env python3
"""
告警规则测试脚本
使用 promtool test rules 验证告警规则在不同数据场景下的触发情况
"""

import subprocess
import yaml
import json
import tempfile
import os
from typing import List, Dict
from datetime import datetime

class AlertRuleTester:
    """告警规则测试器"""

    def __init__(self, rules_file: str):
        self.rules_file = rules_file
        self.promtool = 'promtool'

    def generate_test_file(self, test_cases: List[Dict]) -> str:
        """生成 promtool 测试文件"""
        test_config = {
            'rule_files': [self.rules_file],
            'evaluation_interval': '1m',
            'tests': test_cases
        }

        # 写入临时文件
        fd, tmp_path = tempfile.mkstemp(suffix='.yml', prefix='prom_test_')
        with os.fdopen(fd, 'w') as f:
            yaml.dump(test_config, f, default_flow_style=False)

        return tmp_path

    def run_test(self, test_file: str) -> Dict:
        """运行测试"""
        result = subprocess.run(
            [self.promtool, 'test', 'rules', test_file],
            capture_output=True,
            text=True
        )

        return {
            'success': result.returncode == 0,
            'stdout': result.stdout,
            'stderr': result.stderr,
        }

    def test_high_cpu_alert(self):
        """测试高 CPU 告警规则"""
        test_cases = [
            # 场景1：CPU 正常，不应告警
            {
                'interval': '1m',
                'input_series': [{
                    'series': 'node_cpu_seconds_total{cpu="0",mode="idle",instance="web-01"}',
                    'values': '0+100x10'  # idle 持续增长
                }],
                'alert_rule_test': [{
                    'eval_time': '10m',
                    'alertname': 'HighCpuUsage',
                    'exp_alerts': []  # 期望不触发
                }]
            },
            # 场景2：CPU 飙高，应告警
            {
                'interval': '1m',
                'input_series': [{
                    'series': 'node_cpu_seconds_total{cpu="0",mode="idle",instance="web-01"}',
                    'values': '0+10x10'  # idle 增长慢，CPU 高
                }],
                'alert_rule_test': [{
                    'eval_time': '10m',
                    'alertname': 'HighCpuUsage',
                    'exp_alerts': [{
                        'exp_labels': {
                            'severity': 'warning',
                            'instance': 'web-01'
                        },
                        'exp_annotations': {}
                    }]
                }]
            }
        ]

        test_file = self.generate_test_file(test_cases)
        result = self.run_test(test_file)
        os.unlink(test_file)

        return result

    def test_disk_space_alert(self):
        """测试磁盘空间告警"""
        test_cases = [
            # 磁盘使用率 > 85% 持续 5 分钟
            {
                'interval': '1m',
                'input_series': [
                    {
                        'series': 'node_filesystem_size_bytes{mountpoint="/",instance="db-01"}',
                        'values': '100000000000x10'
                    },
                    {
                        'series': 'node_filesystem_free_bytes{mountpoint="/",instance="db-01"}',
                        'values': '10000000000x10'  # 10% free
                    }
                ],
                'alert_rule_test': [{
                    'eval_time': '5m',
                    'alertname': 'DiskSpaceWarning',
                    'exp_alerts': [{
                        'exp_labels': {
                            'severity': 'warning',
                            'instance': 'db-01'
                        },
                        'exp_annotations': {}
                    }]
                }]
            }
        ]

        test_file = self.generate_test_file(test_cases)
        result = self.run_test(test_file)
        os.unlink(test_file)

        return result

# === 使用示例 ===
tester = AlertRuleTester('/etc/prometheus/rules/node_alerts.yml')

print("=== 告警规则测试 ===\n")

tests = [
    ('HighCpuUsage', tester.test_high_cpu_alert),
    ('DiskSpaceWarning', tester.test_disk_space_alert),
]

all_passed = True
for name, test_func in tests:
    result = test_func()
    status = "✓ PASS" if result['success'] else "✗ FAIL"
    print(f"  {status}  {name}")
    if not result['success']:
        all_passed = False
        print(f"    {result['stderr'][:200]}")

print(f"\n{'✅ 所有测试通过' if all_passed else '❌ 存在测试失败'}")

#!/usr/bin/env python3
"""
Prometheus 自动化巡检报表生成器
汇总所有检查项，生成 HTML 报表
"""

import requests
import json
from datetime import datetime, timedelta
from typing import Dict, List
import os

class InspectionReportGenerator:
    """巡检报表生成器"""

    def __init__(self, prometheus_url: str):
        self.prometheus_url = prometheus_url.rstrip('/')
        self.session = requests.Session()

    def query(self, promql: str) -> List[Dict]:
        """执行 PromQL 查询"""
        resp = self.session.get(
            f"{self.prometheus_url}/api/v1/query",
            params={'query': promql},
            timeout=30
        )
        data = resp.json()
        return data.get('data', {}).get('result', [])

    def collect_metrics(self) -> Dict:
        """收集巡检指标"""
        metrics = {}

        # 1. Prometheus 自身状态
        metrics['prometheus_up'] = self.query('prometheus_tsdb_head_series')
        metrics['prometheus_ingest_rate'] = self.query(
            'sum(rate(prometheus_tsdb_head_samples_appended_total[5m]))'
        )
        metrics['prometheus_storage'] = self.query(
            'prometheus_tsdb_head_samples_appended_total'
        )

        # 2. Target 健康
        metrics['targets_up'] = self.query(
            'count(up == 1)'
        )
        metrics['targets_down'] = self.query(
            'count(up == 0)'
        )

        # 3. 告警状态
        metrics['alerts_firing'] = self.query(
            'sum(ALERTS{alertstate="firing"}) by (alertname, severity)'
        )
        metrics['alerts_pending'] = self.query(
            'sum(ALERTS{alertstate="pending"}) by (alertname)'
        )

        # 4. 节点健康
        metrics['nodes_total'] = self.query('count(node_uname_info)')
        metrics['nodes_down'] = self.query('count(up{job="node_exporter"} == 0)')

        # 5. 资源使用率 Top 5
        metrics['top_cpu'] = self.query(
            'topk(5, 100 - (avg by(instance)(rate(node_cpu_seconds_total{mode="idle"}[5m])) * 100))'
        )
        metrics['top_memory'] = self.query(
            'topk(5, 100 * (1 - node_memory_MemAvailable_bytes / node_memory_MemTotal_bytes))'
        )
        metrics['top_disk'] = self.query(
            'topk(5, 100 * (1 - node_filesystem_free_bytes{mountpoint="/"} / node_filesystem_size_bytes{mountpoint="/"}))'
        )

        # 6. 网络延迟
        metrics['http_latency_p99'] = self.query(
            'histogram_quantile(0.99, sum(rate(http_request_duration_seconds_bucket[5m])) by (le, handler))'
        )

        # 7. 服务可用性
        metrics['service_availability'] = self.query(
            'avg by(service)(rate(http_requests_total{status!~"5.."}[5m]) / rate(http_requests_total[5m])) * 100'
        )

        return metrics

    def generate_html_report(self) -> str:
        """生成 HTML 报表"""
        metrics = self.collect_metrics()
        now = datetime.now()

        # 解析关键指标
        targets_up = int(float(metrics['targets_up'][0]['value'][1])) if metrics['targets_up'] else 0
        targets_down = int(float(metrics['targets_down'][0]['value'][1])) if metrics['targets_down'] else 0
        total_targets = targets_up + targets_down
        availability = (targets_up / total_targets * 100) if total_targets > 0 else 0

        firing_alerts = metrics['alerts_firing']
        alert_count = len(firing_alerts)

        # Top CPU 表格行
        cpu_rows = ""
        for item in metrics.get('top_cpu', [])[:5]:
            instance = item['metric'].get('instance', 'N/A')
            value = float(item['value'][1])
            cpu_rows += f"<tr><td>{instance}</td><td>{value:.1f}%</td></tr>"

        # 告警表格行
        alert_rows = ""
        for item in firing_alerts:
            name = item['metric'].get('alertname', 'N/A')
            severity = item['metric'].get('severity', 'N/A')
            count = item['value'][1]
            color = '#f44336' if severity == 'critical' else '#ff9800'
            alert_rows += f"""<tr>
                <td style="color:{color};font-weight:bold">{severity}</td>
                <td>{name}</td>
                <td>{count}</td>
            </tr>"""

        html = f"""<!DOCTYPE html>
<html>
<head>
    <meta charset="utf-8">
    <title>巡检报告 - {now:%Y-%m-%d %H:%M}</title>
    <style>
        body {{ font-family: -apple-system, 'Segoe UI', sans-serif; margin: 0; padding: 20px; background: #f0f2f5; }}
        .header {{ background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); color: white;
                   padding: 30px; border-radius: 10px; margin-bottom: 20px; }}
        .header h1 {{ margin: 0; font-size: 24px; }}
        .header .meta {{ margin-top: 10px; opacity: 0.9; }}
        .grid {{ display: grid; grid-template-columns: repeat(4, 1fr); gap: 15px; margin-bottom: 20px; }}
        .card {{ background: white; border-radius: 8px; padding: 20px; box-shadow: 0 2px 8px rgba(0,0,0,0.08); }}
        .card .value {{ font-size: 2.5em; font-weight: 700; }}
        .card .label {{ color: #666; font-size: 0.85em; margin-top: 5px; }}
        .card.ok .value {{ color: #4CAF50; }}
        .card.warn .value {{ color: #FF9800; }}
        .card.error .value {{ color: #f44336; }}
        table {{ width: 100%; border-collapse: collapse; background: white; border-radius: 8px;
                overflow: hidden; box-shadow: 0 2px 8px rgba(0,0,0,0.08); }}
        th {{ background: #f5f5f5; padding: 12px; text-align: left; font-weight: 600; }}
        td {{ padding: 10px 12px; border-bottom: 1px solid #eee; }}
        .section {{ margin-bottom: 20px; }}
        .section h2 {{ color: #333; font-size: 18px; margin-bottom: 10px; }}
    </style>
</head>
<body>
    <div class="header">
        <h1>监控巡检报告</h1>
        <div class="meta">生成时间: {now:%Y-%m-%d %H:%M:%S} | Prometheus: {self.prometheus_url}</div>
    </div>

    <div class="grid">
        <div class="card {'ok' if availability > 95 else 'error' if availability < 90 else 'warn'}">
            <div class="value">{availability:.1f}%</div>
            <div class="label">Target 可用率 ({targets_up}/{total_targets})</div>
        </div>
        <div class="card {'ok' if alert_count == 0 else 'error'}">
            <div class="value">{alert_count}</div>
            <div class="label">活跃告警数</div>
        </div>
        <div class="card ok">
            <div class="value">{targets_up}</div>
            <div class="label">在线 Target</div>
        </div>
        <div class="card {'ok' if targets_down == 0 else 'error'}">
            <div class="value">{targets_down}</div>
            <div class="label">离线 Target</div>
        </div>
    </div>

    <div class="section">
        <h2>活跃告警</h2>
        <table>
            <tr><th>级别</th><th>告警名称</th><th>实例数</th></tr>
            {alert_rows if alert_rows else '<tr><td colspan="3" style="text-align:center;color:#999">暂无活跃告警</td></tr>'}
        </table>
    </div>

    <div class="section">
        <h2>CPU 使用率 Top 5</h2>
        <table>
            <tr><th>实例</th><th>CPU 使用率</th></tr>
            {cpu_rows if cpu_rows else '<tr><td colspan="2" style="text-align:center;color:#999">无数据</td></tr>'}
        </table>
    </div>

    <div class="section" style="color:#999;font-size:0.85em;text-align:center;margin-top:30px;">
        自动生成 by Prometheus 巡检脚本 | {now:%Y-%m-%d %H:%M:%S}
    </div>
</body>
</html>"""

        return html

    def save_report(self, output_dir: str = '/var/www/reports'):
        """保存报表"""
        os.makedirs(output_dir, exist_ok=True)

        html = self.generate_html_report()

        # 保存带时间戳的文件
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
        filepath = os.path.join(output_dir, f'inspection_{timestamp}.html')
        with open(filepath, 'w', encoding='utf-8') as f:
            f.write(html)

        # 同时保存为 latest.html
        latest_path = os.path.join(output_dir, 'latest.html')
        with open(latest_path, 'w', encoding='utf-8') as f:
            f.write(html)

        print(f"巡检报表已生成: {filepath}")
        print(f"最新报表: {latest_path}")
        return filepath

# === 执行 ===
generator = InspectionReportGenerator('http://prometheus:9090')
generator.save_report('/tmp/reports')