Overview

Monitoring systems themselves need to be monitored. Prometheus collects metrics from your entire infrastructure, but if Prometheus’s own configuration has issues, a target goes down, an SSL certificate is about to expire, or an alerting rule has a syntax error — who discovers these problems? The answer: an automated inspection script set. This article builds a complete inspection toolkit around the Prometheus ecosystem, covering “probing → certificate checking → config auditing → rule validation → alert simulation → report generation.”

References: Prometheus Official Documentation, Blackbox Exporter

I. Batch Service Probing Scripts

1.1 Probing via Blackbox Exporter

Blackbox Exporter is Prometheus’s official blackbox probing tool, supporting HTTP, TCP, ICMP, DNS, and other protocols. By querying probe results through the Prometheus API, you can implement batch service health checks:

#!/usr/bin/env python3
"""
Batch service probing script
Queries Blackbox Exporter probe results via Prometheus API
"""

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:
    """Probe result"""
    instance: str
    module: str          # http_2xx, tcp_connect, icmp, etc.
    success: bool
    status_code: Optional[int]
    duration: float      # Probe duration (seconds)
    error: Optional[str]
    last_error: Optional[str]
    ssl_cert_expiry_days: Optional[float]

class PrometheusProber:
    """Prometheus prober"""

    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]:
        """Execute a PromQL instant query"""
        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 query failed: {data.get('error', 'unknown')}")

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

    def query_range(self, promql: str, start: str, end: str, step: str = '60s') -> List[Dict]:
        """Execute a PromQL range query"""
        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]:
        """Probe all registered services"""
        results = []

        # Query all Blackbox probe results
        probe_results = self.query(
            'probe_success * 1'
        )

        # Query probe duration
        duration_results = self.query('probe_duration_seconds')

        # Query HTTP status codes
        status_results = self.query(
            'probe_http_status_code'
        )

        # Query SSL certificate expiry days
        ssl_results = self.query(
            'probe_ssl_earliest_cert_expiry - time()'
        )

        # Query latest error info
        error_results = self.query('probe_last_error')

        # Merge results
        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]:
        """Check a specific list of endpoints"""
        results = self.probe_all_services()
        return [r for r in results if r.instance in endpoints]

class EndpointChecker:
    """Endpoint checker"""

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

    def check_all(self) -> Dict:
        """Perform comprehensive checks"""
        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 the report"""
        print("\n" + "=" * 60)
        print("Service Probe Report")
        print("=" * 60)
        print(f"Time: {report['timestamp']}")

        s = report['summary']
        print(f"\n--- Overview ---")
        print(f"  Total endpoints:  {s['total']}")
        print(f"  Success:          {s['success']} ({s['success_rate']})")
        print(f"  Failed:           {s['failed']}")

        if s['failed'] > 0:
            print(f"\n--- Failed Endpoints ---")
            for item in report['failed']:
                print(f"  ✗ {item['instance']:<40} | {item['module']}")
                if item['error']:
                    print(f"    Error: {item['error'][:80]}")

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

        if report['ssl_expiring']:
            print(f"\n--- SSL Certificates Expiring (<30 days) ---")
            for item in report['ssl_expiring']:
                print(f"  ⚠ {item['instance']:<40} | {item['days_left']} days")

        # Exit code
        if s['failed'] > 0:
            print(f"\n❌ {s['failed']} endpoint(s) unavailable")
            return 1
        else:
            print(f"\n✅ All endpoints healthy")
            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)

    # Save JSON report
    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"\nReport saved: {report_file}")

    sys.exit(exit_code)

if __name__ == '__main__':
    main()

1.2 Direct Probing Script (Without Prometheus)

When Prometheus is unavailable, you can probe targets directly:

#!/usr/bin/env python3
"""Direct HTTP/TCP probing script"""

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:
    """Direct prober"""

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

    def http_probe(self, url: str, expected_status: int = 200) -> Dict:
        """HTTP probe"""
        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

            # Check SSL certificate
            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 port probe"""
        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:
        """Check SSL certificate"""
        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']

                    # Calculate days remaining
                    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]:
        """Batch parallel probing"""
        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:
                    # Parse 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

# === Configuration ===
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',
]

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

# Output results
print("\n=== Batch Probe Results ===\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 certificate {days} days remaining")

print(f"\nTotal: {len(results)}  Success: {sum(1 for r in results if r['success'])}  "
      f"Failed: {sum(1 for r in results if not r['success'])}")

II. SSL Certificate Expiry Checking

2.1 Batch Certificate Check Script

#!/usr/bin/env python3
"""
SSL certificate expiry check script
Batch check certificate validity for multiple domains, alert in advance
"""

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 certificate checker"""

    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:
        """Check SSL certificate for a single domain"""
        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()

                    # Parse certificate info
                    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']

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

                    # Calculate days remaining
                    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')

                    # Status classification
                    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'Certificate verification failed: {e}'
        except socket.timeout:
            result['success'] = False
            result['status'] = 'ERROR'
            result['error'] = f'Connection timeout ({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]:
        """Batch check"""
        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)
                    })

        # Sort by urgency
        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 report"""
        print("\n" + "=" * 70)
        print("SSL Certificate Expiry Check Report")
        print("=" * 70)
        print(f"Check time: {datetime.now():%Y-%m-%d %H:%M:%S}")
        print(f"Alert thresholds: WARNING < {self.warning_days} days, CRITICAL < {self.critical_days} days")
        print("=" * 70)

        # Statistics
        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"\nTotal: {len(results)}  "
              f"OK: {stats['OK']}  "
              f"WARNING: {stats['WARNING']}  "
              f"CRITICAL: {stats['CRITICAL']}  "
              f"ERROR: {stats['ERROR']}")

        # Detailed list
        print(f"\n{'Domain':<35} {'Status':<10} {'Days Left':<10} {'Expiry Date':<15} {'Issuer'}")
        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"   └─ Error: {r.get('error', 'unknown')[:70]}")

        # Exit code
        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'):
        """Send alert email"""
        issues = [r for r in results if r.get('status') in ('CRITICAL', 'WARNING', 'ERROR')]
        if not issues:
            return

        subject = f"[SSL Alert] {len(issues)} certificate(s) need attention"
        body = "The following SSL certificates are expiring soon or have issues:\n\n"

        for r in issues:
            body += f"Domain: {r['hostname']}\n"
            body += f"Status: {r.get('status', 'ERROR')}\n"
            if r.get('days_left') is not None:
                body += f"Days left: {r['days_left']}\n"
                body += f"Expiry date: {r.get('expiry_date', 'N/A')}\n"
            if not r.get('success'):
                body += f"Error: {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"Alert email sent to {len(recipients)} recipient(s)")
        except Exception as e:
            print(f"Email send failed: {e}")

# === Configuration ===
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',
]

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

# Save 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"\nDetailed report: {report_file}")

# Send alert email (only when there are issues)
# checker.send_alert_email(results, ['ops@example.com'])

sys.exit(exit_code)

III. Configuration Drift Detection

3.1 Prometheus Configuration Validation

#!/usr/bin/env python3
"""
Prometheus configuration drift detection
Checks Prometheus config files, rule files, and target status
"""

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

class PrometheusConfigChecker:
    """Prometheus configuration checker"""

    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:
        """Check Prometheus loaded configuration"""
        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', [])

                # Compare with on-disk config
                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"Config mismatch: disk={disk_scrape_count} scrape_configs, "
                            f"running={result['scrape_configs_count']}"
                        )
            else:
                result['passed'] = False
                result['error'] = 'Unable to fetch running config'

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

        return result

    def check_targets_health(self) -> Dict:
        """Check health status of all targets"""
        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  # Alert if more than 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:
        """Check alerting rule loading status"""
        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:
        """Check Prometheus startup flags"""
        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']

            # Check key configuration
            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:
        """Check TSDB statistics"""
        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', []))

            # Check for abnormal series growth
            if result['head_series'] > 5_000_000:
                result['passed'] = False
                result['warning'] = f"Too many series: {result['head_series']:,} (possible high cardinality issue)"

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

        return result

    def run_all_checks(self) -> List[Dict]:
        """Run all checks"""
        return [
            self.check_config_loaded(),
            self.check_targets_health(),
            self.check_rules_loaded(),
            self.check_flags(),
            self.check_tsdb_stats(),
        ]

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

print("\n" + "=" * 60)
print("Prometheus Configuration Inspection Report")
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"  Rules: {r.get('total_rules', 0)} (alerting={r.get('alerting_rules', 0)}, "
              f"recording={r.get('recording_rules', 0)})")
        print(f"  Firing: {r.get('firing_alerts', 0)}")

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

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

print("\n" + "=" * 60)
if all_passed:
    print("✅ All checks passed")
else:
    print("❌ Some checks failed, please investigate")

IV. Alerting Rule Validation

4.1 Prometheus Rule Syntax Check

#!/usr/bin/env bash
#
# prometheus_rules_check.sh - Prometheus alerting rule syntax validation
#
set -euo pipefail

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

echo "========================================"
echo "Prometheus Rule File Check"
echo "========================================"
echo "Directory: ${RULES_DIR}"
echo ""

# Check if promtool is available
if ! command -v "${PROMTOOL}" &>/dev/null; then
    echo "✗ promtool not installed"
    exit 1
fi

# Check all rule files
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}"

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

echo ""
echo "========================================"
echo "Check complete: ${CHECKED} file(s), ${ERRORS} error(s)"

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

echo "✅ All rule files have valid syntax"

4.2 Alerting Rule Testing

#!/usr/bin/env python3
"""
Alerting rule testing script
Uses promtool test rules to verify alerting rule behavior under different data scenarios
"""

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

class AlertRuleTester:
    """Alerting rule tester"""

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

    def generate_test_file(self, test_cases: List[Dict]) -> str:
        """Generate a promtool test file"""
        test_config = {
            'rule_files': [self.rules_file],
            'evaluation_interval': '1m',
            'tests': test_cases
        }

        # Write to temp file
        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:
        """Run the test"""
        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):
        """Test high CPU alert rule"""
        test_cases = [
            # Scenario 1: CPU normal, should not alert
            {
                'interval': '1m',
                'input_series': [{
                    'series': 'node_cpu_seconds_total{cpu="0",mode="idle",instance="web-01"}',
                    'values': '0+100x10'  # idle keeps growing
                }],
                'alert_rule_test': [{
                    'eval_time': '10m',
                    'alertname': 'HighCpuUsage',
                    'exp_alerts': []  # Expect no alert
                }]
            },
            # Scenario 2: CPU spikes, should alert
            {
                'interval': '1m',
                'input_series': [{
                    'series': 'node_cpu_seconds_total{cpu="0",mode="idle",instance="web-01"}',
                    'values': '0+10x10'  # idle grows slowly, CPU is high
                }],
                '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 disk space alert"""
        test_cases = [
            # Disk usage > 85% for 5 minutes
            {
                '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

# === Usage Example ===
tester = AlertRuleTester('/etc/prometheus/rules/node_alerts.yml')

print("=== Alerting Rule Tests ===\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{'✅ All tests passed' if all_passed else '❌ Some tests failed'}")

V. Automated Report Generation

5.1 Inspection Report Generator

#!/usr/bin/env python3
"""
Prometheus automated inspection report generator
Aggregates all check items and generates an HTML report
"""

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

class InspectionReportGenerator:
    """Inspection report generator"""

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

    def query(self, promql: str) -> List[Dict]:
        """Execute a PromQL query"""
        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:
        """Collect inspection metrics"""
        metrics = {}

        # 1. Prometheus self status
        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 health
        metrics['targets_up'] = self.query(
            'count(up == 1)'
        )
        metrics['targets_down'] = self.query(
            'count(up == 0)'
        )

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

        # 4. Node health
        metrics['nodes_total'] = self.query('count(node_uname_info)')
        metrics['nodes_down'] = self.query('count(up{job="node_exporter"} == 0)')

        # 5. Resource utilization 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. Network latency
        metrics['http_latency_p99'] = self.query(
            'histogram_quantile(0.99, sum(rate(http_request_duration_seconds_bucket[5m])) by (le, handler))'
        )

        # 7. Service availability
        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:
        """Generate HTML report"""
        metrics = self.collect_metrics()
        now = datetime.now()

        # Parse key metrics
        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 table rows
        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 table rows
        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>Inspection Report - {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>Monitoring Inspection Report</h1>
        <div class="meta">Generated: {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 Availability ({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">Active Alerts</div>
        </div>
        <div class="card ok">
            <div class="value">{targets_up}</div>
            <div class="label">Online Targets</div>
        </div>
        <div class="card {'ok' if targets_down == 0 else 'error'}">
            <div class="value">{targets_down}</div>
            <div class="label">Offline Targets</div>
        </div>
    </div>

    <div class="section">
        <h2>Active Alerts</h2>
        <table>
            <tr><th>Severity</th><th>Alert Name</th><th>Instance Count</th></tr>
            {alert_rows if alert_rows else '<tr><td colspan="3" style="text-align:center;color:#999">No active alerts</td></tr>'}
        </table>
    </div>

    <div class="section">
        <h2>CPU Usage Top 5</h2>
        <table>
            <tr><th>Instance</th><th>CPU Usage</th></tr>
            {cpu_rows if cpu_rows else '<tr><td colspan="2" style="text-align:center;color:#999">No data</td></tr>'}
        </table>
    </div>

    <div class="section" style="color:#999;font-size:0.85em;text-align:center;margin-top:30px;">
        Auto-generated by Prometheus inspection script | {now:%Y-%m-%d %H:%M:%S}
    </div>
</body>
</html>"""

        return html

    def save_report(self, output_dir: str = '/var/www/reports'):
        """Save the report"""
        os.makedirs(output_dir, exist_ok=True)

        html = self.generate_html_report()

        # Save timestamped file
        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)

        # Also save as 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"Inspection report generated: {filepath}")
        print(f"Latest report: {latest_path}")
        return filepath

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

VI. Scheduled Inspection and Alerting

6.1 Inspection Scheduler Script

#!/usr/bin/env bash
#
# inspection_scheduler.sh - Inspection task scheduler
#
set -euo pipefail

PROMETHEUS_URL="${PROMETHEUS_URL:-http://prometheus:9090}"
SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
REPORT_DIR="${REPORT_DIR:-/tmp/reports}"
LOG_FILE="/var/log/inspection.log"

log() { echo "[$(date '+%Y-%m-%d %H:%M:%S')] $*" | tee -a "$LOG_FILE"; }

mkdir -p "$REPORT_DIR"

# 1. Service probing (every 5 minutes)
run_probes() {
    log "Running service probes..."
    python3 "${SCRIPT_DIR}/probe_services.py" "$PROMETHEUS_URL" \
        > "${REPORT_DIR}/probe_result.json" 2>&1 || \
        log "Probing failed"
}

# 2. SSL certificate check (daily at 9:00)
check_ssl() {
    log "Running SSL certificate check..."
    python3 "${SCRIPT_DIR}/ssl_checker.py" \
        > "${REPORT_DIR}/ssl_result.json" 2>&1 || \
        log "SSL check failed"
}

# 3. Prometheus config check (hourly)
check_prometheus() {
    log "Running Prometheus config check..."
    python3 "${SCRIPT_DIR}/prometheus_config_check.py" "$PROMETHEUS_URL" \
        > "${REPORT_DIR}/config_result.json" 2>&1 || \
        log "Config check failed"
}

# 4. Generate inspection report (hourly)
generate_report() {
    log "Generating inspection report..."
    python3 "${SCRIPT_DIR}/generate_report.py" "$PROMETHEUS_URL" "$REPORT_DIR" \
        2>&1 | tee -a "$LOG_FILE" || \
        log "Report generation failed"
}

# 5. Alerting rule validation (manual after each deploy)
validate_rules() {
    log "Validating alerting rules..."
    bash "${SCRIPT_DIR}/prometheus_rules_check.sh" /etc/prometheus/rules/ \
        2>&1 | tee -a "$LOG_FILE" || \
        log "Rule validation failed"
}

# === Main Loop ===
while true; do
    CURRENT_HOUR=$(date +%H)
    CURRENT_MIN=$(date +%M)

    # Probe every 5 minutes
    if (( CURRENT_MIN % 5 == 0 )); then
        run_probes
    fi

    # Check config and generate report hourly
    if (( CURRENT_MIN == 0 )); then
        check_prometheus
        generate_report
    fi

    # Check SSL daily at 9:00
    if [[ "$CURRENT_HOUR" == "09" && "$CURRENT_MIN" == "00" ]]; then
        check_ssl
    fi

    sleep 60
done

6.2 systemd Timer Configuration

# /etc/systemd/system/prom-inspection.service
[Unit]
Description=Prometheus Automated Inspection
After=network.target

[Service]
Type=oneshot
ExecStart=/opt/scripts/inspection/run_inspection.sh
User=monitor
Group=monitor
Environment=PROMETHEUS_URL=http://prometheus:9090
Environment=REPORT_DIR=/var/lib/inspection/reports
TimeoutStartSec=300

# /etc/systemd/system/prom-inspection.timer
[Unit]
Description=Run Prometheus Inspection Hourly

[Timer]
OnCalendar=hourly
Persistent=true

[Install]
WantedBy=timers.target

# Enable the timer
systemctl daemon-reload
systemctl enable --now prom-inspection.timer

# View next execution time
systemctl list-timers prom-inspection.timer

# Trigger manually
systemctl start prom-inspection.service

Summary

The reliability of a monitoring system can’t be guaranteed by “looking at it when something breaks” — you need an active inspection mechanism that continuously validates. The toolkit built in this article covers the core dimensions of inspection:

  1. Probing is the most direct availability check: Through Blackbox Exporter or direct HTTP/TCP probing, continuously verify whether service endpoints are reachable and responding normally. Parallel probing + threshold judgment — hundreds of endpoints checked in 10 seconds
  2. SSL certificate expiry is a frequent incident source: Batch check all domain certificates, classify alerts by WARNING (30 days)/CRITICAL (7 days). Auto-send email notifications to avoid service outages from expired certificates
  3. Configuration drift needs continuous detection: Compare on-disk config with runtime config, check target health status, verify rule loading count, monitor TSDB series count. Any anomaly in these dimensions can create monitoring blind spots
  4. Alerting rules need testing: Use promtool test rules to construct test data and verify alert behavior under normal and abnormal scenarios. Deploying rule changes without tests is flying blind
  5. Reports should be automated: Aggregate all check results into an HTML report, generate on schedule, retain history, provide a latest link. Let ops staff get a full picture by opening one page every day
  6. Scheduling must be reliable: Use systemd timer instead of cron for better log integration and failure retry capabilities. The execution status of inspection scripts themselves must also be monitored

The essence of inspection is “monitoring the monitoring system” — bringing every link in the monitoring chain (collection → storage → rules → alerting → notification) under inspection scope, ensuring the monitoring itself is reliable. When the inspection script reports “all clear,” you can truly trust that the data you see in Prometheus is accurate.

References & Acknowledgments

This article referenced the following materials during writing. We thank the original authors for their contributions:

  1. Prometheus Official Documentation — Prometheus Authors, referenced for Prometheus Official Documentation
  2. Blackbox Exporter — GitHub, referenced for Blackbox Exporter