Logfire

Logfire

Structured observability for Python using Pydantic Logfire - fast setup, powerful features, OpenTelemetry-compatible.

Quick Start

uv pip install logfire

import logfire

logfire.configure(service_name="my-api", service_version="1.0.0")
logfire.info("Application started")

Core Patterns

1. Service Configuration

Always set service metadata at startup:

import logfire

logfire.configure(
    service_name="backend",
    service_version="1.0.0",
    environment="production",
    console=False,           # Disable console output in production
    send_to_logfire=True,    # Send to Logfire platform
)

2. Framework Instrumentation

Instrument frameworks before creating clients/apps:

import logfire
from fastapi import FastAPI

# Configure FIRST
logfire.configure(service_name="backend")

# Then instrument
logfire.instrument_fastapi()
logfire.instrument_httpx()
logfire.instrument_sqlalchemy()

# Then create app
app = FastAPI()

3. Log Levels and Structured Logging

# All log levels (trace → fatal)
logfire.trace("Detailed trace", step=1)
logfire.debug("Debug context", variable=locals())
logfire.info("User action", action="login", success=True)
logfire.notice("Important event", event_type="milestone")
logfire.warn("Potential issue", threshold_exceeded=True)
logfire.error("Operation failed", error_code=500)
logfire.fatal("Critical failure", component="database")

# Python 3.11+ f-string magic (auto-extracts variables)
user_id = 123
status = "active"
logfire.info(f"User {user_id} status: {status}")
# Equivalent to: logfire.info("User {user_id}...", user_id=user_id, status=status)

# Exception logging with automatic traceback
try:
    risky_operation()
except Exception:
    logfire.exception("Operation failed", context="extra_info")

4. Manual Spans

# Spans for tracing operations
with logfire.span("Process order {order_id}", order_id="ORD-123"):
    logfire.info("Validating cart")
    # ... processing logic
    logfire.info("Order complete")

# Dynamic span attributes
with logfire.span("Database query") as span:
    results = execute_query()
    span.set_attribute("result_count", len(results))
    span.message = f"Query returned {len(results)} results"

5. Custom Metrics

# Counter - monotonically increasing
request_counter = logfire.metric_counter("http.requests", unit="1")
request_counter.add(1, {"endpoint": "/api/users", "method": "GET"})

# Gauge - current value
temperature = logfire.metric_gauge("temperature", unit="°C")
temperature.set(23.5)

# Histogram - distribution of values
latency = logfire.metric_histogram("request.duration", unit="ms")
latency.record(45.2, {"endpoint": "/api/data"})

6. LLM Observability

import logfire
from pydantic_ai import Agent

logfire.configure()
logfire.instrument_pydantic_ai()  # Traces all agent interactions

agent = Agent("openai:gpt-4o", system_prompt="You are helpful.")
result = agent.run_sync("Hello!")

7. Suppress Noisy Instrumentation

# Suppress entire scope (e.g., noisy library)
logfire.suppress_scopes("google.cloud.bigquery.opentelemetry_tracing")

# Suppress specific code block
with logfire.suppress_instrumentation():
    client.get("https://internal-healthcheck.local")  # Not traced

8. Sensitive Data Scrubbing

import logfire

# Add custom patterns to scrub
logfire.configure(
    scrubbing=logfire.ScrubbingOptions(
        extra_patterns=["api_key", "secret", "token"]
    )
)

# Custom callback for fine-grained control
def scrubbing_callback(match: logfire.ScrubMatch):
    if match.path == ("attributes", "safe_field"):
        return match.value  # Don't scrub this field
    return None  # Use default scrubbing

logfire.configure(
    scrubbing=logfire.ScrubbingOptions(callback=scrubbing_callback)
)

9. Sampling for High-Traffic Services

import logfire

# Sample 50% of traces
logfire.configure(sampling=logfire.SamplingOptions(head=0.5))

# Disable metrics to reduce volume
logfire.configure(metrics=False)

10. Testing

import logfire
from logfire.testing import CaptureLogfire

def test_user_creation(capfire: CaptureLogfire):
    create_user("Alice", "alice@example.com")
    
    spans = capfire.exporter.exported_spans
    assert len(spans) >= 1
    assert spans[0].attributes["user_name"] == "Alice"
    
    capfire.exporter.clear()  # Clean up for next test

Available Integrations

Common Pitfalls

References

• Configuration Options - All configure() parameters
• Integrations Guide - Framework-specific setup
• Metrics Guide - Counter, gauge, histogram, system metrics
• Advanced Patterns - Sampling, scrubbing, suppression, testing
• Pitfalls & Troubleshooting - Common issues and solutions
• Official Docs

Logfire Advanced Patterns

Sampling, scrubbing, suppression, testing, and custom configurations.

Sampling Strategies

Simple Ratio Sampling

import logfire

# Sample 50% of traces randomly
logfire.configure(sampling=logfire.SamplingOptions(head=0.5))

# Sample 10% of traces
logfire.configure(sampling=logfire.SamplingOptions(head=0.1))

Custom Sampler

from opentelemetry.sdk.trace.sampling import (
    ALWAYS_OFF,
    ALWAYS_ON,
    ParentBased,
    Sampler,
    TraceIdRatioBased,
)
import logfire


class MySampler(Sampler):
    def should_sample(self, parent_context, trace_id, name, *args, **kwargs):
        if name == "healthcheck":
            sampler = ALWAYS_OFF  # Never sample healthchecks
        elif name.startswith("internal/"):
            sampler = TraceIdRatioBased(0.01)  # 1% for internal
        elif name.startswith("api/"):
            sampler = TraceIdRatioBased(0.5)  # 50% for API
        else:
            sampler = ALWAYS_ON  # Sample everything else
        
        return sampler.should_sample(parent_context, trace_id, name, *args, **kwargs)

    def get_description(self):
        return "MySampler"


logfire.configure(
    sampling=logfire.SamplingOptions(
        head=ParentBased(MySampler())  # Respect parent sampling decisions
    )
)

Disable Metrics

# Reduce volume by disabling aggregate metrics
logfire.configure(metrics=False)

Sensitive Data Scrubbing

Default Scrubbing

Logfire automatically scrubs common patterns: password, secret, token, key, auth, credential, credit_card, ssn.

Add Custom Patterns

import logfire

logfire.configure(
    scrubbing=logfire.ScrubbingOptions(
        extra_patterns=["api_key", "private_key", "my_secret"]
    )
)

# These will be scrubbed
logfire.info("Request", data={
    "api_key": "sk-12345",  # → [REDACTED]
    "user": "alice",        # Kept
})

Custom Scrubbing Callback

import logfire

def scrubbing_callback(match: logfire.ScrubMatch):
    # Don't scrub known safe fields
    if match.path == ("attributes", "password_reset_requested"):
        return match.value  # Return original value
    
    # Custom redaction format
    if "email" in str(match.path):
        return "[EMAIL REDACTED]"
    
    return None  # Use default scrubbing

logfire.configure(
    scrubbing=logfire.ScrubbingOptions(callback=scrubbing_callback)
)

OTel Collector Scrubbing

For server-side scrubbing:

processors:
  attributes:
    actions:
      - key: session_id
        action: update
        value: "SCRUBBED"
      - pattern: "password"
        action: delete
  
  redaction:
    allow_all_keys: true
    blocked_values:
      - '[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}'  # Emails

pipelines:
  traces:
    processors: [attributes, redaction]

Suppression

Suppress by Scope

Suppress all spans/metrics from a specific OpenTelemetry scope:

import logfire

logfire.configure()

# Suppress noisy BigQuery auto-instrumentation
logfire.suppress_scopes("google.cloud.bigquery.opentelemetry_tracing")

# Suppress multiple scopes
logfire.suppress_scopes(
    "google.cloud.bigquery.opentelemetry_tracing",
    "some.other.noisy.scope",
)

Suppress Code Block

Suppress instrumentation for specific code:

import logfire
import httpx

logfire.configure()
logfire.instrument_httpx()

client = httpx.Client()

# This request IS traced
client.get("https://api.example.com/users")

# This request is NOT traced
with logfire.suppress_instrumentation():
    client.get("https://internal-healthcheck.local")
    client.get("https://another-internal-service.local")

# Tracing resumes here
client.get("https://api.example.com/orders")

Use cases:

• Health check endpoints
• Internal service calls
• High-frequency polling
• Debug/test requests

Testing

Pytest Fixture

import logfire
from logfire.testing import CaptureLogfire


def test_order_processing(capfire: CaptureLogfire):
    process_order("ORD-123")
    
    spans = capfire.exporter.exported_spans
    
    # Verify span was created
    assert len(spans) >= 1
    
    # Verify span attributes
    order_span = next(s for s in spans if "order" in s.name.lower())
    assert order_span.attributes["order_id"] == "ORD-123"
    
    # Clean up
    capfire.exporter.clear()


def test_multiple_operations(capfire: CaptureLogfire):
    logfire.info("First operation")
    assert len(capfire.exporter.exported_spans) == 1
    
    logfire.info("Second operation")
    assert len(capfire.exporter.exported_spans) == 2
    
    capfire.exporter.clear()
    assert len(capfire.exporter.exported_spans) == 0

Manual Test Setup

import logfire
from logfire.testing import TestExporter

def setup_test_logfire():
    exporter = TestExporter()
    logfire.configure(
        send_to_logfire=False,
        advanced=logfire.AdvancedOptions(
            additional_span_processors=[exporter]
        )
    )
    return exporter

# In tests
exporter = setup_test_logfire()
my_function()
assert len(exporter.exported_spans) == 1

Automatic Test Mode

Logfire automatically sets send_to_logfire=False when running under pytest.

LLM Cost Tracking

import logfire
from pydantic_ai import Agent

# Enable metrics collection within spans
logfire.configure(metrics=logfire.MetricsOptions(collect_in_spans=True))
logfire.instrument_pydantic_ai()

agent = Agent("gpt-4o")

with logfire.span("batch_processing"):
    # Token usage and costs are aggregated in parent span
    agent.run_sync("Question 1")
    agent.run_sync("Question 2")
    agent.run_sync("Question 3")

Distributed Tracing

Multi-Service Setup

Service A (API):

import logfire

logfire.configure(service_name="api-gateway")
logfire.instrument_fastapi()
logfire.instrument_httpx()

Service B (Backend):

import logfire

logfire.configure(service_name="order-service")
logfire.instrument_fastapi()
logfire.instrument_sqlalchemy()

Trace context propagates automatically via HTTP headers.

Manual Context Propagation

from opentelemetry import trace
from opentelemetry.propagate import inject, extract

# Inject context into headers
headers = {}
inject(headers)
# headers now contains: {"traceparent": "00-..."}

# Extract context from headers
context = extract(incoming_headers)
with trace.get_tracer(__name__).start_as_current_span("child", context=context):
    pass

Custom Span Processors

from opentelemetry.sdk.trace import SpanProcessor
from opentelemetry.sdk.trace.export import ReadableSpan
import logfire


class MyProcessor(SpanProcessor):
    def on_start(self, span, parent_context):
        # Add custom attribute to all spans
        span.set_attribute("custom.processor", "active")
    
    def on_end(self, span: ReadableSpan):
        if span.status.is_ok:
            print(f"Span {span.name} completed successfully")
    
    def shutdown(self):
        pass
    
    def force_flush(self, timeout_millis=30000):
        return True


logfire.configure(
    advanced=logfire.AdvancedOptions(
        additional_span_processors=[MyProcessor()]
    )
)

Baggage Propagation

from opentelemetry import baggage
from opentelemetry.context import attach, detach

# Set baggage (propagates across services)
ctx = baggage.set_baggage("tenant_id", "acme-corp")
token = attach(ctx)

try:
    # All spans in this context have access to tenant_id
    with logfire.span("tenant_operation"):
        tenant = baggage.get_baggage("tenant_id")
        logfire.info("Processing for tenant", tenant_id=tenant)
finally:
    detach(token)

Logfire Configuration Reference

Complete reference for logfire.configure() options.

Basic Configuration

import logfire

logfire.configure(
    service_name="my-api",           # Required: identifies your service
    service_version="1.0.0",          # Semantic version
    environment="production",         # dev/staging/production
    send_to_logfire=True,             # Send to Logfire platform
    token="your-project-token",       # Auth token (or use env var)
    console=True,                     # Print to console
    min_level="info",                 # Minimum log level
)

Log Levels

Logfire provides 7 log levels (lowest to highest severity):

# All log levels
logfire.trace("Step 1 of algorithm", step=1)
logfire.debug("Variable state", data=locals())
logfire.info("User logged in", user_id=123)
logfire.notice("Cache invalidated", reason="manual")
logfire.warn("Rate limit approaching", current=95, limit=100)
logfire.error("API call failed", status=500, retry=True)
logfire.fatal("Database connection lost", host="db.example.com")

# Dynamic log level
logfire.log("warn", "Dynamic level message")

# Exception logging (shortcut for error + traceback)
try:
    risky_operation()
except Exception:
    logfire.exception("Operation failed")  # Includes full traceback

f-string Magic (Python 3.11+)

user_id = 123
status = "active"

# Automatically extracts variable names
logfire.info(f"User {user_id} status: {status}")
# Equivalent to:
logfire.info("User {user_id} status: {status}", user_id=123, status="active")

Environment Variables

Set these instead of passing to configure():

export LOGFIRE_SERVICE_NAME=my-api
export LOGFIRE_SERVICE_VERSION=1.0.0
export LOGFIRE_ENVIRONMENT=production
export LOGFIRE_TOKEN=your-project-token

Configuration Parameters

Service Metadata

Output Control

Authentication

Console Options

# Disable console completely
logfire.configure(console=False)

# Custom console settings
logfire.configure(
    console={
        "colors": True,
        "include_timestamps": True,
        "verbose": False,
    }
)

Scrubbing Options

from logfire import ScrubbingOptions

logfire.configure(
    scrubbing=ScrubbingOptions(
        # Add custom patterns (combined with defaults)
        extra_patterns=["my_secret_pattern", "api_key"],
        
        # Custom callback for fine-grained control
        callback=my_scrubbing_callback,
    )
)

Default patterns automatically scrub: password, secret, token, key, auth, credential, credit_card, ssn.

Sampling Options

from logfire import SamplingOptions

# Simple ratio sampling
logfire.configure(sampling=SamplingOptions(head=0.5))  # 50%

# Custom sampler
from opentelemetry.sdk.trace.sampling import TraceIdRatioBased, ParentBased

logfire.configure(
    sampling=SamplingOptions(
        head=ParentBased(TraceIdRatioBased(0.1))  # 10% with parent context
    )
)

Metrics Options

from logfire import MetricsOptions

# Enable metrics collection within spans
logfire.configure(
    metrics=MetricsOptions(collect_in_spans=True)
)

# Disable metrics entirely
logfire.configure(metrics=False)

Advanced Options

from logfire import AdvancedOptions
from logfire.testing import TestExporter

exporter = TestExporter()

logfire.configure(
    advanced=AdvancedOptions(
        additional_span_processors=[exporter],
    )
)

Testing Configuration

When running under pytest, Logfire automatically sets send_to_logfire=False.

# Explicit test mode
logfire.configure(
    send_to_logfire=False,
    console=True,  # See output during tests
)

Multiple Configurations

For microservices or multi-tenant apps:

import logfire

# Primary configuration
logfire.configure(service_name="main-api")

# Create isolated instance
tenant_logfire = logfire.Logfire(
    service_name="tenant-service",
    environment="production",
)
tenant_logfire.info("Tenant-specific log")

OTLP Export (Vendor-Neutral)

For OpenTelemetry-compatible backends:

from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor

resource = Resource(attributes={"service.name": "my-service"})
trace.set_tracer_provider(TracerProvider(resource=resource))

exporter = OTLPSpanExporter(endpoint="https://your-otlp-backend/v1/traces")
trace.get_tracer_provider().add_span_processor(BatchSpanProcessor(exporter))

tracer = trace.get_tracer(__name__)
with tracer.start_as_current_span("operation"):
    pass

Logfire Integrations Guide

Detailed setup for each supported framework and library.

Web Frameworks

FastAPI

import logfire
from fastapi import FastAPI

logfire.configure(service_name="my-api")
logfire.instrument_fastapi()  # Global instrumentation

app = FastAPI()

# Or instrument specific app
logfire.instrument_fastapi(app)

Features:

• Automatic request/response spans
• Pydantic validation logging (422 errors)
• Exception tracking
• Request timing

Starlette

import logfire
from starlette.applications import Starlette

logfire.configure()
app = Starlette()
logfire.instrument_starlette(app)

Django

# settings.py
import logfire

logfire.configure(service_name="django-app")
logfire.instrument_django()

Flask

import logfire
from flask import Flask

logfire.configure()
app = Flask(__name__)
logfire.instrument_flask(app)

AIOHTTP Server

import logfire
from aiohttp import web

logfire.configure()
logfire.instrument_aiohttp_server()

async def hello(request):
    return web.Response(text="Hello, World!")

app = web.Application()
app.router.add_get("/", hello)

if __name__ == "__main__":
    web.run_app(app, host="localhost", port=8080)

ASGI (Generic)

import logfire
from my_asgi_app import app

logfire.configure()
logfire.instrument_asgi(app)

WSGI (Generic)

import logfire
from my_wsgi_app import app

logfire.configure()
logfire.instrument_wsgi(app)

HTTP Clients

HTTPX

import logfire
import httpx

logfire.configure()
logfire.instrument_httpx()

# All HTTPX requests are now traced
async with httpx.AsyncClient() as client:
    response = await client.get("https://api.example.com/users")

Requests

import logfire
import requests

logfire.configure()
logfire.instrument_requests()

response = requests.get("https://api.example.com/data")

AIOHTTP Client

import logfire
import aiohttp

logfire.configure()
logfire.instrument_aiohttp_client()

async with aiohttp.ClientSession() as session:
    async with session.get("https://api.example.com") as response:
        data = await response.text()

Databases

SQLAlchemy

import logfire
from sqlalchemy import create_engine

logfire.configure()

# Instrument specific engine
engine = create_engine("postgresql://user:pass@localhost/db")
logfire.instrument_sqlalchemy(engine=engine)

# Or instrument all engines globally
logfire.instrument_sqlalchemy()

Features:

• SQL query logging
• Parameter capture
• Duration tracking
• Row counts

Asyncpg

import logfire
import asyncpg

logfire.configure()
logfire.instrument_asyncpg()

conn = await asyncpg.connect("postgresql://user:pass@localhost/db")

Psycopg

import logfire

logfire.configure()
logfire.instrument_psycopg()

Redis

import logfire

logfire.configure()
logfire.instrument_redis()

PyMongo

import logfire

logfire.configure()
logfire.instrument_pymongo()

LLM Integrations

Pydantic AI (Recommended)

import logfire
from pydantic_ai import Agent

logfire.configure()
logfire.instrument_pydantic_ai()

agent = Agent("openai:gpt-4o", system_prompt="You are helpful.")
result = agent.run_sync("What is the capital of France?")

Traces:

• Agent interactions
• Tool calls
• Token usage
• Costs (with collect_in_spans=True)

OpenAI

import logfire
from openai import OpenAI

logfire.configure()
logfire.instrument_openai()

client = OpenAI()
response = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "Hello!"}]
)

Anthropic

import logfire
from anthropic import Anthropic

logfire.configure()
logfire.instrument_anthropic()

client = Anthropic()
response = client.messages.create(
    model="claude-3-sonnet",
    messages=[{"role": "user", "content": "Hello!"}]
)

LangChain

import os
import logfire

# Set environment variables BEFORE importing langchain
os.environ["LANGSMITH_OTEL_ENABLED"] = "true"
os.environ["LANGSMITH_TRACING"] = "true"

from langchain.agents import create_agent

logfire.configure()

agent = create_agent("openai:gpt-4o", tools=[my_tool])
result = agent.invoke({"messages": [{"role": "user", "content": "Hello"}]})

Mirascope

import logfire
from mirascope.core import anthropic, prompt_template
from mirascope.integrations.logfire import with_logfire

logfire.configure()

@with_logfire()
@anthropic.call("claude-3-5-sonnet")
@prompt_template("Recommend some {genre} books")
def recommend_books(genre: str): ...

response = recommend_books("fantasy")

MCP (Model Context Protocol)

import logfire
from mcp.server.fastmcp import FastMCP

logfire.configure(service_name="mcp-server")
logfire.instrument_mcp()

app = FastMCP()

@app.tool()
def add(a: int, b: int) -> int:
    return a + b

Task Queues

Celery

import logfire
from celery import Celery
from celery.signals import worker_init

@worker_init.connect()
def init_worker(*args, **kwargs):
    logfire.configure(service_name="celery-worker")
    logfire.instrument_celery()

app = Celery("tasks", broker="redis://localhost:6379/0")

@app.task
def add(x: int, y: int):
    return x + y

AWS Lambda

pip install logfire[aws-lambda]

import logfire

logfire.configure()
logfire.instrument_aws_lambda()

def lambda_handler(event, context):
    logfire.info("Processing Lambda event", event_type=event.get("type"))
    
    with logfire.span("Business logic"):
        result = process_event(event)
    
    return {"statusCode": 200, "body": result}

Features:

• Invocation details
• Duration tracking
• Cold start detection
• Error capture

Logging Libraries

Standard Logging

import logfire
import logging

logfire.configure()
logfire.instrument_logging()

logger = logging.getLogger(__name__)
logger.info("This goes to Logfire")

Structlog

import logfire
import structlog

logfire.configure()
logfire.instrument_structlog()

logger = structlog.get_logger()
logger.info("Structured log", user_id=123)

Loguru

import logfire
from loguru import logger

logfire.configure()
logfire.instrument_loguru()

logger.info("Loguru message")

Print Statements

import logfire

logfire.configure()
logfire.instrument_print()

# Now all print() calls are captured as logs
name = "World"
print("Hello", name)  # Logged with arguments

# Or use as context manager
with logfire.instrument_print():
    print("This is logged")
print("This is NOT logged")

Pydantic Validation

import logfire
from pydantic import BaseModel

logfire.configure()
logfire.instrument_pydantic()

class User(BaseModel):
    name: str
    email: str

# Validation is now traced
user = User(name="Alice", email="alice@example.com")

System Metrics

import logfire

logfire.configure()
logfire.instrument_system_metrics()

# Captures: CPU, memory, disk, network metrics

JavaScript/TypeScript

Next.js

import * as logfire from "logfire";

logfire.configure({
  serviceName: "my-nextjs-app",
  environment: "production",
});

Cloudflare Workers

import * as logfire from "logfire";
import { instrument } from "@pydantic/logfire-cf-workers";

const handler = {
  async fetch(): Promise<Response> {
    logfire.info("Worker invoked");
    return new Response("Hello!");
  },
} satisfies ExportedHandler;

export default instrument(handler, {
  service: { name: "my-worker", version: "1.0.0" },
});

Logfire Metrics Reference

Complete guide to creating and using custom metrics with Logfire.

Metric Types

Logfire supports three metric types aligned with OpenTelemetry:

Counter

Tracks cumulative values that only increase (or reset on restart).

import logfire

logfire.configure()

# Create counter
request_counter = logfire.metric_counter(
    "http.requests.total",
    unit="1",
    description="Total HTTP requests received"
)

# Increment by 1
request_counter.add(1)

# Increment with attributes (labels)
request_counter.add(1, {
    "endpoint": "/api/users",
    "method": "GET",
    "status_code": 200
})

# Increment by custom amount
request_counter.add(5, {"batch": "true"})

Counter Examples

# Exception counter
exception_counter = logfire.metric_counter(
    "exceptions.caught",
    unit="1",
    description="Number of exceptions caught"
)

try:
    risky_operation()
except Exception as e:
    exception_counter.add(1, {"exception_type": type(e).__name__})
    logfire.exception("Operation failed")

# Message counter
messages_sent = logfire.metric_counter("messages.sent", unit="1")

def send_message(channel: str):
    # ... send message
    messages_sent.add(1, {"channel": channel})

Gauge

Tracks current value that can increase or decrease.

import logfire

logfire.configure()

# Create gauge
temperature = logfire.metric_gauge(
    "temperature",
    unit="°C",
    description="Current temperature reading"
)

# Set current value
temperature.set(23.5)

# Set with attributes
temperature.set(25.0, {"location": "server_room", "sensor": "A1"})

# Update over time
def update_temperature(value: float, location: str):
    temperature.set(value, {"location": location})

Gauge Examples

# Active connections
active_connections = logfire.metric_gauge(
    "connections.active",
    unit="1",
    description="Current active connections"
)

def on_connect(client_id: str):
    # Logic to get current count
    current = get_connection_count()
    active_connections.set(current)

# Queue size
queue_size = logfire.metric_gauge("queue.size", unit="1")

def process_queue():
    while True:
        queue_size.set(len(pending_items))
        item = pending_items.pop()
        process(item)

# Memory usage
memory_usage = logfire.metric_gauge("memory.used", unit="bytes")

def report_memory():
    import psutil
    memory_usage.set(psutil.Process().memory_info().rss)

Histogram

Tracks distribution of values for statistical analysis.

import logfire
import time

logfire.configure()

# Create histogram
latency = logfire.metric_histogram(
    "http.request.duration",
    unit="ms",
    description="HTTP request latency distribution"
)

# Record a value
latency.record(45.2)

# Record with attributes
latency.record(123.5, {
    "endpoint": "/api/users",
    "method": "GET"
})

# Timing pattern
def timed_operation():
    start = time.time()
    try:
        perform_work()
    finally:
        duration_ms = (time.time() - start) * 1000
        latency.record(duration_ms)

Histogram Examples

# Response size
response_size = logfire.metric_histogram(
    "http.response.size",
    unit="bytes",
    description="HTTP response body size"
)

def send_response(data: bytes):
    response_size.record(len(data))
    return data

# Database query time
query_duration = logfire.metric_histogram(
    "db.query.duration",
    unit="ms",
    description="Database query execution time"
)

async def execute_query(sql: str):
    start = time.time()
    result = await db.execute(sql)
    query_duration.record(
        (time.time() - start) * 1000,
        {"table": extract_table_name(sql)}
    )
    return result

# File processing
file_size = logfire.metric_histogram("file.processed.size", unit="bytes")

def process_file(path: str):
    size = os.path.getsize(path)
    file_size.record(size, {"extension": path.split(".")[-1]})
    # ... process file

System Metrics

Automatically collect system-level metrics:

import logfire

logfire.configure()

# Collect all system metrics
logfire.instrument_system_metrics()

# Selective collection
logfire.instrument_system_metrics(
    metrics={"system.cpu.utilization": None},
    base=None  # Only collect explicitly specified
)

# Collect with exclusions
logfire.instrument_system_metrics(
    metrics={"system.disk.operations": ["read"]},  # Only reads
    base="full"  # Start with all metrics
)

Available System Metrics

Metrics in Spans

Aggregate metrics within span context:

import logfire
from pydantic_ai import Agent

# Enable metrics collection in spans
logfire.configure(
    metrics=logfire.MetricsOptions(collect_in_spans=True)
)
logfire.instrument_pydantic_ai()

agent = Agent("gpt-4o")

# Token usage aggregated in parent span
with logfire.span("batch_processing"):
    agent.run_sync("Question 1")
    agent.run_sync("Question 2")
    agent.run_sync("Question 3")
    # Parent span shows total tokens/costs

Disable Metrics

# Disable all metrics
logfire.configure(metrics=False)

# Keep traces, disable only aggregate metrics from integrations
logfire.configure(
    metrics=logfire.MetricsOptions(
        include_metrics=False
    )
)

Best Practices

Naming Conventions

# Good: hierarchical, lowercase, dots as separators
logfire.metric_counter("http.requests.total")
logfire.metric_histogram("db.query.duration")
logfire.metric_gauge("cache.size.bytes")

# Bad: inconsistent, unclear
logfire.metric_counter("RequestCount")
logfire.metric_histogram("time")

Cardinality Control

# Good: bounded attribute values
request_counter.add(1, {
    "method": "GET",           # Limited set
    "status_class": "2xx",     # Grouped status codes
    "endpoint_pattern": "/api/users/{id}"  # Pattern, not actual ID
})

# Bad: unbounded attribute values (causes storage explosion)
request_counter.add(1, {
    "user_id": user_id,        # Unique per user
    "request_id": request_id,  # Unique per request
    "full_path": "/api/users/12345"  # Contains variable data
})

Units

Use standard units:

Logfire Pitfalls & Troubleshooting

Common issues and their solutions.

Configuration Issues

Missing Service Name

Symptom: Spans appear as "unknown_service" in UI, hard to filter.

Fix:

# Always set service_name
logfire.configure(service_name="my-api")

Late Instrumentation

Symptom: No spans captured for some requests.

Cause: Clients/apps created before configure() is called.

Fix:

import logfire

# 1. Configure FIRST
logfire.configure(service_name="backend")

# 2. Instrument SECOND
logfire.instrument_fastapi()
logfire.instrument_httpx()

# 3. Create clients/apps THIRD
from fastapi import FastAPI
app = FastAPI()

Console Noise in Production

Symptom: Stdout polluted with trace output.

Fix:

logfire.configure(
    console=False,  # Disable console in production
    send_to_logfire=True,
)

Performance Issues

High-Cardinality Attributes

Symptom: Storage costs explode, slow queries.

Cause: Using unbounded values as attributes.

Bad:

logfire.info("Request", body=full_request_body)  # Huge, unique
logfire.info("User", email=user_email)            # High cardinality

Good:

logfire.info("Request", request_id=request_id, size_bytes=len(body))
logfire.info("User", user_id=user_id)  # Use IDs, not raw values

Too Many Spans

Symptom: High costs, slow UI, drowning in data.

Fixes:

# 1. Enable sampling
logfire.configure(sampling=logfire.SamplingOptions(head=0.1))

# 2. Suppress noisy operations
with logfire.suppress_instrumentation():
    frequent_healthcheck()

# 3. Suppress entire scopes
logfire.suppress_scopes("noisy.library.scope")

# 4. Disable metrics if not needed
logfire.configure(metrics=False)

Memory Growth

Symptom: Application memory grows over time.

Cause: Unbounded span processors or exporters.

Fix:

# Use batch processors (default) with limits
from opentelemetry.sdk.trace.export import BatchSpanProcessor

processor = BatchSpanProcessor(
    exporter,
    max_queue_size=2048,
    max_export_batch_size=512,
)

Tracing Issues

Missing Child Spans

Symptom: Parent span exists but children are missing.

Cause: Child operations happen in different thread/context.

Fix:

from opentelemetry import trace, context

# Capture current context
current_context = context.get_current()

def background_task():
    # Restore context in new thread
    token = context.attach(current_context)
    try:
        with logfire.span("child_operation"):
            pass
    finally:
        context.detach(token)

Spans Not Appearing

Symptom: logfire.span() calls produce no output.

Causes & Fixes: 1. Not configured: Call logfire.configure() first 2. Sampled out: Check sampling settings 3. Suppressed: Check for suppress_instrumentation() context 4. Not flushed: Add explicit flush on shutdown

import atexit
from opentelemetry import trace

atexit.register(lambda: trace.get_tracer_provider().force_flush())

Broken Trace Context

Symptom: Traces don't connect across services.

Cause: Headers not propagated.

Fix: Ensure instrumented HTTP clients propagate context:

logfire.instrument_httpx()  # Handles propagation automatically

Scrubbing Issues

Sensitive Data Leaking

Symptom: PII visible in logs.

Fix:

logfire.configure(
    scrubbing=logfire.ScrubbingOptions(
        extra_patterns=["email", "phone", "address", "ssn"]
    )
)

Over-Scrubbing

Symptom: Non-sensitive data being redacted.

Fix:

def scrubbing_callback(match: logfire.ScrubMatch):
    # Whitelist safe fields
    safe_fields = ["password_changed", "reset_password_requested"]
    if any(f in str(match.path) for f in safe_fields):
        return match.value
    return None

logfire.configure(
    scrubbing=logfire.ScrubbingOptions(callback=scrubbing_callback)
)

Testing Issues

Tests Sending Real Data

Symptom: Test runs appear in production Logfire.

Fix: Logfire auto-disables under pytest, but verify:

logfire.configure(send_to_logfire=False)

Flaky Span Assertions

Symptom: Tests randomly fail on span counts.

Cause: Background instrumentation creating extra spans.

Fix:

def test_specific_operation(capfire: CaptureLogfire):
    capfire.exporter.clear()  # Clear any setup spans
    
    my_operation()
    
    # Filter for specific spans
    my_spans = [s for s in capfire.exporter.exported_spans 
                if "my_operation" in s.name]
    assert len(my_spans) == 1

Integration-Specific Issues

FastAPI 422 Validation Errors Not Logged

Symptom: Pydantic validation failures don't appear.

Fix: Use instrument_fastapi(app) not just instrument_fastapi():

logfire.instrument_fastapi(app)  # Pass the app instance

SQLAlchemy Queries Not Traced

Symptom: Database queries missing from traces.

Fix: Instrument the specific engine:

engine = create_engine(url)
logfire.instrument_sqlalchemy(engine=engine)  # Pass engine explicitly

Async Code Not Traced

Symptom: Async operations missing from traces.

Cause: Context not propagated across async boundaries.

Fix: Use async-aware instrumentation:

logfire.instrument_httpx()  # Supports AsyncClient
logfire.instrument_asyncpg()  # Async PostgreSQL

Debug Checklist

When spans aren't appearing:

1. Check configuration: logfire.configure() called? 2. Check ordering: Configure → Instrument → Create clients 3. Check sampling: Is sampling set too aggressively? 4. Check suppression: Inside suppress_instrumentation() context? 5. Check min_level: Is min_level higher than your log level? 6. Check console: Is console=True for debugging? 7. Force flush: Add explicit flush to ensure spans export

# Debug configuration
logfire.configure(
    service_name="debug-service",
    console=True,
    min_level="trace",
    send_to_logfire=True,
)

# Debug logging
logfire.debug("Configuration complete")