clevermicro/amq-adapter-python
14
0
Fork 0
Code Issues 16 Pull Requests 4 Actions Packages Projects Releases Wiki Activity
Files
00018c79ac6ef2889966a2040ec3ce3c648ef0a1
amq-adapter-python/demo.py
T
Stanislav Hejny 00018c79ac feat: add CI/CD, Docker Compose, and update demo for Docker environment 
Co-authored-by: aider (openrouter/openai/o3-mini-high) <aider@aider.chat>
2025-07-01 20:29:50 +01:00

337 lines
15 KiB
Python
Raw Blame History

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.9
# 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.
Reference in New Issue View Git Blame Copy Permalink