# -*- coding: utf-8 -*- """ Monitoring Services for Stream Processor Comprehensive monitoring services including: - Performance metrics collection - Structured logging - Alert management - Health check services """ import logging import time import psutil import json from typing import Dict, Any from datetime import datetime from contextlib import contextmanager from functools import wraps from django.utils import timezone from django.core.cache import cache from django.conf import settings from django.db import connection # Set up specialized loggers performance_logger = logging.getLogger('stream_processor.monitoring') security_logger = logging.getLogger('apps.core.authentication') stream_logger = logging.getLogger('stream_processor.tasks') class PerformanceMonitor: """ Performance monitoring service for tracking system and application metrics. """ def __init__(self): self.logger = performance_logger @contextmanager def measure_time(self, operation: str, **kwargs): """ Context manager to measure operation execution time. """ start_time = time.time() start_memory = psutil.Process().memory_info().rss try: yield finally: end_time = time.time() end_memory = psutil.Process().memory_info().rss duration = end_time - start_time memory_delta = end_memory - start_memory self.log_performance_metric( operation=operation, duration=duration, memory_delta=memory_delta, **kwargs ) def log_performance_metric(self, operation: str, duration: float, memory_delta: int = 0, **kwargs): """ Log a performance metric with structured data. """ metric_data = { 'timestamp': timezone.now().isoformat(), 'operation': operation, 'duration_seconds': round(duration, 4), 'memory_delta_mb': round(memory_delta / 1024 / 1024, 2), **kwargs } self.logger.info(json.dumps(metric_data)) def log_system_metrics(self): """ Collect and log current system metrics. """ try: cpu_percent = psutil.cpu_percent(interval=1) memory = psutil.virtual_memory() disk = psutil.disk_usage('/') network = psutil.net_io_counters() system_metrics = { 'timestamp': timezone.now().isoformat(), 'cpu_percent': cpu_percent, 'memory_percent': memory.percent, 'disk_percent': round((disk.used / disk.total) * 100, 2), 'network_bytes_sent': network.bytes_sent, 'network_bytes_recv': network.bytes_recv, } self.logger.info(json.dumps(system_metrics)) cache.set('system_metrics', system_metrics, timeout=300) return system_metrics except Exception as e: self.logger.error(f"Failed to collect system metrics: {e}") return None class SecurityMonitor: """ Security monitoring service for tracking authentication and security events. """ def __init__(self): self.logger = security_logger def log_login_attempt(self, username: str, ip_address: str, user_agent: str, success: bool, failure_reason: str = '', **kwargs): """ Log authentication attempt with security context. """ auth_data = { 'timestamp': timezone.now().isoformat(), 'event_type': 'login_attempt', 'username': username, 'ip_address': ip_address, 'user_agent': user_agent, 'success': success, 'failure_reason': failure_reason, **kwargs } if success: self.logger.info(f"Successful login: {json.dumps(auth_data)}") else: self.logger.warning(f"Failed login attempt: {json.dumps(auth_data)}") def log_security_violation(self, violation_type: str, details: Dict[str, Any]): """ Log security violations and suspicious activities. """ violation_data = { 'timestamp': timezone.now().isoformat(), 'event_type': 'security_violation', 'violation_type': violation_type, 'details': details } self.logger.error(f"Security violation: {json.dumps(violation_data)}") class StreamMonitor: """ Stream processing monitoring service. """ def __init__(self): self.logger = stream_logger def log_stream_event(self, event_type: str, channel_name: str, session_id: str = None, **kwargs): """ Log stream processing events. """ stream_data = { 'timestamp': timezone.now().isoformat(), 'event_type': event_type, 'channel_name': channel_name, 'session_id': session_id, **kwargs } self.logger.info(json.dumps(stream_data)) def monitor_performance(operation: str): """ Decorator to monitor function performance. """ def decorator(func): @wraps(func) def wrapper(*args, **kwargs): monitor = PerformanceMonitor() with monitor.measure_time(operation, function=func.__name__): return func(*args, **kwargs) return wrapper return decorator class HealthChecker: """ System health checking service. """ def __init__(self): self.logger = performance_logger def check_database_health(self) -> Dict[str, Any]: """ Check database connectivity and performance. """ try: start_time = time.time() with connection.cursor() as cursor: cursor.execute("SELECT 1") cursor.fetchone() response_time = time.time() - start_time return { 'status': 'healthy', 'response_time_ms': round(response_time * 1000, 2), 'timestamp': timezone.now().isoformat() } except Exception as e: self.logger.error(f"Database health check failed: {e}") return { 'status': 'unhealthy', 'error': str(e), 'timestamp': timezone.now().isoformat() } def check_redis_health(self) -> Dict[str, Any]: """ Check Redis connectivity and performance. """ try: start_time = time.time() cache.set('health_check', 'test', timeout=10) result = cache.get('health_check') if result != 'test': raise Exception("Redis read/write test failed") response_time = time.time() - start_time return { 'status': 'healthy', 'response_time_ms': round(response_time * 1000, 2), 'timestamp': timezone.now().isoformat() } except Exception as e: self.logger.error(f"Redis health check failed: {e}") return { 'status': 'unhealthy', 'error': str(e), 'timestamp': timezone.now().isoformat() } def get_overall_health(self) -> Dict[str, Any]: """ Get overall system health status. """ db_health = self.check_database_health() redis_health = self.check_redis_health() overall_status = 'healthy' if db_health['status'] != 'healthy' or redis_health['status'] != 'healthy': overall_status = 'unhealthy' return { 'status': overall_status, 'timestamp': timezone.now().isoformat(), 'components': { 'database': db_health, 'redis': redis_health } } # Global instances performance_monitor = PerformanceMonitor() security_monitor = SecurityMonitor() stream_monitor = StreamMonitor() health_checker = HealthChecker()