Как отслеживать работоспособность программы?

Как отслеживать работоспособность программы?

Отслеживание здоровья приложения — критическое умение для production. За 10+ лет я разработал комплексный подход.

1. Логирование как основа

Все начинается с хорошего логирования:

import logging
import json
from datetime import datetime

# Настройка логирования
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)

logger = logging.getLogger(__name__)

# Пример использования
def process_order(order_id: int):
    logger.info(f"Processing order {order_id}")
    
    try:
        # основной код
        order = fetch_order(order_id)
        validate_order(order)
        save_order(order)
        logger.info(f"Order {order_id} processed successfully")
    except Exception as e:
        logger.error(f"Failed to process order {order_id}", exc_info=True)
        raise

2. Структурированное логирование

Лучше использовать JSON логи для удобного парсинга:

from pythonjsonlogger import jsonlogger
import logging
import sys

# JSON логирование
logHandler = logging.StreamHandler(sys.stdout)
formatter = jsonlogger.JsonFormatter()
logHandler.setFormatter(formatter)

logger = logging.getLogger()
logger.addHandler(logHandler)
logger.setLevel(logging.INFO)

# Использование
logger.info(
    "Processing started",
    extra={
        "order_id": 12345,
        "user_id": 999,
        "timestamp": datetime.utcnow().isoformat(),
        "status": "processing"
    }
)

3. Система health checks

Программа должна иметь эндпоинты для проверки здоровья:

from fastapi import FastAPI
import asyncio
from datetime import datetime

app = FastAPI()

@app.get("/health")
async def simple_health():
    return {
        "status": "ok",
        "timestamp": datetime.utcnow().isoformat()
    }

@app.get("/health/deep")
async def deep_health():
    checks = {
        "database": await check_database_health(),
        "cache": await check_cache_health(),
        "external_api": await check_external_api(),
    }
    
    all_healthy = all(checks.values())
    
    return {
        "status": "healthy" if all_healthy else "degraded",
        "checks": checks,
        "timestamp": datetime.utcnow().isoformat()
    }

4. Мониторинг использования ресурсов

import psutil

def monitor_resources():
    # CPU
    cpu_percent = psutil.cpu_percent(interval=1)
    print(f"CPU: {cpu_percent}%")
    
    # Память
    memory = psutil.virtual_memory()
    print(f"Memory: {memory.percent}% used")
    
    # Диск
    disk = psutil.disk_usage("/")
    print(f"Disk: {disk.percent}% used")

5. Профилирование производительности

import time
from functools import wraps

def measure_time(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        
        logger.info(
            f"Function {func.__name__} took {elapsed:.3f}s",
            extra={"function": func.__name__, "duration_ms": elapsed * 1000}
        )
        return result
    return wrapper

6. Отслеживание ошибок

def safe_function(data):
    try:
        result = risky_operation(data)
        logger.info("Operation successful")
        return result
    except ValueError as e:
        logger.warning(f"Validation error: {e}")
        return None
    except Exception as e:
        logger.error(
            f"Unexpected error",
            exc_info=True,
            extra={"error_type": type(e).__name__}
        )
        raise

7. Prometheus метрики

from prometheus_client import Counter, Histogram, Gauge

app_errors = Counter(
    "app_errors_total",
    "Total application errors",
    ["error_type"]
)

request_duration = Histogram(
    "request_duration_seconds",
    "Request duration"
)

active_connections = Gauge(
    "active_connections",
    "Active connections"
)

8. Алерты и уведомления

import smtplib
from email.mime.text import MIMEText

def send_alert(subject: str, message: str):
    msg = MIMEText(message)
    msg["Subject"] = subject
    msg["From"] = "monitoring@example.com"
    msg["To"] = "devops@example.com"
    
    with smtplib.SMTP("localhost") as server:
        server.send_message(msg)

if memory_usage > 90:
    send_alert("Memory critical", f"Memory at {memory_usage}%")

9. Инструменты мониторинга

Системные утилиты:

• top — процессы в реальном времени
• htop — удобный top
• df -h — использование диска
• ps aux — список процессов
• netstat -an — сетевые подключения

10. Пример скрипта мониторинга

import psutil
import time
from datetime import datetime
import requests

def monitor_app():
    while True:
        timestamp = datetime.now().isoformat()
        
        cpu = psutil.cpu_percent(interval=1)
        mem = psutil.virtual_memory()
        
        try:
            resp = requests.get("http://localhost:8000/health", timeout=2)
            health_status = "ok" if resp.status_code == 200 else "error"
        except Exception:
            health_status = "unreachable"
        
        print(f"{timestamp} | CPU: {cpu:5.1f}% | MEM: {mem.percent:5.1f}% | Health: {health_status}")
        
        if cpu > 80:
            print(f"  WARNING: High CPU usage: {cpu}%")
        if mem.percent > 85:
            print(f"  WARNING: High memory usage: {mem.percent}%")
        
        time.sleep(5)

Чек-лист отслеживания

• Логирование — все важные события логируются
• Структурированные логи — JSON формат
• Health checks — простой и глубокий endpoints
• Метрики — Prometheus для сбора данных
• Дашборды — Grafana для визуализации
• Алерты — автоматические уведомления
• Мониторинг ресурсов — CPU, память, диск
• Трассировка — отслеживание запросов
• Централизованное хранилище — ELK, Datadog
• Историчность — просмотр прошлых данных

Хорошее отслеживание — это разница между быстрым fixing инцидента и часами отладки.