Как разобраться в проблеме, если все логи смешаны в Logstash?

Анализ смешанных логов в Logstash

Это частая проблема в микросервисной архитектуре, когда логи от разных сервисов, потоков и запросов смешиваются в едином хранилище. Я применяю структурированный подход для отладки.

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

Первым шагом должно быть внедрение структурированного логирования — логи в JSON формате с контекстными полями.

import logging
import json
from pythonjsonlogger import jsonlogger
import uuid
from contextvars import ContextVar

request_id_var: ContextVar[str] = ContextVar("request_id", default=None)
trace_id_var: ContextVar[str] = ContextVar("trace_id", default=None)

class StructuredLogFormatter(jsonlogger.JsonFormatter):
    def add_fields(self, log_record, record, message_dict):
        super().add_fields(log_record, record, message_dict)
        log_record["request_id"] = request_id_var.get()
        log_record["trace_id"] = trace_id_var.get()
        log_record["service_name"] = "my-service"

2. Middleware для трассировки

@app.middleware("http")
async def add_request_context(request: Request, call_next):
    request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
    trace_id = request.headers.get("X-Trace-ID", str(uuid.uuid4()))
    
    request_id_var.set(request_id)
    trace_id_var.set(trace_id)
    
    response = await call_next(request)
    response.headers["X-Request-ID"] = request_id
    response.headers["X-Trace-ID"] = trace_id
    return response

3. Kibana поиск

service_name: "auth-service" AND level: ERROR
user_id: "user-123" AND timestamp >= 2024-01-01T00:00:00Z
duration_ms: >= 1000 AND path: "/api/*"

4. Скрипт анализа логов

import json
from collections import defaultdict

def analyze_logs(log_file: str, request_id: str = None):
    logs_by_service = defaultdict(list)
    errors = []
    
    with open(log_file, "r") as f:
        for line in f:
            try:
                log = json.loads(line)
            except json.JSONDecodeError:
                continue
            
            if request_id and log.get("request_id") != request_id:
                continue
            
            service = log.get("service_name", "unknown")
            logs_by_service[service].append(log)
            
            if log.get("level") == "ERROR":
                errors.append(log)
    
    return logs_by_service, errors

Лучшие практики

1. request_id в каждом логе — связывает логи одного запроса 2. trace_id для распределённых систем — отслеживает микросервисы 3. Структурированный JSON — упрощает парсинг и поиск 4. Контекст сервиса — группирует логи по источникам 5. Метаданные — duration_ms, user_id, status_code для анализа

Инструменты

grep "request_id" /var/log/app.log | jq "."
jq -s "sort_by(.timestamp)" /var/log/app.log
jq "select(.level == \"ERROR\")" /var/log/app.log

Структурированные логи с правильной трассировкой превращают хаос в упорядоченную диагностику.