import json import os import random import shutil import time import httpx # For emulating remote service call # import uvicorn from fastapi import FastAPI, File, Form, UploadFile # OpenTelemetry imports from opentelemetry import metrics, trace from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter from opentelemetry.exporter.prometheus import PrometheusMetricReader from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor from opentelemetry.sdk.metrics import MeterProvider # from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter from amqp.config.amq_configuration import AMQConfiguration from otdemo.otdemo_adapter import OTDemoAmqpAdapter # --- OpenTelemetry Configuration --- # Configure Resource: Defines common attributes for traces and metrics resource = Resource.create( attributes={ "service.name": "fastapi-document-service", "application": "CleverMicroDemo", "environment": "development", } ) # Configure TracerProvider: Responsible for creating and managing Tracers trace_provider = TracerProvider(resource=resource) # OTLPSpanExporter: Exports spans to an OTLP collector (e.g., Jaeger) via gRPC # Endpoint can be overridden by OTEL_EXPORTER_OTLP_ENDPOINT environment variable otlp_exporter = OTLPSpanExporter( endpoint=os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317") ) # BatchSpanProcessor: Batches spans for efficient export trace_provider.add_span_processor(BatchSpanProcessor(otlp_exporter)) # Optional: ConsoleSpanExporter for local debugging to print traces to console if os.environ.get("OTEL_DEBUG_CONSOLE_TRACES", "false").lower() == "true": trace_provider.add_span_processor(BatchSpanProcessor(ConsoleSpanExporter())) # Set the configured TracerProvider globally trace.set_tracer_provider(trace_provider) # Configure MeterProvider: Responsible for creating and managing Meters for metrics # PrometheusMetricReader: Exposes metrics via an HTTP endpoint for Prometheus to scrape # Host and port can be overridden by OTEL_EXPORTER_PROMETHEUS_HOST and OTEL_EXPORTER_PROMETHEUS_PORT prometheus_reader = PrometheusMetricReader( # handler_address=os.environ.get("OTEL_EXPORTER_PROMETHEUS_HOST", "0.0.0.0"), # handler_port=int(os.environ.get("OTEL_EXPORTER_PROMETHEUS_PORT", "8001")) ) # MeterProvider: Uses the PrometheusMetricReader to periodically export metrics metric_provider = MeterProvider(resource=resource, metric_readers=[prometheus_reader]) # Set the configured MeterProvider globally metrics.set_meter_provider(metric_provider) # Get tracer and meter instances from the global providers tracer = trace.get_tracer("document-service.tracer") meter = metrics.get_meter("document-service.meter") # --- Custom Metrics Definition --- # Counters: For cumulative sums (e.g., total requests) documents_uploaded_counter = meter.create_counter( name="documents_uploaded_total", description="Total number of documents uploaded", unit="1" ) metadata_processed_counter = meter.create_counter( name="documents_metadata_processed_total", description="Total number of document metadata processed", unit="1", ) documents_retrieved_counter = meter.create_counter( name="documents_retrieved_total", description="Total number of documents retrieved", unit="1" ) # Histograms: For distributions of values (e.g., request durations) internal_api_call_duration_histogram = meter.create_histogram( name="internal_api_call_duration_seconds", description="Duration of internal API calls to metadata endpoint", unit="s", ) # --- FastAPI Application Setup --- app = FastAPI( title="CleverMicro Document Service Demo", description="A FastAPI application demonstrating OpenTelemetry integration with Jaeger and Prometheus.", ) # Instrument FastAPI: Automatically creates spans for incoming requests FastAPIInstrumentor.instrument_app(app) # Instrument httpx: Automatically propagates trace context for outgoing HTTP calls HTTPXClientInstrumentor().instrument() # Directory for storing uploaded files (temporary for demo) UPLOAD_DIR = "/tmp/clevermicro_documents" os.makedirs(UPLOAD_DIR, exist_ok=True) otadapter = OTDemoAmqpAdapter(AMQConfiguration("otdemo/otdemo.properties"), app.routes) @app.post("/api/v3/documents") async def upload_document( file: UploadFile = File(...), # Multipart file upload name: str = Form(...), # Form field for document name description: str = Form(...), # Form field for document description ): # Create a custom span for this specific endpoint's logic with tracer.start_as_current_span("upload_document_endpoint_logic"): # Save the uploaded file to the temporary directory file_location = os.path.join(UPLOAD_DIR, file.filename) try: with open(file_location, "wb") as buffer: shutil.copyfileobj(file.file, buffer) print(f"File '{file.filename}' saved to {file_location}") # Increment the documents_uploaded_total metric documents_uploaded_counter.add(1, {"file.name": file.filename, "document.name": name}) # Emulate a REST call to a "remote" service (which is actually this same app) # This demonstrates trace context propagation across HTTP calls. metadata_payload = {"name": name, "description": description} print( f"Emulating remote call to /api/v3/documents/metadata with: {json.dumps(metadata_payload)}" ) start_time = time.time() async with httpx.AsyncClient() as client: # httpx instrumentation ensures the current trace context is propagated response = await client.post( # IMPORTANT: Use the actual host and port where your FastAPI app is running # If running directly on host, use http://localhost:8000 # If running inside Docker and calling itself, use http://host.docker.internal:8000 # "http://localhost:8000/api/v3/documents/metadata", "http://app:8000/api/v3/documents/metadata", json=metadata_payload, ) response.raise_for_status() # Raise an exception for bad status codes (4xx or 5xx) end_time = time.time() duration = end_time - start_time # Record the duration of the internal API call internal_api_call_duration_histogram.record( duration, {"endpoint": "/api/v3/documents/metadata"} ) print(f"Internal metadata service responded: {response.json()}") return { "message": "Document uploaded and metadata processing initiated", "filename": file.filename, "name": name, "description": description, } except httpx.HTTPStatusError as e: print(f"Error calling metadata service: {e.response.status_code} - {e.response.text}") # Propagate the error in the trace current_span = trace.get_current_span() current_span.set_attribute("error.type", "HTTPStatusError") current_span.record_exception(e) return { "message": "Document uploaded but metadata processing failed", "detail": str(e), }, 500 except Exception as e: print(f"An unexpected error occurred during document upload: {e}") # Propagate the error in the trace current_span = trace.get_current_span() current_span.set_attribute("error.type", "UnexpectedError") current_span.record_exception(e) return {"message": "Error processing document", "detail": str(e)}, 500 @app.post("/api/v3/documents/metadata") async def process_document_metadata(metadata: dict): # This endpoint is automatically instrumented by FastAPIInstrumentor # A new span will be created, and its parent will be the span from the calling /api/v3/documents # due to trace context propagation via httpx. print(f"Received metadata for processing: {json.dumps(metadata, indent=2)}") # Increment the metadata_processed_total metric metadata_processed_counter.add(1, {"document.name": metadata.get("name")}) # Simulate some processing time to make traces more interesting time.sleep(random.uniform(0.05, 0.2)) return {"status": "metadata processed", "received_data": metadata} @app.get("/api/v3/documents") async def get_documents(): # Create a custom span for the logic within this endpoint with tracer.start_as_current_span("get_documents_endpoint_logic"): documents = [] # Generate several random documents to emulate a list for i in range(random.randint(2, 5)): documents.append( { "id": f"doc-{random.randint(1000, 9999)}", "name": f"Document {random.randint(1, 100)}", "description": f"Description for document {random.randint(1, 1000)}", "uploaded_at": time.strftime( "%Y-%m-%dT%H:%M:%SZ", time.gmtime(time.time() - random.randint(0, 86400 * 30)), ), } ) # Increment the documents_retrieved_total metric documents_retrieved_counter.add(1) # Simulate some processing time time.sleep(random.uniform(0.01, 0.1)) return {"documents": documents} # --- How to Run the Demo --- # 1. Save the code: # Save the Python code above as `main.py` in a new directory. # 2. Create `requirements.txt`: # Create a file named `requirements.txt` in the same directory with the following content: # ``` # fastapi==0.111.0 # uvicorn[standard]==0.29.0 # python-multipart==0.0.20 # httpx==0.27.0 # opentelemetry-api==1.25.0 # opentelemetry-sdk==1.25.0 # opentelemetry-exporter-otlp-proto-grpc==1.25.0 # For Jaeger traces # opentelemetry-exporter-prometheus==1.25.0 # For Prometheus metrics # opentelemetry-instrumentation-fastapi==0.46b0 # IMPORTANT: Check compatibility with your FastAPI version! # opentelemetry-instrumentation-httpx==0.46b0 # IMPORTANT: Check compatibility with your httpx version! # protobuf==4.25.3 # Required by OTLP exporter, ensure version compatibility if issues arise # ``` # 3. Install dependencies: # Open your terminal in the directory where you saved `main.py` and `requirements.txt`, then run: # `pip install -r requirements.txt` # 4. Run Jaeger and Prometheus (using Docker Compose for simplicity): # Create a file named `docker-compose.yaml` in the same directory: # ```yaml # version: '3.8' # services: # jaeger: # image: jaegertracing/all-in-one:latest # ports: # - "6831:6831/udp" # UDP Thrift # - "14268:14268" # HTTP Thrift # - "14250:14250" # gRPC # - "4317:4317" # OTLP gRPC collector (for traces from our app) # - "4318:4318" # OTLP HTTP collector # - "16686:16686" # Jaeger UI # environment: # - COLLECTOR_OTLP_ENABLED=true # Enable OTLP reception # # prometheus: # image: prom/prometheus:latest # volumes: # - ./prometheus.yml:/etc/prometheus/prometheus.yml # Mount Prometheus config # ports: # - "9090:9090" # Prometheus UI # command: # - '--config.file=/etc/prometheus/prometheus.yml' # ``` # Create a file named `prometheus.yml` in the same directory: # ```yaml # global: # scrape_interval: 15s # How frequently to scrape targets # # scrape_configs: # - job_name: 'fastapi_app' # # The 'metrics' endpoint of our FastAPI app exposed by PrometheusMetricReader # # 'host.docker.internal' allows Docker containers to connect to the host machine's localhost # static_configs: # - targets: ['host.docker.internal:8001'] # ``` # Start Jaeger and Prometheus: # `docker-compose up -d` (the `-d` runs them in the background) # 5. Run the FastAPI application: # In your terminal (where `main.py` is located), run: # `uvicorn main:app --host 0.0.0.0 --port 8000` # (Optional: To see traces printed to console, set `export OTEL_DEBUG_CONSOLE_TRACES=true` before `uvicorn`) # --- Accessing the UIs --- # * **Jaeger UI:** Open your web browser and go to `http://localhost:16686` # * **Prometheus UI:** Open your web browser and go to `http://localhost:9090` # * **FastAPI App (Root):** `http://localhost:8000` # * **FastAPI Metrics Endpoint:** `http://localhost:8001/metrics` (Prometheus scrapes this, you can also view it directly) # --- How to Test and Observe --- # 1. Ensure all services are running (`docker-compose ps` should show `jaeger` and `prometheus` up, and `uvicorn` running in your terminal). # 2. **Trigger GET requests:** # Open your browser or use `curl` to hit: `http://localhost:8000/api/v3/documents` # Refresh a few times. # 3. **Trigger POST requests (for file upload and internal call):** # You'll need a tool like Postman, Insomnia, or `curl` (more complex for multipart) for this. # **Using Postman/Insomnia:** # * Method: `POST` # * URL: `http://localhost:8000/api/v3/documents` # * Body: Select `form-data` # * Add a key `file`, Type `File`, Value: Choose any small file from your computer (e.g., a `.txt` or `.png`). # * Add a key `name`, Type `Text`, Value: `My Demo Document` # * Add a key `description`, Type `Text`, Value: `A test document for OpenTelemetry.` # * Send the request multiple times. # 4. **Observe in Jaeger UI (`http://localhost:16686`):** # * Select "Service": `fastapi-document-service` # * Click "Find Traces". # * You should see traces for `/api/v3/documents` (POST) and `/api/v3/documents` (GET). # * Crucially, for POST requests, expand the trace: you will see the main `/api/v3/documents` span, and nested within it, a child span for the `POST /api/v3/documents/metadata` HTTP request. This demonstrates automatic context propagation. # 5. **Observe in Prometheus UI (`http://localhost:9090`):** # * Go to the "Graph" tab. # * In the expression bar, type and execute queries like: # * `documents_uploaded_total` # * `documents_metadata_processed_total` # * `documents_retrieved_total` # * `internal_api_call_duration_seconds_sum` # * `internal_api_call_duration_seconds_count` # * You will see the values of these metrics increasing with your requests. # This setup provides a clear, runnable demo for OpenTelemetry traces and metrics in a FastAPI application, including automatic context propagation for internal HTTP calls.