""" Admin Monitoring Module - System Health, API Monitor, Debug Console Provides read-only monitoring and debugging tools for Django Admin """ from django.shortcuts import render from django.contrib.admin.views.decorators import staff_member_required from django.utils import timezone from django.db import connection from django.conf import settings import time import os @staff_member_required def system_health_dashboard(request): """ System infrastructure health monitoring Checks: Database, Redis, Celery, File System """ context = { 'page_title': 'System Health Monitor', 'checked_at': timezone.now(), 'checks': [] } # Database Check db_check = { 'name': 'PostgreSQL Database', 'status': 'unknown', 'message': '', 'details': {} } try: start = time.time() with connection.cursor() as cursor: cursor.execute("SELECT version()") version = cursor.fetchone()[0] cursor.execute("SELECT COUNT(*) FROM django_session") session_count = cursor.fetchone()[0] elapsed = (time.time() - start) * 1000 db_check.update({ 'status': 'healthy', 'message': f'Connected ({elapsed:.2f}ms)', 'details': { 'version': version.split('\n')[0], 'response_time': f'{elapsed:.2f}ms', 'active_sessions': session_count } }) except Exception as e: db_check.update({ 'status': 'error', 'message': f'Connection failed: {str(e)}' }) context['checks'].append(db_check) # Redis Check redis_check = { 'name': 'Redis Cache', 'status': 'unknown', 'message': '', 'details': {} } try: import redis r = redis.Redis( host=settings.CACHES['default']['LOCATION'].split(':')[0] if ':' in settings.CACHES['default'].get('LOCATION', '') else 'redis', port=6379, db=0, socket_connect_timeout=2 ) start = time.time() r.ping() elapsed = (time.time() - start) * 1000 info = r.info() redis_check.update({ 'status': 'healthy', 'message': f'Connected ({elapsed:.2f}ms)', 'details': { 'version': info.get('redis_version', 'unknown'), 'uptime': f"{info.get('uptime_in_seconds', 0) // 3600}h", 'connected_clients': info.get('connected_clients', 0), 'used_memory': f"{info.get('used_memory_human', 'unknown')}", 'response_time': f'{elapsed:.2f}ms' } }) except Exception as e: redis_check.update({ 'status': 'error', 'message': f'Connection failed: {str(e)}' }) context['checks'].append(redis_check) # Celery Workers Check celery_check = { 'name': 'Celery Workers', 'status': 'unknown', 'message': '', 'details': {} } try: from igny8_core.celery import app inspect = app.control.inspect(timeout=2) stats = inspect.stats() active = inspect.active() if stats: worker_count = len(stats) total_tasks = sum(len(tasks) for tasks in active.values()) if active else 0 celery_check.update({ 'status': 'healthy', 'message': f'{worker_count} worker(s) active', 'details': { 'workers': worker_count, 'active_tasks': total_tasks, 'worker_names': list(stats.keys()) } }) else: celery_check.update({ 'status': 'warning', 'message': 'No workers responding' }) except Exception as e: celery_check.update({ 'status': 'error', 'message': f'Check failed: {str(e)}' }) context['checks'].append(celery_check) # File System Check fs_check = { 'name': 'File System', 'status': 'unknown', 'message': '', 'details': {} } try: import shutil media_root = settings.MEDIA_ROOT static_root = settings.STATIC_ROOT media_stat = shutil.disk_usage(media_root) if os.path.exists(media_root) else None if media_stat: free_gb = media_stat.free / (1024**3) total_gb = media_stat.total / (1024**3) used_percent = (media_stat.used / media_stat.total) * 100 fs_check.update({ 'status': 'healthy' if used_percent < 90 else 'warning', 'message': f'{free_gb:.1f}GB free of {total_gb:.1f}GB', 'details': { 'media_root': media_root, 'free_space': f'{free_gb:.1f}GB', 'total_space': f'{total_gb:.1f}GB', 'used_percent': f'{used_percent:.1f}%' } }) else: fs_check.update({ 'status': 'warning', 'message': 'Media directory not found' }) except Exception as e: fs_check.update({ 'status': 'error', 'message': f'Check failed: {str(e)}' }) context['checks'].append(fs_check) # Overall system status statuses = [check['status'] for check in context['checks']] if 'error' in statuses: context['overall_status'] = 'error' context['overall_message'] = 'System has errors' elif 'warning' in statuses: context['overall_status'] = 'warning' context['overall_message'] = 'System has warnings' else: context['overall_status'] = 'healthy' context['overall_message'] = 'All systems operational' return render(request, 'admin/monitoring/system_health.html', context) @staff_member_required def api_monitor_dashboard(request): """ API endpoint health monitoring Tests key endpoints and displays response times """ from django.test.client import Client context = { 'page_title': 'API Monitor', 'checked_at': timezone.now(), 'endpoint_groups': [] } # Define endpoint groups to check endpoint_configs = [ { 'name': 'Authentication', 'endpoints': [ {'path': '/api/v1/auth/check/', 'method': 'GET', 'auth_required': False}, ] }, { 'name': 'System Settings', 'endpoints': [ {'path': '/api/v1/system/health/', 'method': 'GET', 'auth_required': False}, ] }, { 'name': 'Planner Module', 'endpoints': [ {'path': '/api/v1/planner/keywords/', 'method': 'GET', 'auth_required': True}, ] }, { 'name': 'Writer Module', 'endpoints': [ {'path': '/api/v1/writer/tasks/', 'method': 'GET', 'auth_required': True}, ] }, { 'name': 'Billing', 'endpoints': [ {'path': '/api/v1/billing/credits/balance/', 'method': 'GET', 'auth_required': True}, ] }, ] client = Client() for group_config in endpoint_configs: group_results = { 'name': group_config['name'], 'endpoints': [] } for endpoint in group_config['endpoints']: result = { 'path': endpoint['path'], 'method': endpoint['method'], 'status': 'unknown', 'status_code': None, 'response_time': None, 'message': '' } try: start = time.time() if endpoint['method'] == 'GET': response = client.get(endpoint['path']) else: response = client.post(endpoint['path']) elapsed = (time.time() - start) * 1000 result.update({ 'status_code': response.status_code, 'response_time': f'{elapsed:.2f}ms', }) # Determine status if response.status_code < 300: result['status'] = 'healthy' result['message'] = 'OK' elif response.status_code == 401 and endpoint.get('auth_required'): result['status'] = 'healthy' result['message'] = 'Auth required (expected)' elif response.status_code < 500: result['status'] = 'warning' result['message'] = 'Client error' else: result['status'] = 'error' result['message'] = 'Server error' except Exception as e: result.update({ 'status': 'error', 'message': str(e)[:100] }) group_results['endpoints'].append(result) context['endpoint_groups'].append(group_results) # Calculate overall stats all_endpoints = [ep for group in context['endpoint_groups'] for ep in group['endpoints']] total = len(all_endpoints) healthy = len([ep for ep in all_endpoints if ep['status'] == 'healthy']) warnings = len([ep for ep in all_endpoints if ep['status'] == 'warning']) errors = len([ep for ep in all_endpoints if ep['status'] == 'error']) context['stats'] = { 'total': total, 'healthy': healthy, 'warnings': warnings, 'errors': errors, 'health_percentage': (healthy / total * 100) if total > 0 else 0 } return render(request, 'admin/monitoring/api_monitor.html', context) @staff_member_required def debug_console(request): """ System debug information (read-only) Shows environment, database config, cache config, etc. """ context = { 'page_title': 'Debug Console', 'checked_at': timezone.now(), 'sections': [] } # Environment Variables Section env_section = { 'title': 'Environment', 'items': { 'DEBUG': settings.DEBUG, 'ENVIRONMENT': os.getenv('ENVIRONMENT', 'not set'), 'DJANGO_SETTINGS_MODULE': os.getenv('DJANGO_SETTINGS_MODULE', 'not set'), 'ALLOWED_HOSTS': settings.ALLOWED_HOSTS, 'TIME_ZONE': settings.TIME_ZONE, 'USE_TZ': settings.USE_TZ, } } context['sections'].append(env_section) # Database Configuration db_config = settings.DATABASES.get('default', {}) db_section = { 'title': 'Database Configuration', 'items': { 'ENGINE': db_config.get('ENGINE', 'not set'), 'NAME': db_config.get('NAME', 'not set'), 'HOST': db_config.get('HOST', 'not set'), 'PORT': db_config.get('PORT', 'not set'), 'CONN_MAX_AGE': db_config.get('CONN_MAX_AGE', 'not set'), } } context['sections'].append(db_section) # Cache Configuration cache_config = settings.CACHES.get('default', {}) cache_section = { 'title': 'Cache Configuration', 'items': { 'BACKEND': cache_config.get('BACKEND', 'not set'), 'LOCATION': cache_config.get('LOCATION', 'not set'), 'KEY_PREFIX': cache_config.get('KEY_PREFIX', 'not set'), } } context['sections'].append(cache_section) # Celery Configuration celery_section = { 'title': 'Celery Configuration', 'items': { 'BROKER_URL': getattr(settings, 'CELERY_BROKER_URL', 'not set'), 'RESULT_BACKEND': getattr(settings, 'CELERY_RESULT_BACKEND', 'not set'), 'TASK_ALWAYS_EAGER': getattr(settings, 'CELERY_TASK_ALWAYS_EAGER', False), } } context['sections'].append(celery_section) # Media & Static Files files_section = { 'title': 'Media & Static Files', 'items': { 'MEDIA_ROOT': settings.MEDIA_ROOT, 'MEDIA_URL': settings.MEDIA_URL, 'STATIC_ROOT': settings.STATIC_ROOT, 'STATIC_URL': settings.STATIC_URL, } } context['sections'].append(files_section) # Installed Apps (count) apps_section = { 'title': 'Installed Applications', 'items': { 'Total Apps': len(settings.INSTALLED_APPS), 'Custom Apps': len([app for app in settings.INSTALLED_APPS if app.startswith('igny8_')]), } } context['sections'].append(apps_section) # Middleware (count) middleware_section = { 'title': 'Middleware', 'items': { 'Total Middleware': len(settings.MIDDLEWARE), } } context['sections'].append(middleware_section) return render(request, 'admin/monitoring/debug_console.html', context)