Run Fastapi-observability using Gunicorn and multiple uvicorn-workers

[denis-k2]

Is it possible to run fastapi-observability using Gunicorn and multiple uvicorn-workers?
Could you share an example code?
I try to use this manual (+ this), but it doesn’t work for me.

[daler-api]

Hi. Yes, it's possible.
Just inside docker-compose.yml in the sections that start with app, uncomment the line build
and inside your application, remove the section if __name__ == '__main__'
and change it to

class StandaloneApplication(BaseApplication):
    def __init__(self, application: Callable, options: Dict[str, Any] = None):
        self.options = options or {}
        self.application = application
        super().__init__()

    def load_config(self):
        config = {
            key: value
            for key, value in self.options.items()
            if key in self.cfg.settings and value is not None
        }
        For key, value in config.items():
            self.cfg.set(key.lower(), value)

    def load(self):
        return self.application


if __name__ == "__main__":
    options = {
        "bind": "%s:%s" % ("0.0.0.0.0", "8000"),
        "workers": number_of_workers(),
        "worker_class": "uvicorn.workers.UvicornWorker",
    }
    StandaloneApplication(app, options).run()

[daler-api]

The full version will look like this:

import logging
import multiprocessing
import os
import random
import time
from typing import Optional, Callable, Dict, Any

import httpx
import uvicorn
from fastapi import FastAPI, Response
from gunicorn.app.base import BaseApplication
from opentelemetry.propagate import inject
from utils import PrometheusMiddleware, metrics, setting_otlp

APP_NAME = os.environ.get("APP_NAME", "app")
EXPOSE_PORT = os.environ.get("EXPOSE_PORT", 8000)
OTLP_GRPC_ENDPOINT = os.environ.get("OTLP_GRPC_ENDPOINT", "http://tempo:4317")

TARGET_ONE_HOST = os.environ.get("TARGET_ONE_HOST", "app-b")
TARGET_TWO_HOST = os.environ.get("TARGET_TWO_HOST", "app-c")

app = FastAPI()

# Setting metrics middleware
app.add_middleware(PrometheusMiddleware, app_name=APP_NAME)
app.add_route("/metrics", metrics)

# Setting OpenTelemetry exporter
setting_otlp(app, APP_NAME, OTLP_GRPC_ENDPOINT)


class EndpointFilter(logging.Filter):
    # Uvicorn endpoint access log filter
    def filter(self, record: logging.LogRecord) -> bool:
        return record.getMessage().find("GET /metrics") == -1


# Filter out /endpoint
logging.getLogger("uvicorn.access").addFilter(EndpointFilter())


@app.get("/")
async def read_root():
    logging.error("Hello World")
    return {"Hello": "World"}


@app.get("/items/{item_id}")
async def read_item(item_id: int, q: Optional[str] = None):
    logging.error("items")
    return {"item_id": item_id, "q": q}


@app.get("/io_task")
async def io_task():
    time.sleep(1)
    logging.error("io task")
    return "IO bound task finish!"


@app.get("/cpu_task")
async def cpu_task():
    for i in range(1000):
        _ = i * i * i
    logging.error("cpu task")
    return "CPU bound task finish!"


@app.get("/random_status")
async def random_status(response: Response):
    response.status_code = random.choice([200, 200, 300, 400, 500])
    logging.error("random status")
    return {"path": "/random_status"}


@app.get("/random_sleep")
async def random_sleep(response: Response):
    time.sleep(random.randint(0, 5))
    logging.error("random sleep")
    return {"path": "/random_sleep"}


@app.get("/error_test")
async def error_test(response: Response):
    logging.error("got error!!!!")
    raise ValueError("value error")


@app.get("/chain")
async def chain(response: Response):
    headers = {}
    inject(headers)  # inject trace info to header
    logging.critical(headers)

    async with httpx.AsyncClient() as client:
        await client.get(
            "http://localhost:8000/",
            headers=headers,
        )
    async with httpx.AsyncClient() as client:
        await client.get(
            f"http://{TARGET_ONE_HOST}:8000/io_task",
            headers=headers,
        )
    async with httpx.AsyncClient() as client:
        await client.get(
            f"http://{TARGET_TWO_HOST}:8000/cpu_task",
            headers=headers,
        )
    logging.info("Chain Finished")
    return {"path": "/chain"}


def number_of_workers():
    return multiprocessing.cpu_count()


class StandaloneApplication(BaseApplication):
    def __init__(self, application: Callable, options: Dict[str, Any] = None):
        self.options = options or {}
        self.application = application
        super().__init__()

    def load_config(self):
        config = {
            key: value
            for key, value in self.options.items()
            if key in self.cfg.settings and value is not None
        }
        for key, value in config.items():
            self.cfg.set(key.lower(), value)

    def load(self):
        return self.application


if __name__ == "__main__":
    options = {
        "bind": "%s:%s" % ("0.0.0.0", "8000"),
        "workers": number_of_workers(),
        "worker_class": "uvicorn.workers.UvicornWorker",
    }
    StandaloneApplication(app, options).run()

[denis-k2]

Thank you @daler-api for your reply.
When I run this code, the result is that the metrics (Total Requests, Requests Count ...) are not summarized, but displayed separately for each worker.
I get the same result by removing one line if __name__ == '__main__' from the original fastapi-observability and making docker command: gunicorn main:app -w N -k uvicorn.workers.UvicornWorker.

Using links to Prometeus FastAPI + Gunicornl and Multiprocess Mode I managed to get *.db-files in PROMETHEUS_MULTIPROC_DIR directory but I don't know how to pass them to Prometheus.

[faradox]

I'm having the same issue that the instrumentation works but the metrics are tracked separately for each worker. As far as I understand, I need to use gunicorn's post_fork() hook as mentioned in open-telemetry/opentelemetry-python-contrib#385 (comment) and the opentelemetry documentation but I can't put that together with the FastAPIInstrumentor.instrument_app(app, tracer_provider=tracer) call as shown in your example because the worker does not seem have access to the app object post-fork...

[timesbyusman]

Hi @blueswen , I really appreciate your work.

I am first time adding observability into any project, so kind of new in this domain.

I am trying to do something similar (integrating fastapi-observability) into my 3 micro services, I got it to work so that I am able to see metrics for my 3 services on dashboard. The problem I am facing is that logs are not showing on dashboard. I have attached the screenshot for referenece.

I am using loki driver to push logs to loki as mentioned in the readme, when using Explore on grafana i am able to see logs for services.

I used the same utils.py that is available in the repository

auth/main.py

import multiprocessing
import os
import logging
import uvicorn
from fastapi import FastAPI, status
from fastapi.openapi.utils import get_openapi
from fastapi.responses import JSONResponse
from fastapi.routing import APIRoute
from starlette.middleware.cors import CORSMiddleware
from gunicorn.app.base import BaseApplication
from typing import Optional, Callable, Dict, Any
from auth.app.api.api import router
from shared.core.config import settings
from shared.middlewares.device_verification import DeviceTypeMiddleware
from shared.middlewares.logging import LoggingMiddleware
from shared.utils import split_cors_origins
from shared.monitoring.new_utils import PrometheusMiddleware, setting_otlp, metrics
from shared.monitoring.logging_setup import setup_logging

OTLP_GRPC_ENDPOINT = os.environ.get("OTLP_GRPC_ENDPOINT", "tempo:4317")
LOG_LEVEL = os.environ.get("LOG_LEVEL", "INFO")

setup_logging("auth", LOG_LEVEL)
logger = logging.getLogger(__name__)

def custom_generate_unique_id(route: APIRoute) -> str:
    return f"{route.tags[0]}-{route.name}" if route.tags else route.name


app = FastAPI(
    title=f"{settings.PROJECT_NAME} - SSO",
    openapi_url=f"{settings.API_V1_STR}/openapi.json",
    docs_url="/docs",
    redoc_url="/redoc",
    generate_unique_id_function=custom_generate_unique_id,
)

app.add_middleware(PrometheusMiddleware, app_name='auth')
app.add_route("/metrics", metrics)

# Setting OpenTelemetry exporter
setting_otlp(app, 'auth', OTLP_GRPC_ENDPOINT)

class EndpointFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        return record.getMessage().find("GET /metrics") == -1

logging.getLogger("uvicorn.access").addFilter(EndpointFilter())


# setup_monitoring(app, 'auth', OTLP_GRPC_ENDPOINT, LOG_LEVEL)

logger.info("Auth service initialized")
logger.debug("Debug log from auth application")
logger.info("Info log from auth application")
logger.warning("Warning log from auth application")
logger.error("Error log from auth application")
logger.critical("Critical error from auth application")


@app.get("/health")
async def health_check():
    logger.info("Health check endpoint called")
    return JSONResponse(
        content={"status": "OK"},
        status_code=status.HTTP_200_OK
    )

app.add_middleware(LoggingMiddleware) #my request logging middleware


app.include_router(router, prefix=settings.API_V1_STR) #adding some routers


class StandaloneApplication(BaseApplication):
    def __init__(self, application: Callable, options: Dict[str, Any] = None):
        self.options = options or {}
        self.application = application
        super().__init__()

    def load_config(self):
        config = {
            key: value
            for key, value in self.options.items()
            if key in self.cfg.settings and value is not None
        }
        for key, value in config.items():
            self.cfg.set(key.lower(), value)

    def load(self):
        return self.application

def number_of_workers():
    return multiprocessing.cpu_count()

if __name__ == "__main__":
    logger.info(f"Starting auth service with {number_of_workers()} workers")
    options = {
        "bind": "%s:%s" % ("0.0.0.0", "8001"),
        "workers": number_of_workers(),
        "worker_class": "uvicorn.workers.UvicornWorker",
        "accesslog": "-",
        "errorlog": "-",
        "loglevel": LOG_LEVEL.lower(),
    }
    StandaloneApplication(app, options).run()

docker-compose.yml

x-logging: &default-logging
  driver: loki
  options:
    loki-url: 'http://localhost:3100/api/prom/push'
    loki-pipeline-stages: |
      - multiline:
          firstline: '^\d{4}-\d{2}-\d{2} \d{1,2}:\d{2}:\d{2}'
          max_wait_time: 3s
      - regex:
          expression: '^(?P<time>\d{4}-\d{2}-\d{2} \d{1,2}:\d{2}:\d{2},\d{3}) (?P<message>(?s:.*))$$'

services:
  loki:
    image: grafana/loki:3.5.0
    command: -config.file=/etc/loki/local-config.yaml
    ports:
      - "3100:3100"


  auth:
    build:
      context: .
      dockerfile: auth/Dockerfile
    restart: always
    command: bash -c "alembic upgrade head && python initial_data.py && gunicorn auth.app.main:app --workers 4 --worker-class uvicorn.workers.UvicornWorker --bind 0.0.0.0:8001"
    depends_on:
      postgres:
        condition: service_healthy
      redis:
        condition: service_healthy
      celery:
        condition: service_started
    ports:
      - "8001:8001"
    env_file:
      - .env
    environment:
      - LOG_LEVEL=DEBUG
    volumes:
      - geoip-data:/usr/share/GeoIP
    logging: *default-logging

#some more services here

prometheus:
    image: prom/prometheus:v3.3.1
    ports:
      - "9090:9090"
    volumes:
      - ./shared/monitoring/prometheus:/workspace
    command:
      - --config.file=/workspace/prometheus.yml
      - --enable-feature=exemplar-storage
    depends_on:
      - loki
    logging: *default-logging

  tempo:
    image: grafana/tempo:2.7.2
    command: [ "-config.file=/etc/tempo.yml", "--target=all", "--storage.trace.backend=local", "--storage.trace.local.path=/var/tempo", "--auth.enabled=false" ]
    ports:
      - "4317:4317"
      - "4318:4318"
    volumes:
      - ./shared/monitoring/tempo/tempo.yml:/etc/tempo.yml
    depends_on:
      - loki
    logging: *default-logging

  grafana:
    image: grafana/grafana:12.0.0
    ports:
      - "3000:3000"
    volumes:
      - ./shared/monitoring/grafana/:/etc/grafana/provisioning/datasources
      - ./shared/monitoring/dashboards.yaml:/etc/grafana/provisioning/dashboards/dashboards.yaml
      - ./shared/monitoring/dashboards:/etc/grafana/dashboards
    depends_on:
      - loki
      - prometheus
    logging: *default-logging

volumes:
  postgres_data:
  geoip-data:

NOTE: I got the the .yml and .yaml files from the etc folder of this repo and used them.

Can anyone guide me what am I doing wrong...! I am lost here.