""" Simplified Test Server for Monitoring Load Testing Includes only monitoring infrastructure without heavy dependencies """ from fastapi import FastAPI, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from typing import Dict, Any from datetime import datetime import uuid import logging # Import monitoring modules from monitoring_service import get_monitoring_service from model_versioning import get_versioning_system from production_logging import get_medical_logger from compliance_reporting import get_compliance_system from admin_endpoints import admin_router # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Initialize FastAPI app app = FastAPI( title="Medical AI Platform - Monitoring Test Server", description="Simplified server for monitoring infrastructure load testing", version="2.0.0" ) # CORS configuration app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Initialize monitoring and infrastructure services monitoring_service = get_monitoring_service() versioning_system = get_versioning_system() medical_logger = get_medical_logger("medical_ai_test") compliance_system = get_compliance_system() logger.info("Monitoring test server initialized") # In-memory job tracking for testing job_tracker: Dict[str, Dict[str, Any]] = {} # Add monitoring middleware @app.middleware("http") async def monitoring_middleware(request: Request, call_next): """Monitoring middleware for request tracking""" start_time = datetime.utcnow() request_id = str(uuid.uuid4()) medical_logger.info("Request received", { "request_id": request_id, "method": request.method, "path": request.url.path, "client": request.client.host if request.client else "unknown" }) try: response = await call_next(request) end_time = datetime.utcnow() latency_ms = (end_time - start_time).total_seconds() * 1000 monitoring_service.track_request( endpoint=request.url.path, latency_ms=latency_ms, status_code=response.status_code ) medical_logger.info("Request completed", { "request_id": request_id, "method": request.method, "path": request.url.path, "status_code": response.status_code, "latency_ms": round(latency_ms, 2) }) return response except Exception as e: end_time = datetime.utcnow() latency_ms = (end_time - start_time).total_seconds() * 1000 monitoring_service.track_error( endpoint=request.url.path, error_type=type(e).__name__, error_message=str(e) ) medical_logger.error("Request failed", { "request_id": request_id, "method": request.method, "path": request.url.path, "error": str(e), "error_type": type(e).__name__, "latency_ms": round(latency_ms, 2) }) raise # Startup event handler @app.on_event("startup") async def startup_event(): """Initialize all services on startup""" medical_logger.info("Starting monitoring test server initialization", { "version": "2.0.0", "timestamp": datetime.utcnow().isoformat() }) # Initialize monitoring service monitoring_service.start_monitoring() medical_logger.info("Monitoring service initialized", { "cache_enabled": True, "alert_threshold": 0.05 }) # Register test model versions model_versions = [ {"model_id": "bio_clinical_bert", "version": "1.0.0", "source": "HuggingFace"}, {"model_id": "biogpt", "version": "1.0.0", "source": "HuggingFace"}, {"model_id": "pubmed_bert", "version": "1.0.0", "source": "HuggingFace"}, {"model_id": "hubert_ecg", "version": "1.0.0", "source": "HuggingFace"}, {"model_id": "monai_unetr", "version": "1.0.0", "source": "HuggingFace"}, {"model_id": "medgemma_2b", "version": "1.0.0", "source": "HuggingFace"} ] for model_config in model_versions: versioning_system.register_model_version( model_id=model_config["model_id"], version=model_config["version"], metadata={"source": model_config["source"]} ) medical_logger.info("Model versioning initialized", { "total_models": len(model_versions) }) # Test health check try: health_status = monitoring_service.get_system_health() medical_logger.info("Health check successful", { "status": health_status["status"], "components_ready": True }) except Exception as e: medical_logger.error("Health check failed during startup", { "error": str(e) }) medical_logger.info("Monitoring test server startup complete", { "status": "ready", "timestamp": datetime.utcnow().isoformat() }) # Include admin router app.include_router(admin_router) @app.get("/health") async def health_check(): """Basic health check endpoint""" system_health = monitoring_service.get_system_health() return { "status": system_health["status"], "components": { "monitoring": "active", "versioning": "active", "logging": "active", "compliance": "active" }, "monitoring": { "uptime_seconds": system_health["uptime_seconds"], "error_rate": system_health["error_rate"], "active_alerts": system_health["active_alerts"], "critical_alerts": system_health["critical_alerts"] }, "timestamp": datetime.utcnow().isoformat() } @app.get("/health/dashboard") async def get_health_dashboard(): """Comprehensive health dashboard endpoint""" try: system_health = monitoring_service.get_system_health() cache_stats = monitoring_service.get_cache_statistics() recent_alerts = monitoring_service.get_recent_alerts(limit=10) # Get model performance metrics model_metrics = {} try: active_models = versioning_system.list_model_versions() for model_info in active_models[:10]: model_id = model_info.get("model_id") if model_id: perf = versioning_system.get_model_performance(model_id) if perf: model_metrics[model_id] = { "version": model_info.get("version", "unknown"), "total_inferences": perf.get("total_inferences", 0), "avg_latency_ms": perf.get("avg_latency_ms", 0), "error_rate": perf.get("error_rate", 0.0), "last_used": perf.get("last_used", "never") } except Exception as e: medical_logger.warning("Failed to get model metrics", {"error": str(e)}) # Pipeline statistics pipeline_stats = { "total_jobs_processed": len(job_tracker), "completed_jobs": sum(1 for job in job_tracker.values() if job.get("status") == "completed"), "failed_jobs": sum(1 for job in job_tracker.values() if job.get("status") == "failed"), "processing_jobs": sum(1 for job in job_tracker.values() if job.get("status") == "processing"), "success_rate": 0.0 } if pipeline_stats["total_jobs_processed"] > 0: pipeline_stats["success_rate"] = ( pipeline_stats["completed_jobs"] / pipeline_stats["total_jobs_processed"] ) # Synthesis statistics (mock for testing) synthesis_stats = { "total_syntheses": 0, "avg_confidence": 0.0, "requiring_review": 0, "avg_processing_time_ms": 0 } # Compliance overview compliance_overview = { "hipaa_compliant": True, "gdpr_compliant": True, "audit_logging_active": True, "phi_removal_active": True, "encryption_enabled": True } dashboard = { "status": "operational" if system_health["status"] == "operational" else "degraded", "timestamp": datetime.utcnow().isoformat(), "system": { "uptime_seconds": system_health["uptime_seconds"], "uptime_human": f"{system_health['uptime_seconds'] // 3600}h {(system_health['uptime_seconds'] % 3600) // 60}m", "error_rate": system_health["error_rate"], "total_requests": system_health["total_requests"], "error_threshold": 0.05, "status": system_health["status"] }, "pipeline": pipeline_stats, "models": { "total_registered": len(model_metrics), "performance": model_metrics }, "synthesis": synthesis_stats, "cache": cache_stats, "alerts": { "active_count": system_health["active_alerts"], "critical_count": system_health["critical_alerts"], "recent": recent_alerts }, "compliance": compliance_overview, "components": { "monitoring_system": "operational", "versioning_system": "operational", "logging_system": "operational", "compliance_reporting": "operational", "cache_service": "operational" } } return dashboard except Exception as e: medical_logger.error("Dashboard generation failed", { "error": str(e), "timestamp": datetime.utcnow().isoformat() }) return { "status": "error", "timestamp": datetime.utcnow().isoformat(), "error": "Failed to generate complete dashboard", "message": str(e) } @app.get("/") async def root(): """Root endpoint""" return { "message": "Medical AI Platform - Monitoring Test Server", "version": "2.0.0", "endpoints": { "health": "/health", "dashboard": "/health/dashboard", "admin": "/admin/*" } } if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=7860)