Files
DevServer/app/main.py
Hardik Sankhla 2304c3ed12 Initial commit: Complete DevServer monitoring framework
- FastAPI application with OpenTelemetry instrumentation
- Redis caching and Celery task processing
- Docker Compose setup with monitoring stack
- Grafana dashboards and Prometheus metrics
- Jaeger distributed tracing
- Comprehensive documentation in docs/ folder
- Monitoring configuration and troubleshooting guides
2025-10-13 09:55:17 +05:30

723 lines
28 KiB
Python

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)}")