Fix dependency versions

.forgejo/workflows/publish-docker.yaml

@@ -0,0 +1,41 @@
+on: [push]
+
+env:
+  REGISTRY_URL: "git.cleverthis.com"
+  REPOSITORY: "clevermicro/amq-adapter-python"
+  DOCKER_HOST: "tcp://dind:2375"
+
+jobs:
+  build-and-push:
+    runs-on: general
+    services:
+      dind:
+        image: docker:dind
+        cmd:
+          - dockerd
+          - -H
+          - tcp://0.0.0.0:2375
+          - --tls=false
+    container:
+      image: "ghcr.io/catthehacker/ubuntu:js-22.04"
+    steps:
+      - name: checkout
+        uses: actions/checkout@v4
+        with:
+          ref: ${{ github.ref_name }}
+
+      - name: docker login
+        uses: docker/login-action@v3
+        with:
+          registry: ${{ env.REGISTRY_URL }}
+          username: ${{ secrets.REGISTRY_USER }}
+          password: ${{ secrets.REGISTRY_PASSWORD }}
+
+      - name: docker build & push
+        uses: docker/build-push-action@v6
+        with:
+          context: .
+          file: ./Dockerfile
+          no-cache: true
+          push: true
+          tags: ${{ env.REGISTRY_URL }}/${{ env.REPOSITORY }}:latest

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.34.1 \
+    opentelemetry-sdk==1.34.1 \
+    opentelemetry-exporter-otlp-proto-grpc==1.34.1 \
+    opentelemetry-exporter-prometheus==0.55b1 \
+    opentelemetry-instrumentation-fastapi==0.55b1 \
+    opentelemetry-instrumentation-httpx==0.55b1 \
+    protobuf==5.28.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"]

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.

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

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