diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..4ad455f --- /dev/null +++ b/Dockerfile @@ -0,0 +1,43 @@ +# Use the official Python image as the base image +FROM python:3.11-slim + +# Set the working directory to /app +WORKDIR /app +ENV PYTHONPATH=/app + +# Install system dependencies +RUN apt-get update && apt-get install -y \ + gcc \ + && rm -rf /var/lib/apt/lists/* + +# Copy requirements and install Python dependencies +COPY pyproject.toml . +RUN pip install --no-cache-dir build +RUN pip install --no-cache-dir . + +# Install additional dependencies for demo.py +RUN pip install --no-cache-dir \ + 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 \ + opentelemetry-exporter-prometheus==1.25.0 \ + opentelemetry-instrumentation-fastapi==0.46b0 \ + opentelemetry-instrumentation-httpx==0.46b0 \ + protobuf==4.25.3 + +# Copy the application code +COPY ./amqp ./amqp +COPY ./otdemo ./otdemo +COPY ./demo.py . + +# Create directory for uploaded files +RUN mkdir -p /tmp/clevermicro_documents + +# Expose the port +EXPOSE 8000 8001 + +# Run the FastAPI application +CMD ["uvicorn", "demo:app", "--host", "0.0.0.0", "--port", "8000"] diff --git a/demo.py b/demo.py new file mode 100644 index 0000000..18b38eb --- /dev/null +++ b/demo.py @@ -0,0 +1,336 @@ +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://host.docker.internal: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. diff --git a/otdemo/otdemo.properties b/otdemo/otdemo.properties new file mode 100644 index 0000000..5fb0120 --- /dev/null +++ b/otdemo/otdemo.properties @@ -0,0 +1,41 @@ +# ==================================================================== +# CleverMicro AMQ Adapter settings +# ==================================================================== +[CleverMicro-AMQ] +# CleverMicro AMQ Adapter settings that identify this AMQ Adapter instance to other adapters + +# A name to identify this service for the AMQ Adapter +cm.amq-adapter.service-name=otdemo +# The CleverMicro AMQ Adapter instance ID, used to identify this instance in the CleverMicro AMQ Adapter cluster +cm.amq-adapter.generator-id=1 +# Route mapping. For route detail description see section 3.1 in Onboarding Guide for Python: +# https://docs.cleverthis.com/en/architecture/microservices/onboarding/python +# NOTE: this is a comma-separated list of route mappings (multiple, comma-separated routes are allowed here) +cm.amq-adapter.route-mapping=otdemo::otdemo:otdemo.localhost.#:5:0 +# Indicate if AMQ Adapter should respond with HTTP 401 Unauthorized if the request does not contain valid JWT token +cm.amq-adapter.require-authenticated-user=false + +# CleverMicro AMQ Adapter dispatcher, settings related to RabbitMQ connection and optional REST forwarding +cm.dispatch.use-dlq=true +cm.dispatch.amq-host=rabbitmq +cm.dispatch.amq-port=5672 +cm.dispatch.amq-port-tls=5671 +cm.dispatch.download-dir=/tmp/downloaded_files +# Not used for CleveSwarm nor CleverBRAG or in tight coupling mode. +# applicable only in loose coupling mode, specify the destination where to forward the REST request +#cm.dispatch.service-host=cleverthis-service-name.localhost +#cm.dispatch.service-port=8080 + +# Backpressure monitor settings +# Define maximum number of messages of being processed in parallel before triggering backpressure OVERLOAD alert +cm.backpressure.threshold=5 +# Define maximum CPU usage (%) before triggering backpressure OVERLOAD alert +cm.backpressure.threshold-cpu-overload=90 +# Define maximum CPU usage (%) before triggering backpressure IDLE alert +cm.backpressure.threshold-cpu-idle=10 +# Define the time window between the Backpressure reports, in seconds +cm.backpressure.time-window=10 +# Define the time window length, in seconds, during which, if CPU usage is consistently above the threshold, the system will trigger OVERLOAD alert +cm.backpressure.cpu-overload-duration=30 +# Define the time window length, in seconds, during which, if no request is made to the Service, the system will trigger IDLE alert +cm.backpressure.cpu-idle-duration=30 diff --git a/otdemo/otdemo_adapter.py b/otdemo/otdemo_adapter.py new file mode 100644 index 0000000..cdfa5dd --- /dev/null +++ b/otdemo/otdemo_adapter.py @@ -0,0 +1,59 @@ +""" +This module provides the CleverSwarm AMQP Adapter for handling API requests via AMQP. +The code IS MEANT to be part of CleverSwarm codebase, and has direct dependencies on the CleverSwarm codebase. +It is NOT MEANT to be used as a standalone module. It WILL NOT run outside the CleverSwarm endpoints context. +""" + +from threading import Thread +from typing import Any, List + +# CleverSwarm specific imports +from fastapi.routing import APIRoute + +from amqp.adapter.cleverthis_service_adapter import CleverThisServiceAdapter +from amqp.config.amq_configuration import AMQConfiguration +from amqp.service.amq_service import AMQService + +PREFERRED_SUFFIX = "_json" + + +# ================================================================================================= +# ======================== C l e v e r S w a r m A M Q A d a p t e r ======================== +# ================================================================================================= +class OTDemoAmqpAdapter(CleverThisServiceAdapter): + """ + CleverSwarm AMQP Adapter for CleverSwarm API. Implements the CleverSwarm specific logic that overrides + common logic defined by AMQ Adapter library. The specific logic here is: + + 1. Extract subject from JWT token and convert it to UserSchema to avail authorized user argument + 2. Find JSON wrapper for endpoint Callable, and if exists, use it instead of the original endpoint. + + """ + + def __init__(self, amq_configuration: AMQConfiguration, routes: List[APIRoute]): + super().__init__(amq_configuration, routes=routes, session=None) + _amq_service: AMQService = AMQService(amq_configuration, self) + self.thread = Thread(target=_amq_service.run) + self.thread.start() + + async def get_current_user(self, token: str) -> str: + """ + Overrides the default get_current_user method to use the token from the AMQP message. + In CleverSwarm context, it means simply to call provided get_current_user function. + + :param token: JWT token from the AMQP message + + :return: UserSchema object + """ + return token + + def get_endpoint(self, endpoint: Any) -> Any: + """ + If the specific endpoint has a matching JSON wrapper (identified as {endpoint.name}_json ), + then use it as preferred way to invoke the endpoint. + + :param endpoint: the Callable for the endpoint + + :return: new Callable representing the JSON wrapper endpoint, if it exists or the original Callable. + """ + return endpoint