FastAPI Observability Demo Application

FastAPI instrumentation only covers the ASGI app's request/response cycle. To maintain a single trace across multiple services when making outbound HTTP calls, you must manually inject the current span context into the request headers using opentelemetry.propagate.inject or use the HTTPXClientInstrumentor to automate the process.

# Manual injection using inject()
from opentelemetry.propagate import inject

@app.get("/chain")
async def chain(response: Response):
    headers = {}
    inject(headers)  # inject trace info to header
    async with httpx.AsyncClient() as client:
        await client.get(f"http://other-service:8000/", headers=headers)

# Automatic injection using HTTPX instrumentation
import httpx
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor

HTTPXClientInstrumentor().instrument()

@app.get("/chain")
async def chain(response: Response):
    async with httpx.AsyncClient() as client:
        await client.get(f"http://other-service:8000/") # Headers injected automatically

The project demonstrates how to navigate between the three pillars of observability using Trace IDs and exemplars in Grafana:

• Metrics to Traces: Identify a specific request by clicking an exemplar (a specific trace ID associated with a metric data point) in a Prometheus metric graph. This allows you to jump directly to the corresponding trace in Tempo.
  • Example Query: histogram_quantile(.99,sum(rate(fastapi_requests_duration_seconds_bucket{app_name="app-a", path!="/metrics"}[1m])) by(path, le))
• Traces to Logs: From a span in Tempo, use the Trace ID and tags (such as service.name) to query Loki and find the exact logs produced during that specific trace.
• Logs to Traces: Use a regex-defined Trace ID within a Loki log entry to jump back to the distributed trace in Tempo.

# Example query to find latency quantiles and identify exemplars
histogram_quantile(.99,sum(rate(fastapi_requests_duration_seconds_bucket{app_name="app-a", path!="/metrics"}[1m])) by(path, le))

To observe a FastAPI application using the three pillars of observability (Traces with Tempo, Metrics with Prometheus, and Logs with Loki), follow these steps:

1. Install the Loki Docker Driver: This allows Docker to send logs directly to Loki.
   • For ARM64 architectures, use the arm64 image.
   • For AMD64 architectures, use the amd64 image.
2. Start the stack: Use docker-compose up -d to launch all observability services.
   • Troubleshooting: If you see an error stating the loki plugin is disabled, run docker plugin enable loki.
3. Generate traffic: Use tools like siege, curl, Locust, or k6 to send requests to the FastAPI application (typically at http://localhost:8000) so that data is collected.
4. View results: Access the predefined FastAPI Observability dashboard at http://localhost:3000/ (default credentials: admin:admin).

# 1. Install Loki Docker Driver (AMD64 example)
docker plugin install grafana/loki-docker-driver:3.3.2-amd64 --alias loki --grant-all-permissions

# 2. Start services
docker-compose up -d

# If plugin is disabled:
docker plugin enable loki

# 3. Send requests (using locust as an example)
locust -f locustfile.py --headless --users 10 --spawn-rate 1 -H http://localhost:8000

To link metrics to traces in Grafana, use Prometheus exemplars. This involves retrieving the current trace_id from the OpenTelemetry span and adding it as an exemplar when observing a metric (e.g., a Histogram).

Important: When exposing metrics, you must use generate_latest and CONTENT_TYPE_LATEST from prometheus_client.openmetrics.exposition instead of the standard prometheus_client module to ensure the exemplars are correctly included in the output.

# Adding exemplar to a metric
from opentelemetry import trace
from prometheus_client import Histogram

REQUESTS_PROCESSING_TIME = Histogram(
    "fastapi_requests_duration_seconds",
    "Histogram of requests processing time by path (in seconds)",
    ["method", "path", "app_name"],
)

span = trace.get_current_span()
trace_id = trace.format_trace_id(span.get_span_context().trace_id)

REQUESTS_PROCESSING_TIME.labels(
    method=method, path=path, app_name=self.app_name
).observe(after_time - before_time, exemplar={'TraceID': trace_id})

# Exposing metrics correctly for OpenMetrics support
from prometheus_client import REGISTRY
from prometheus_client.openmetrics.exposition import CONTENT_TYPE_LATEST, generate_latest

def metrics(request: Request) -> Response:
    return Response(generate_latest(REGISTRY), headers={"Content-Type": CONTENT_TYPE_LATEST})

To enable observability in a FastAPI application, use a setup function to configure the OpenTelemetry TracerProvider, attach an OTLPSpanExporter to send traces to a backend like Tempo, and instrument the FastAPI app. You can also enable log correlation to include trace_id and span_id in your logs using LoggingInstrumentor.

# fastapi_app/utils.py

def setting_otlp(app: ASGIApp, app_name: str, endpoint: str, log_correlation: bool = True) -> None:
    # set the service name to show in traces
    resource = Resource.create(attributes={
        "service.name": app_name
    })

    # set the tracer provider
    tracer = TracerProvider(resource=resource)
    trace.set_tracer_provider(tracer)

    tracer.add_span_processor(BatchSpanProcessor(
        OTLPSpanExporter(endpoint=endpoint)))

    if log_correlation:
        LoggingInstrumentor().instrument(set_logging_format=True)

    FastAPIInstrumentor.instrument_app(app, tracer_provider=tracer)

The observability stack uses a custom Docker logging driver configuration to send logs to Loki. This is defined via the x-logging anchor in the docker-compose.yaml file.

To enable this logging for your services, apply the *default-logging anchor to the logging key. This configuration includes:

• driver: loki
• loki-url: http://localhost:3100/api/prom/push
• loki-pipeline-stages: A pipeline that handles multiline logs (using a timestamp regex) and parses the log line into time and message fields using a regex expression.

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:
  app-a:
    logging: *default-logging

To enable seamless navigation between metrics, traces, and logs, configure your Grafana data sources with specific linking settings:

1. Prometheus (Exemplars): Set exemplarTraceIdDestinations to map the TraceID label to your Tempo datasource.
2. Tempo (Trace to Logs): Configure tracesToLogsV2 with datasourceUid (e.g., loki) and tags to allow jumping from a trace to relevant logs.
3. Loki (Derived Fields): Use derivedFields with a matcherRegex to extract the trace_id from log lines and create a link to Tempo.

# Prometheus Exemplar Config
jsonData:
  exemplarTraceIdDestinations:
     - datasourceUid: tempo
        name: TraceID

# Tempo Trace-to-Logs Config
jsonData:
  tracesToLogsV2:
    datasourceUid: loki
    tags:
      - key: service.name
        value: compose_service

# Loki Derived Fields Config
jsonData:
  derivedFields:
    - datasourceUid: tempo
      matcherRegex: (?:trace_id)=(\w+)
      matcherType: regex
      name: TraceID
      url: $${__value.raw}