mirror of
https://github.com/taylanbakircioglu/haproxy-openmanager.git
synced 2026-09-17 16:15:20 +00:00
taylanbakircioglu
269 lines
9.1 KiB
Python
269 lines
9.1 KiB
Python
"""
|
|
Production-Ready Logging Configuration
|
|
Provides structured logging, correlation IDs, and comprehensive error tracking
|
|
"""
|
|
|
|
import logging
|
|
import sys
|
|
import json
|
|
import uuid
|
|
import time
|
|
from datetime import datetime
|
|
from typing import Optional, Dict, Any
|
|
from contextvars import ContextVar
|
|
from functools import wraps
|
|
|
|
# Context variable for correlation ID tracking across requests
|
|
correlation_id_context: ContextVar[Optional[str]] = ContextVar('correlation_id', default=None)
|
|
|
|
class StructuredFormatter(logging.Formatter):
|
|
"""Custom formatter for structured JSON logging in production"""
|
|
|
|
def format(self, record: logging.LogRecord) -> str:
|
|
# Create base log structure
|
|
log_entry = {
|
|
"timestamp": datetime.utcnow().isoformat() + "Z",
|
|
"level": record.levelname,
|
|
"logger": record.name,
|
|
"message": record.getMessage(),
|
|
"module": record.module,
|
|
"function": record.funcName,
|
|
"line": record.lineno,
|
|
}
|
|
|
|
# Add correlation ID if available
|
|
correlation_id = correlation_id_context.get()
|
|
if correlation_id:
|
|
log_entry["correlation_id"] = correlation_id
|
|
|
|
# Add process and thread info
|
|
log_entry["process_id"] = record.process
|
|
log_entry["thread_id"] = record.thread
|
|
|
|
# Add exception info if present
|
|
if record.exc_info:
|
|
log_entry["exception"] = {
|
|
"type": record.exc_info[0].__name__ if record.exc_info[0] else None,
|
|
"message": str(record.exc_info[1]) if record.exc_info[1] else None,
|
|
"traceback": self.formatException(record.exc_info)
|
|
}
|
|
|
|
# Add extra fields from record
|
|
if hasattr(record, 'extra_fields'):
|
|
log_entry.update(record.extra_fields)
|
|
|
|
# Add performance metrics if available
|
|
if hasattr(record, 'duration_ms'):
|
|
log_entry["duration_ms"] = record.duration_ms
|
|
|
|
if hasattr(record, 'status_code'):
|
|
log_entry["status_code"] = record.status_code
|
|
|
|
if hasattr(record, 'method'):
|
|
log_entry["http_method"] = record.method
|
|
|
|
if hasattr(record, 'path'):
|
|
log_entry["http_path"] = record.path
|
|
|
|
if hasattr(record, 'client_ip'):
|
|
log_entry["client_ip"] = record.client_ip
|
|
|
|
return json.dumps(log_entry, ensure_ascii=False, separators=(',', ':'))
|
|
|
|
def setup_production_logging(log_level: str = "INFO"):
|
|
"""Setup production-ready structured logging"""
|
|
|
|
# Create structured formatter
|
|
formatter = StructuredFormatter()
|
|
|
|
# Setup root logger
|
|
root_logger = logging.getLogger()
|
|
root_logger.setLevel(getattr(logging, log_level.upper()))
|
|
|
|
# Remove default handlers
|
|
for handler in root_logger.handlers[:]:
|
|
root_logger.removeHandler(handler)
|
|
|
|
# Add structured console handler
|
|
console_handler = logging.StreamHandler(sys.stdout)
|
|
console_handler.setFormatter(formatter)
|
|
root_logger.addHandler(console_handler)
|
|
|
|
# Setup specific loggers with appropriate levels
|
|
logging.getLogger("uvicorn.access").setLevel(logging.WARNING) # Reduce uvicorn noise
|
|
logging.getLogger("asyncpg").setLevel(logging.WARNING) # Reduce database noise
|
|
|
|
# Application loggers
|
|
app_logger = logging.getLogger("haproxy_openmanager")
|
|
app_logger.setLevel(getattr(logging, log_level.upper()))
|
|
|
|
return app_logger
|
|
|
|
def get_correlation_id() -> str:
|
|
"""Get or create correlation ID for request tracking"""
|
|
correlation_id = correlation_id_context.get()
|
|
if not correlation_id:
|
|
correlation_id = str(uuid.uuid4())[:8] # Short UUID for logs
|
|
correlation_id_context.set(correlation_id)
|
|
return correlation_id
|
|
|
|
def log_with_correlation(logger: logging.Logger, level: str, message: str, **extra_fields):
|
|
"""Log message with correlation ID and extra fields"""
|
|
correlation_id = get_correlation_id()
|
|
|
|
# Create LogRecord with extra fields
|
|
record = logger.makeRecord(
|
|
logger.name, getattr(logging, level.upper()),
|
|
__file__, 0, message, (), None
|
|
)
|
|
|
|
# Add extra fields to record
|
|
if extra_fields:
|
|
record.extra_fields = extra_fields
|
|
|
|
logger.handle(record)
|
|
|
|
class PerformanceLogger:
|
|
"""Context manager for performance logging"""
|
|
|
|
def __init__(self, logger: logging.Logger, operation: str, **context):
|
|
self.logger = logger
|
|
self.operation = operation
|
|
self.context = context
|
|
self.start_time = None
|
|
|
|
def __enter__(self):
|
|
self.start_time = time.time()
|
|
log_with_correlation(
|
|
self.logger, "INFO",
|
|
f"Starting operation: {self.operation}",
|
|
**self.context
|
|
)
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
|
duration_ms = round((time.time() - self.start_time) * 1000, 2)
|
|
|
|
if exc_type:
|
|
log_with_correlation(
|
|
self.logger, "ERROR",
|
|
f"Operation failed: {self.operation}",
|
|
duration_ms=duration_ms,
|
|
error_type=exc_type.__name__,
|
|
error_message=str(exc_val),
|
|
**self.context
|
|
)
|
|
else:
|
|
log_with_correlation(
|
|
self.logger, "INFO",
|
|
f"Operation completed: {self.operation}",
|
|
duration_ms=duration_ms,
|
|
**self.context
|
|
)
|
|
|
|
def log_function_performance(logger: logging.Logger):
|
|
"""Decorator for automatic function performance logging"""
|
|
def decorator(func):
|
|
@wraps(func)
|
|
async def async_wrapper(*args, **kwargs):
|
|
operation = f"{func.__module__}.{func.__name__}"
|
|
with PerformanceLogger(logger, operation, function=func.__name__):
|
|
return await func(*args, **kwargs)
|
|
|
|
@wraps(func)
|
|
def sync_wrapper(*args, **kwargs):
|
|
operation = f"{func.__module__}.{func.__name__}"
|
|
with PerformanceLogger(logger, operation, function=func.__name__):
|
|
return func(*args, **kwargs)
|
|
|
|
return async_wrapper if hasattr(func, '__code__') and func.__code__.co_flags & 0x80 else sync_wrapper
|
|
|
|
return decorator
|
|
|
|
class ErrorTracker:
|
|
"""Track and categorize application errors"""
|
|
|
|
def __init__(self, logger: logging.Logger):
|
|
self.logger = logger
|
|
self.error_counts = {}
|
|
|
|
def track_error(self, error: Exception, context: Dict[str, Any] = None):
|
|
"""Track error occurrence with context"""
|
|
error_type = type(error).__name__
|
|
error_message = str(error)
|
|
|
|
# Increment error count
|
|
self.error_counts[error_type] = self.error_counts.get(error_type, 0) + 1
|
|
|
|
# Log structured error
|
|
log_with_correlation(
|
|
self.logger, "ERROR",
|
|
f"Application error: {error_message}",
|
|
error_type=error_type,
|
|
error_count=self.error_counts[error_type],
|
|
**(context or {})
|
|
)
|
|
|
|
def get_error_summary(self) -> Dict[str, int]:
|
|
"""Get error count summary"""
|
|
return self.error_counts.copy()
|
|
|
|
# Global error tracker instance
|
|
_error_tracker = None
|
|
|
|
def get_error_tracker(logger: logging.Logger = None) -> ErrorTracker:
|
|
"""Get global error tracker instance"""
|
|
global _error_tracker
|
|
if _error_tracker is None:
|
|
if logger is None:
|
|
logger = logging.getLogger("haproxy_openmanager")
|
|
_error_tracker = ErrorTracker(logger)
|
|
return _error_tracker
|
|
|
|
# Utility functions for common logging patterns
|
|
def log_api_request(logger: logging.Logger, method: str, path: str, client_ip: str, **extra):
|
|
"""Log API request with structured data"""
|
|
log_with_correlation(
|
|
logger, "INFO",
|
|
f"API Request: {method} {path}",
|
|
method=method,
|
|
path=path,
|
|
client_ip=client_ip,
|
|
**extra
|
|
)
|
|
|
|
def log_api_response(logger: logging.Logger, method: str, path: str, status_code: int, duration_ms: float, **extra):
|
|
"""Log API response with performance metrics"""
|
|
level = "ERROR" if status_code >= 500 else "WARNING" if status_code >= 400 else "INFO"
|
|
|
|
log_with_correlation(
|
|
logger, level,
|
|
f"API Response: {method} {path} - {status_code}",
|
|
method=method,
|
|
path=path,
|
|
status_code=status_code,
|
|
duration_ms=duration_ms,
|
|
**extra
|
|
)
|
|
|
|
def log_database_operation(logger: logging.Logger, operation: str, table: str, duration_ms: float, **extra):
|
|
"""Log database operation with performance metrics"""
|
|
log_with_correlation(
|
|
logger, "DEBUG",
|
|
f"Database operation: {operation} on {table}",
|
|
db_operation=operation,
|
|
db_table=table,
|
|
duration_ms=duration_ms,
|
|
**extra
|
|
)
|
|
|
|
def log_agent_interaction(logger: logging.Logger, agent_id: str, operation: str, result: str, **extra):
|
|
"""Log agent interaction with structured data"""
|
|
log_with_correlation(
|
|
logger, "INFO",
|
|
f"Agent interaction: {operation} for agent {agent_id} - {result}",
|
|
agent_id=agent_id,
|
|
agent_operation=operation,
|
|
result=result,
|
|
**extra
|
|
) |