from fastapi import FastAPI, HTTPException, Request, Path from random import randint import copy import json import redis from celery.result import AsyncResult from opentelemetry import trace from app.logging import setup_logging, get_logger, log_with_trace from prometheus_fastapi_instrumentator import Instrumentator import time # Setup logging logger = setup_logging(service_name="devserver.api.fastapi", log_level="DEBUG") tracer = trace.get_tracer("devserver.api.tracer") # Initialize Redis client logger.info("Initializing Redis connection...") try: redis_client = redis.Redis(host='redis', port=6379, db=0, decode_responses=True) redis_client.ping() logger.info("Redis connection established successfully") except Exception as e: logger.error(f"Failed to connect to Redis: {e}", exc_info=True) raise app = FastAPI( title="DevServer API", version="1.0.0", description="FastAPI with OpenTelemetry, Redis, and Celery" ) # ============================================================================ # INITIALIZE OPENTELEMETRY - Explicit Instrumentation # ============================================================================ # Import and apply OpenTelemetry instrumentation BEFORE any requests from app.opentelemetry_config import instrument_fastapi_app logger.info("=" * 80) logger.info("🔧 Applying OpenTelemetry Instrumentation...") logger.info("=" * 80) try: instrument_fastapi_app(app) logger.info("=" * 80) logger.info("✅ OpenTelemetry instrumentation applied successfully!") logger.info("=" * 80) except Exception as e: logger.error("=" * 80) logger.error(f"❌ Failed to apply OpenTelemetry instrumentation: {e}") logger.error("⚠️ Traces will NOT be generated!") logger.error("=" * 80) # ============================================================================ # Initialize Prometheus metrics instrumentator = Instrumentator( should_group_status_codes=False, should_ignore_untemplated=True, should_respect_env_var=False, should_instrument_requests_inprogress=True, excluded_handlers=[".*admin.*", "/metrics"], env_var_name="ENABLE_METRICS", inprogress_name="http_requests_inprogress", inprogress_labels=True, ) instrumentator.instrument(app).expose(app, endpoint="/metrics") dice_rolls = [[], [], []] # Middleware to log all requests @app.middleware("http") async def log_requests(request: Request, call_next): """Log all incoming requests with trace context""" start_time = time.time() # Get trace context span = trace.get_current_span() span_context = span.get_span_context() trace_id = format(span_context.trace_id, '032x') if span_context.trace_id else 'N/A' logger.info(f"→ Incoming request: {request.method} {request.url.path}") logger.debug(f" TraceID: {trace_id}") logger.debug(f" Client: {request.client.host if request.client else 'unknown'}") response = await call_next(request) duration = time.time() - start_time logger.info(f"← Response: {response.status_code} | Duration: {duration:.3f}s | {request.method} {request.url.path}") return response @app.on_event("startup") async def startup_event(): """Log application startup""" logger.info("=" * 60) logger.info("DevServer FastAPI Application Starting...") logger.info("=" * 60) logger.info("Service: devserver.api.fastapi") logger.info("OpenTelemetry tracing: ENABLED") logger.info("Redis caching: ENABLED") logger.info("Celery tasks: ENABLED") logger.info("API Base URL: http://localhost:8000") logger.info("=" * 60) @app.on_event("shutdown") async def shutdown_event(): """Log application shutdown""" logger.info("=" * 60) logger.info("DevServer FastAPI Application Shutting Down...") logger.info("=" * 60) @app.get("/", tags=["Info"]) async def root(): """Root endpoint with basic info.""" logger.info("API information requested") return { "service": "devserver.api.fastapi", "message": "DevServer API with OpenTelemetry, Redis, and Celery", "version": "1.0.0", "endpoints": { "info": { "GET /": "API information", "GET /health": "Health check", "GET /docs": "Interactive API documentation (Swagger UI)" }, "dice_game": { "GET /api/dice/roll?session_id=xxx&use_cache=true": "Roll dice with Redis caching and session tracking", "Description": "Returns 15 dice rolls with statistics. Use session_id for unique sessions, use_cache to enable/disable caching (5min TTL)" }, "celery_tasks": { "POST /api/tasks/fibonacci?n={number}&use_cache=true": "Calculate Fibonacci with Redis caching", "GET /api/tasks/{task_id}": "Get task result by ID", "Description": "Fibonacci results are cached for 1 hour. Set use_cache=false to force recalculation" }, "redis_cache": { "POST /api/cache/set": "Set cache value", "GET /api/cache/{key}": "Get cache value", "DELETE /api/cache/{key}": "Delete cache entry", "GET /api/cache/keys?pattern=*": "List all cache keys", "GET /api/cache/stats": "Get Redis statistics", "POST /api/cache/flush": "Clear all cache (danger)", "POST /api/cache/batch-set": "Set multiple cache values", "GET /api/cache/batch-get?keys=k1,k2": "Get multiple cache values" }, "data_processing": { "POST /api/process": "Process data with cache + Celery workflow" } }, "monitoring": { "jaeger_ui": "http://localhost:16686", "rabbitmq_ui": "http://localhost:15672" } } @app.get("/health", tags=["Info"]) async def health_check(): """Health check endpoint""" with tracer.start_as_current_span("health_check") as span: span.set_attribute("service.name", "devserver.api.fastapi") # Check Redis try: redis_client.ping() redis_status = "healthy" except Exception as e: redis_status = f"unhealthy: {e}" logger.error(f"Redis health check failed: {e}") logger.debug("Health check performed") return { "status": "healthy", "service": "devserver.api.fastapi", "redis": redis_status } @app.get("/api/dice/roll", tags=["Dice Game"]) async def roll_dice(session_id: str = "default", use_cache: bool = True): """Roll dice 15 times with distributed tracing and Redis caching. Args: session_id: Unique session identifier for caching results use_cache: If True, check cache first and store results """ with tracer.start_as_current_span("api.dice.roll") as span: span.set_attribute("service.name", "devserver.api.fastapi") span.set_attribute("operation", "dice_roll") span.set_attribute("session_id", session_id) span.set_attribute("use_cache", use_cache) cache_key = f"dice:session:{session_id}" logger.info(f"Received dice roll request: session_id={session_id}, use_cache={use_cache}") # Check cache first if enabled if use_cache: try: cached_result = redis_client.get(cache_key) if cached_result: span.set_attribute("cache.hit", True) logger.info(f"Cache hit! Returning cached dice rolls for session: {session_id}") result = json.loads(str(cached_result)) result["source"] = "cache" result["cache_key"] = cache_key return result except Exception as e: logger.warning(f"Cache check failed: {e}") span.set_attribute("cache.hit", False) logger.info("Cache miss - rolling new dice") count = 0 # Roll a dice 15 times and store the results in three separate lists while count < 15: if count < 5: dice_rolls[0].append(roll()) # Append to the first list elif count < 10: dice_rolls[1].append(roll()) # Append to the second list else: dice_rolls[2].append(roll()) # Append to the third list count += 1 # Store the results to return results = copy.deepcopy(dice_rolls) total_rolls = sum(len(r) for r in results) span.set_attribute("dice.total_rolls", total_rolls) # Calculate statistics flat_results = [roll for sublist in results for roll in sublist] avg_roll = sum(flat_results) / len(flat_results) logger.info(f"Dice rolled successfully: {total_rolls} total rolls, avg={avg_roll:.2f}") logger.debug(f"Roll results: {results}") # Clear the original lists for the next request dice_rolls[0] = [] dice_rolls[1] = [] dice_rolls[2] = [] response = { "service": "devserver.api.fastapi", "operation": "dice_roll", "session_id": session_id, "dice_rolls": results, "total": total_rolls, "statistics": { "average": round(avg_roll, 2), "min": min(flat_results), "max": max(flat_results), "sum": sum(flat_results) }, "source": "fresh", "cache_key": cache_key } # Store in cache if enabled (TTL: 5 minutes) if use_cache: try: redis_client.setex(cache_key, 300, json.dumps(response)) span.set_attribute("cache.stored", True) logger.info(f"Stored dice rolls in cache: {cache_key} (TTL: 300s)") except Exception as e: logger.error(f"Failed to store in cache: {e}") return response def roll(): """Simulates rolling a six-sided dice with tracing.""" with tracer.start_as_current_span("dice.single_roll") as roll_span: result = randint(1, 6) roll_span.set_attribute("dice.value", result) roll_span.set_attribute("service.name", "devserver.api.fastapi") logger.debug(f"Single dice roll: {result}") return result @app.post("/api/tasks/fibonacci", tags=["Celery Tasks"]) async def create_fibonacci_task(n: int, use_cache: bool = True): """Create a Celery task to calculate Fibonacci number with Redis caching. Args: n: The Fibonacci number to calculate (0-40) use_cache: If True, check cache before creating task """ with tracer.start_as_current_span("api.tasks.fibonacci.create") as span: span.set_attribute("service.name", "devserver.api.fastapi") span.set_attribute("operation", "create_task") span.set_attribute("task.type", "fibonacci") span.set_attribute("fibonacci.n", n) span.set_attribute("use_cache", use_cache) logger.info(f"Creating Fibonacci task for n={n}, use_cache={use_cache}") if n < 0 or n > 40: logger.warning(f"Invalid Fibonacci number requested: n={n}") raise HTTPException(status_code=400, detail="n must be between 0 and 40") cache_key = f"fibonacci:result:{n}" # Check cache first if enabled if use_cache: try: cached_result = redis_client.get(cache_key) if cached_result: span.set_attribute("cache.hit", True) span.set_attribute("result.source", "cache") result_value = int(str(cached_result)) logger.info(f"Cache hit! Returning cached Fibonacci({n}) = {result_value}") return { "service": "devserver.api.fastapi", "operation": "fibonacci", "n": n, "result": result_value, "source": "cache", "cache_key": cache_key, "status": "completed" } except (ValueError, TypeError) as e: logger.warning(f"Cache value conversion failed: {e}") except Exception as e: logger.warning(f"Cache check failed: {e}") span.set_attribute("cache.hit", False) logger.debug(f"Cache miss - submitting Fibonacci calculation task to Celery: n={n}") from app.worker import calculate_fibonacci task = calculate_fibonacci.delay(n) # type: ignore span.set_attribute("task.id", task.id) span.set_attribute("task.status", "submitted") logger.info(f"Fibonacci task created successfully: task_id={task.id}, n={n}") return { "service": "devserver.api.fastapi", "operation": "create_fibonacci_task", "n": n, "task_id": task.id, "status": "submitted", "source": "celery_task", "message": f"Calculating fibonacci({n})", "check_result_at": f"/api/tasks/{task.id}", "cache_key": cache_key } @app.get("/api/tasks/{task_id}", tags=["Celery Tasks"]) async def get_task_result(task_id: str): """Get the result of a Celery task.""" with tracer.start_as_current_span("api.tasks.get_result") as span: span.set_attribute("service.name", "devserver.api.fastapi") span.set_attribute("operation", "get_task_result") span.set_attribute("task.id", task_id) logger.info(f"Checking task status: task_id={task_id}") from app.worker import app as celery_app try: task_result = AsyncResult(task_id, app=celery_app) if task_result.ready(): result = task_result.result span.set_attribute("task.status", "completed") logger.info(f"Task completed: task_id={task_id}") logger.debug(f"Task result: {result}") return { "service": "devserver.api.fastapi", "operation": "get_task_result", "task_id": task_id, "status": "completed", "result": result } else: span.set_attribute("task.status", "pending") logger.debug(f"Task still pending: task_id={task_id}") return { "service": "devserver.api.fastapi", "operation": "get_task_result", "task_id": task_id, "status": "pending", "message": "Task is still processing" } except Exception as e: logger.error(f"Error retrieving task result: task_id={task_id}, error={e}", exc_info=True) raise HTTPException(status_code=500, detail=f"Error retrieving task: {str(e)}") @app.post("/api/cache/set", tags=["Redis Cache"]) async def set_cache(key: str, value: str, ttl: int = 300): """Set a value in Redis cache with optional TTL.""" with tracer.start_as_current_span("api.cache.set") as span: span.set_attribute("service.name", "devserver.api.fastapi") span.set_attribute("operation", "cache_set") span.set_attribute("cache.key", key) span.set_attribute("cache.ttl", ttl) logger.info(f"Setting cache: key={key}, ttl={ttl}s") try: redis_client.setex(key, ttl, value) span.set_attribute("cache.operation", "success") logger.info(f"Cache set successfully: key={key}") logger.debug(f"Cache value: {value[:50]}..." if len(value) > 50 else f"Cache value: {value}") return { "service": "devserver.api.fastapi", "operation": "cache_set", "message": "Cache set successfully", "key": key, "ttl": ttl } except Exception as e: span.set_attribute("cache.operation", "failed") logger.error(f"Failed to set cache: key={key}, error={e}", exc_info=True) raise HTTPException(status_code=500, detail=f"Cache error: {str(e)}") @app.get("/api/cache/batch-get", tags=["Redis Cache"]) async def batch_get_cache(keys: str): """Get multiple cache values at once (comma-separated keys)""" with tracer.start_as_current_span("api.cache.batch_get") as span: span.set_attribute("service.name", "devserver.api.fastapi") span.set_attribute("operation", "cache_batch_get") key_list = [k.strip() for k in keys.split(',')] span.set_attribute("cache.batch_size", len(key_list)) logger.info(f"Batch getting {len(key_list)} cache entries") try: pipeline = redis_client.pipeline() for key in key_list: pipeline.get(key) values = pipeline.execute() result = {} found_count = 0 for key, value in zip(key_list, values): if value is not None: result[key] = value found_count += 1 span.set_attribute("cache.found_count", found_count) span.set_attribute("cache.hit_ratio", (found_count / len(key_list) * 100) if key_list else 0) logger.info(f"Batch get complete: {found_count}/{len(key_list)} found") return { "service": "devserver.api.fastapi", "operation": "cache_batch_get", "requested": len(key_list), "found": found_count, "data": result } except Exception as e: logger.error(f"Failed to batch get: error={e}", exc_info=True) raise HTTPException(status_code=500, detail=f"Cache error: {str(e)}") @app.post("/api/process", tags=["Data Processing"]) async def process_data(data: dict): """Process data using both Redis cache and Celery task.""" with tracer.start_as_current_span("api.process.data") as span: span.set_attribute("service.name", "devserver.api.fastapi") span.set_attribute("operation", "process_data") span.set_attribute("data.keys_count", len(data)) logger.info(f"Processing data request received") logger.debug(f"Data keys: {list(data.keys())}") from app.worker import process_data_with_cache # Generate a cache key based on data cache_key = f"processed:{hash(json.dumps(data, sort_keys=True))}" span.set_attribute("cache.key", cache_key) logger.debug(f"Generated cache key: {cache_key}") # Check cache first try: cached_result = redis_client.get(cache_key) if cached_result: span.set_attribute("cache.hit", True) span.set_attribute("result.source", "cache") logger.info(f"Cache hit! Returning cached result: cache_key={cache_key}") result = json.loads(str(cached_result)) logger.debug(f"Cached result: {result}") return { "service": "devserver.api.fastapi", "operation": "process_data", "source": "cache", "result": result } except Exception as e: logger.warning(f"Cache check failed: {e}") span.set_attribute("cache.hit", False) span.set_attribute("result.source", "celery_task") logger.info(f"Cache miss - submitting to Celery: cache_key={cache_key}") # If not in cache, process with Celery try: task = process_data_with_cache.delay(data, cache_key) # type: ignore span.set_attribute("task.id", task.id) logger.info(f"Task submitted to Celery: task_id={task.id}") return { "service": "devserver.api.fastapi", "operation": "process_data", "source": "celery_task", "task_id": task.id, "message": "Processing data, check /api/tasks/{task_id} for result" } except Exception as e: logger.error(f"Failed to submit task: error={e}", exc_info=True) raise HTTPException(status_code=500, detail=f"Task submission failed: {str(e)}") @app.get("/api/cache/keys", tags=["Redis Cache"]) async def list_cache_keys(pattern: str = "*"): """List all cache keys matching a pattern""" with tracer.start_as_current_span("api.cache.list_keys") as span: span.set_attribute("service.name", "devserver.api.fastapi") span.set_attribute("operation", "list_cache_keys") span.set_attribute("cache.pattern", pattern) logger.info(f"Listing cache keys with pattern: {pattern}") try: keys = redis_client.keys(pattern) key_count = len(keys) if keys else 0 # type: ignore span.set_attribute("cache.keys_found", key_count) logger.info(f"Found {key_count} cache keys") return { "service": "devserver.api.fastapi", "operation": "list_cache_keys", "pattern": pattern, "count": key_count, "keys": keys } except Exception as e: logger.error(f"Failed to list keys: error={e}", exc_info=True) raise HTTPException(status_code=500, detail=f"Cache error: {str(e)}") @app.post("/api/cache/flush", tags=["Redis Cache"]) async def flush_cache(): """Flush all cache entries (use with caution!)""" with tracer.start_as_current_span("api.cache.flush") as span: span.set_attribute("service.name", "devserver.api.fastapi") span.set_attribute("operation", "cache_flush") logger.warning("Flushing entire cache!") try: redis_client.flushdb() span.set_attribute("cache.flushed", True) logger.info("Cache flushed successfully") return { "service": "devserver.api.fastapi", "operation": "cache_flush", "message": "All cache entries deleted", "status": "success" } except Exception as e: logger.error(f"Failed to flush cache: error={e}", exc_info=True) raise HTTPException(status_code=500, detail=f"Cache error: {str(e)}") @app.get("/api/cache/stats", tags=["Redis Cache"]) async def cache_stats(): """Get Redis cache statistics""" with tracer.start_as_current_span("api.cache.stats") as span: span.set_attribute("service.name", "devserver.api.fastapi") span.set_attribute("operation", "cache_stats") logger.info("Fetching cache statistics") try: info = redis_client.info() # type: ignore stats = { "used_memory": info.get('used_memory_human', 'N/A'), # type: ignore "connected_clients": info.get('connected_clients', 0), # type: ignore "total_keys": redis_client.dbsize(), "uptime_seconds": info.get('uptime_in_seconds', 0), # type: ignore "hits": info.get('keyspace_hits', 0), # type: ignore "misses": info.get('keyspace_misses', 0) # type: ignore } # Calculate hit ratio total_ops = stats['hits'] + stats['misses'] hit_ratio = (stats['hits'] / total_ops * 100) if total_ops > 0 else 0 stats['hit_ratio_percent'] = round(hit_ratio, 2) span.set_attribute("cache.total_keys", stats['total_keys']) span.set_attribute("cache.hit_ratio", hit_ratio) logger.info(f"Cache stats: {stats['total_keys']} keys, {hit_ratio:.2f}% hit ratio") return { "service": "devserver.api.fastapi", "operation": "cache_stats", "stats": stats } except Exception as e: logger.error(f"Failed to get cache stats: error={e}", exc_info=True) raise HTTPException(status_code=500, detail=f"Cache error: {str(e)}") @app.post("/api/cache/batch-set", tags=["Redis Cache"]) async def batch_set_cache(items: dict): """Set multiple cache values at once""" with tracer.start_as_current_span("api.cache.batch_set") as span: span.set_attribute("service.name", "devserver.api.fastapi") span.set_attribute("operation", "cache_batch_set") span.set_attribute("cache.batch_size", len(items)) logger.info(f"Batch setting {len(items)} cache entries") try: pipeline = redis_client.pipeline() for key, value in items.items(): # Convert value to JSON string if it's a dict, list, or other complex type if isinstance(value, (dict, list)): value_str = json.dumps(value) else: value_str = str(value) pipeline.set(key, value_str) results = pipeline.execute() success_count = sum(1 for r in results if r) span.set_attribute("cache.success_count", success_count) logger.info(f"Batch set complete: {success_count}/{len(items)} successful") return { "service": "devserver.api.fastapi", "operation": "cache_batch_set", "total": len(items), "successful": success_count, "keys": list(items.keys()) } except Exception as e: logger.error(f"Failed to batch set: error={e}", exc_info=True) raise HTTPException(status_code=500, detail=f"Cache error: {str(e)}") @app.get("/api/cache/item/{key:path}", tags=["Redis Cache"]) async def get_cache(key: str): """Get a value from Redis cache.""" with tracer.start_as_current_span("api.cache.get") as span: span.set_attribute("service.name", "devserver.api.fastapi") span.set_attribute("operation", "cache_get") span.set_attribute("cache.key", key) logger.info(f"Getting cache: key={key}") try: value = redis_client.get(key) if value is None: span.set_attribute("cache.hit", False) logger.warning(f"Cache miss: key={key}") raise HTTPException(status_code=404, detail="Key not found in cache") span.set_attribute("cache.hit", True) logger.info(f"Cache hit: key={key}") value_str = str(value) logger.debug(f"Cache value: {value_str[:50]}..." if len(value_str) > 50 else f"Cache value: {value_str}") return { "service": "devserver.api.fastapi", "operation": "cache_get", "key": key, "value": value } except HTTPException: raise except Exception as e: logger.error(f"Failed to get cache: key={key}, error={e}", exc_info=True) raise HTTPException(status_code=500, detail=f"Cache error: {str(e)}") @app.delete("/api/cache/item/{key:path}", tags=["Redis Cache"]) async def delete_cache(key: str): """Delete a cache entry""" with tracer.start_as_current_span("api.cache.delete") as span: span.set_attribute("service.name", "devserver.api.fastapi") span.set_attribute("operation", "cache_delete") span.set_attribute("cache.key", key) logger.info(f"Deleting cache key: {key}") try: result = redis_client.delete(key) deleted = result > 0 # type: ignore span.set_attribute("cache.deleted", deleted) if deleted: logger.info(f"Cache key deleted: {key}") else: logger.warning(f"Cache key not found: {key}") return { "service": "devserver.api.fastapi", "operation": "cache_delete", "key": key, "deleted": deleted } except Exception as e: logger.error(f"Failed to delete key: error={e}", exc_info=True) raise HTTPException(status_code=500, detail=f"Cache error: {str(e)}")