Автоматизация DevOps на Python: скрипты для CI/CD и мониторинга

#!/usr/bin/env python3
"""
Скрипт для проверки доступности сервисов и отправки уведомлений.
"""
import requests
import smtplib
from email.mime.text import MIMEText
from datetime import datetime
import time
import json
from typing import Dict, List
import logging
from dataclasses import dataclass

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@dataclass
class Service:
    name: str
    url: str
    method: str = "GET"
    timeout: int = 5
    expected_status: int = 200
    check_ssl: bool = True

class ServiceMonitor:
    def __init__(self, services: List[Service]):
        self.services = services
        self.results = []
        
    def check_service(self, service: Service) -> Dict:
        """Проверяет доступность одного сервиса."""
        start_time = time.time()
        
        try:
            response = requests.request(
                method=service.method,
                url=service.url,
                timeout=service.timeout,
                verify=service.check_ssl
            )
            elapsed = time.time() - start_time
            
            is_ok = response.status_code == service.expected_status
            status = "UP" if is_ok else "DOWN"
            
            result = {
                "service": service.name,
                "status": status,
                "status_code": response.status_code,
                "response_time": round(elapsed * 1000, 2),  # мс
                "timestamp": datetime.now().isoformat(),
                "error": None
            }
            
            if not is_ok:
                result["error"] = f"Expected {service.expected_status}, got {response.status_code}"
                
        except Exception as e:
            elapsed = time.time() - start_time
            result = {
                "service": service.name,
                "status": "DOWN",
                "status_code": None,
                "response_time": round(elapsed * 1000, 2),
                "timestamp": datetime.now().isoformat(),
                "error": str(e)
            }
        
        return result
    
    def run_checks(self):
        """Запускает проверку всех сервисов."""
        logger.info(f"Запуск проверки {len(self.services)} сервисов...")
        
        for service in self.services:
            result = self.check_service(service)
            self.results.append(result)
            
            if result["status"] == "DOWN":
                logger.error(f"❌ {service.name}: {result['error']}")
                self.send_alert(result)
            else:
                logger.info(f"✅ {service.name}: {result['response_time']}ms")
    
    def send_alert(self, result: Dict):
        """Отправляет уведомление о проблеме."""
        # 1. Email
        self._send_email_alert(result)
        
        # 2. Slack (альтернатива)
        # self._send_slack_alert(result)
        
        # 3. Telegram (альтернатива)
        # self._send_telegram_alert(result)
    
    def _send_email_alert(self, result: Dict):
        """Отправляет алерт по email."""
        subject = f"🚨 ALERT: {result['service']} is DOWN"
        body = f"""
        Service: {result['service']}
        Status: {result['status']}
        Error: {result['error']}
        Time: {result['timestamp']}
        
        Check URL: {next(s.url for s in self.services if s.name == result['service'])}
        """
        
        msg = MIMEText(body)
        msg['Subject'] = subject
        msg['From'] = "monitor@example.com"
        msg['To'] = "devops@example.com"
        
        try:
            with smtplib.SMTP('smtp.example.com', 587) as server:
                server.starttls()
                server.login("user", "password")
                server.send_message(msg)
            logger.info(f"📧 Алерт отправлен для {result['service']}")
        except Exception as e:
            logger.error(f"Не удалось отправить email: {e}")
    
    def generate_report(self) -> str:
        """Генерирует HTML-отчет."""
        up_count = sum(1 for r in self.results if r["status"] == "UP")
        down_count = len(self.results) - up_count
        
        html = f"""
        <html>
        <head><title>Service Health Report</title></head>
        <body>
            <h1>📊 Service Health Report</h1>
            <p>Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>
            
            <div style="margin: 20px 0;">
                <span style="background: {'#4CAF50' if down_count == 0 else '#f44336'}; 
                           color: white; padding: 10px; border-radius: 5px;">
                    ✅ UP: {up_count} | ❌ DOWN: {down_count}
                </span>
            </div>
            
            <table border="1" cellpadding="10" style="border-collapse: collapse;">
                <tr>
                    <th>Service</th><th>Status</th><th>Response Time</th><th>Details</th>
                </tr>
        """
        
        for result in self.results:
            color = "#4CAF50" if result["status"] == "UP" else "#f44336"
            html += f"""
                <tr>
                    <td>{result['service']}</td>
                    <td style="background: {color}; color: white;">{result['status']}</td>
                    <td>{result['response_time']} ms</td>
                    <td>{result['error'] or 'OK'}</td>
                </tr>
            """
        
        html += "</table></body></html>"
        return html

# Конфигурация сервисов для мониторинга
SERVICES = [
    Service("Main API", "https://api.example.com/health/", expected_status=200),
    Service("Database", "https://db.example.com/ping/", timeout=10),
    Service("Cache", "http://cache.example.com:6379/", method="GET", expected_status=200),
    Service("Authentication", "https://auth.example.com/health/"),
    Service("Frontend", "https://example.com/", timeout=8),
]

if __name__ == "__main__":
    # Запуск как демона с интервалом
    import schedule
    
    monitor = ServiceMonitor(SERVICES)
    
    def job():
        monitor.run_checks()
        # Сохраняем отчет в файл
        report = monitor.generate_report()
        with open("/var/log/service_monitor/report.html", "w") as f:
            f.write(report)
    
    # Запускаем каждые 5 минут
    schedule.every(5).minutes.do(job)
    
    # Первый запуск сразу
    job()
    
    # Бесконечный цикл
    while True:
        schedule.run_pending()
        time.sleep(1)

2.2. Мониторинг метрик приложения и отправка в Prometheus

# metrics_exporter.py
from prometheus_client import start_http_server, Gauge, Counter, Histogram
import psutil
import time
import requests
from typing import Dict
import threading

class ApplicationMetrics:
    def __init__(self, port=8000):
        self.port = port
        
        # Метрики системы
        self.cpu_usage = Gauge('app_cpu_usage_percent', 'CPU usage percentage')
        self.memory_usage = Gauge('app_memory_usage_bytes', 'Memory usage in bytes')
        self.disk_usage = Gauge('app_disk_usage_percent', 'Disk usage percentage')
        
        # Метрики приложения
        self.request_count = Counter('app_requests_total', 'Total HTTP requests')
        self.error_count = Counter('app_errors_total', 'Total application errors')
        self.response_time = Histogram('app_response_time_seconds', 'Request response time')
        
        # Бизнес-метрики
        self.active_users = Gauge('app_active_users', 'Number of active users')
        self.queue_size = Gauge('app_queue_size', 'Size of processing queue')
        
    def collect_system_metrics(self):
        """Сбор системных метрик."""
        self.cpu_usage.set(psutil.cpu_percent())
        
        memory = psutil.virtual_memory()
        self.memory_usage.set(memory.used)
        
        disk = psutil.disk_usage('/')
        self.disk_usage.set(disk.percent)
    
    def record_request(self, duration: float, status_code: int):
        """Запись метрик HTTP-запроса."""
        self.request_count.inc()
        self.response_time.observe(duration)
        
        if status_code >= 500:
            self.error_count.inc()
    
    def update_business_metrics(self, users: int, queue: int):
        """Обновление бизнес-метрик."""
        self.active_users.set(users)
        self.queue_size.set(queue)
    
    def run_metrics_server(self):
        """Запуск HTTP-сервера для сбора метрик Prometheus."""
        start_http_server(self.port)
        print(f"📊 Metrics server started on port {self.port}")
        
        while True:
            self.collect_system_metrics()
            time.sleep(5)  # Обновляем каждые 5 секунд

# Middleware для FastAPI/Django для сбора метрик
def metrics_middleware(app):
    """Middleware для сбора метрик HTTP-запросов."""
    metrics = ApplicationMetrics()
    
    # Запускаем сбор системных метрик в отдельном потоке
    threading.Thread(target=metrics.run_metrics_server, daemon=True).start()
    
    async def middleware(request, call_next):
        start_time = time.time()
        
        try:
            response = await call_next(request)
            duration = time.time() - start_time
            
            # Записываем метрики
            metrics.record_request(duration, response.status_code)
            
            return response
        except Exception as e:
            duration = time.time() - start_time
            metrics.record_request(duration, 500)
            raise e
    
    return middleware

# Пример использования с FastAPI
from fastapi import FastAPI, Request

app = FastAPI()
app.middleware("http")(metrics_middleware(app))

@app.get("/metrics/stats")
async def get_stats():
    """Эндпоинт для получения текущих метрик (дополнительно к Prometheus)."""
    return {
        "system": {
            "cpu": psutil.cpu_percent(),
            "memory": psutil.virtual_memory().percent,
            "disk": psutil.disk_usage('/').percent
        }
    }

2.3. Скрипт для мониторинга логов и обнаружения аномалий

# log_monitor.py
import re
from pathlib import Path
import time
from collections import defaultdict, deque
import json
from datetime import datetime, timedelta

class LogMonitor:
    def __init__(self, log_path: Path, patterns: Dict[str, str]):
        self.log_path = log_path
        self.patterns = {name: re.compile(pattern) for name, pattern in patterns.items()}
        self.stats = defaultdict(int)
        self.alert_thresholds = {
            "ERROR": 10,  # Более 10 ошибок в минуту
            "FAILED_LOGIN": 5,  # Более 5 неудачных попыток входа в минуту
        }
        self.recent_events = deque(maxlen=1000)  # Кольцевой буфер последних событий
        
    def tail_log(self):
        """Чтение лога в реальном времени (аналог tail -f)."""
        with open(self.log_path, 'r') as f:
            # Перемещаемся в конец файла
            f.seek(0, 2)
            
            while True:
                line = f.readline()
                if line:
                    yield line
                else:
                    time.sleep(0.1)  # Пауза при отсутствии новых данных
    
    def parse_line(self, line: str) -> Dict:
        """Парсит строку лога и извлекает информацию."""
        timestamp_match = re.search(r'\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}', line)
        timestamp = datetime.now()
        
        if timestamp_match:
            try:
                timestamp = datetime.strptime(timestamp_match.group(), '%Y-%m-%d %H:%M:%S')
            except:
                pass
        
        event_type = "UNKNOWN"
        details = {}
        
        # Проверяем все паттерны
        for name, pattern in self.patterns.items():
            match = pattern.search(line)
            if match:
                event_type = name
                details = match.groupdict()
                break
        
        return {
            "timestamp": timestamp,
            "event_type": event_type,
            "raw_line": line.strip(),
            "details": details
        }
    
    def analyze_events(self):
        """Анализирует события и генерирует алерты."""
        current_time = datetime.now()
        one_minute_ago = current_time - timedelta(minutes=1)
        
        # Фильтруем события за последнюю минуту
        recent = [e for e in self.recent_events if e["timestamp"] > one_minute_ago]
        
        # Считаем статистику
        minute_stats = defaultdict(int)
        for event in recent:
            minute_stats[event["event_type"]] += 1
        
        # Проверяем пороги
        alerts = []
        for event_type, threshold in self.alert_thresholds.items():
            if minute_stats[event_type] > threshold:
                alerts.append({
                    "type": "THRESHOLD_EXCEEDED",
                    "event": event_type,
                    "count": minute_stats[event_type],
                    "threshold": threshold,
                    "period": "1 minute"
                })
        
        # Поиск паттернов атак (пример)
        failed_logins = [e for e in recent if e["event_type"] == "FAILED_LOGIN"]
        if len(failed_logins) >= 3:
            # Проверяем, идут ли неудачные попытки с одного IP
            ips = [e["details"].get("ip") for e in failed_logins if e["details"].get("ip")]
            if len(set(ips)) == 1 and ips[0]:
                alerts.append({
                    "type": "BRUTE_FORCE_ATTEMPT",
                    "ip": ips[0],
                    "attempts": len(failed_logins),
                    "period": "1 minute"
                })
        
        return alerts
    
    def run(self):
        """Основной цикл мониторинга."""
        print(f"🔍 Начинаем мониторинг лога {self.log_path}")
        
        for line in self.tail_log():
            event = self.parse_line(line)
            self.recent_events.append(event)
            
            # Обновляем общую статистику
            self.stats[event["event_type"]] += 1
            
            # Немедленный алерт для критических ошибок
            if event["event_type"] in ["CRITICAL", "FATAL"]:
                print(f"🚨 НЕМЕДЛЕННЫЙ АЛЕРТ: {event['raw_line']}")
            
            # Периодический анализ
            if len(self.recent_events) % 100 == 0:  # Каждые 100 событий
                alerts = self.analyze_events()
                if alerts:
                    for alert in alerts:
                        print(f"⚠️  АЛЕРТ: {json.dumps(alert, indent=2, default=str)}")

# Паттерны для поиска в логах
LOG_PATTERNS = {
    "ERROR": r'ERROR\s+(?P<message>.*)',
    "WARNING": r'WARNING\s+(?P<message>.*)',
    "FAILED_LOGIN": r'Failed login for user (?P<user>\w+) from (?P<ip>[\d.]+)',
    "SUCCESSFUL_LOGIN": r'Successful login for user (?P<user>\w+) from (?P<ip>[\d.]+)',
    "SQL_INJECTION": r'(SELECT|INSERT|UPDATE|DELETE|DROP|UNION).*FROM.*WHERE.*=.*[\'"]',
    "CRITICAL": r'CRITICAL\s+(?P<component>\w+):(?P<message>.*)',
}

if __name__ == "__main__":
    monitor = LogMonitor(
        log_path=Path("/var/log/myapp/app.log"),
        patterns=LOG_PATTERNS
    )
    monitor.run()

Вывод

Python предоставляет DevOps-инженерам и разработчикам мощный инструментарий для автоматизации рутинных задач.