How to "conditional log"?

[mmzeynalli]

Question

So, I am using FastAPI and I have connected Logfire, everything is good.

However, I have many successful GET requests that just takes up space, and it is kind of hard to find errors in interface as well. I want to be able to dump/push logs based on:

1. Request method: GET, POST, PUT
2. Response status code: For example, ignore all GETs < 400
3. Manually forced log/nolog routes: Lets say, @force_log, @nolog decorators, which set flag for the function.

I tried to do with SamplingOption or additional_span_processors, but I do not think I can do option 3, with these.

My Example code:

"""Import that makes the project run"""

# Get the tracer provider
import abc
import enum
import os
from contextvars import ContextVar
from functools import wraps
from logging import getLogger
from types import MappingProxyType
from typing import Callable, Optional, Sequence, Set

import logfire
from logfire._internal.exporters.wrapper import WrapperSpanExporter
from logfire._internal.integrations.asgi import TweakAsgiTracerProvider
from opentelemetry import trace

# from logfire.integrations.fastapi import LogfireSpanProcessor
# pylint: disable=unused-import
from opentelemetry.context import Context
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.environment_variables import (
    OTEL_TRACES_SAMPLER,
    OTEL_TRACES_SAMPLER_ARG,
)
from opentelemetry.sdk.trace import ReadableSpan, SpanProcessor, TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor, SimpleSpanProcessor, SpanExporter
from opentelemetry.sdk.trace.sampling import (
    ALWAYS_OFF,
    ALWAYS_ON,
    ParentBased,
    Sampler,
    TraceIdRatioBased,
)
from opentelemetry.trace import Link, Span, SpanKind, get_current_span, get_tracer_provider
from opentelemetry.trace.span import TraceState
from opentelemetry.util.types import Attributes

from src.config import settings
from src.dependencies.factory import app

# Set to track routes that should NOT be logged
no_log_routes: set[str] = set()
force_log_routes: set[str] = set()


# Decorator to mark routes that should NOT be logged
def no_log(func: Callable):
    """Decorator to exclude a route from logging"""
    no_log_routes.add(func.__name__)

    @wraps(func)
    async def wrapper(*args, **kwargs):
        return await func(*args, **kwargs)

    return wrapper


# Decorator to force logging of specific routes
def force_log(func: Callable):
    """Decorator to force logging of a route regardless of method"""
    force_log_routes.add(func.__name__)

    @wraps(func)
    async def wrapper(*args, **kwargs):
        return await func(*args, **kwargs)

    return wrapper


class CustomFilterSpanProcessor(SpanProcessor):
    """
    Custom span processor that filters spans based on HTTP method and status code
    before they are exported to Logfire.
    """

    def __init__(self, next_processor: SpanProcessor):
        self.next_processor = next_processor

    def on_start(self, span: ReadableSpan, parent_context=None):
        """Called when span starts - pass through to next processor"""
        self.next_processor.on_start(span, parent_context)

    def on_end(self, span: ReadableSpan):
        """
        Called when span ends - decide whether to forward to next processor
        based on our filtering rules
        """
        # Get span attributes
        attrs = span.attributes or {}

        # Extract HTTP information
        http_method = attrs.get('http.request.method') or attrs.get('http.method')
        http_status = attrs.get('http.response.status_code') or attrs.get('http.status_code')
        http_route = attrs.get('http.route')

        # If it's not an HTTP span, always log it
        if not http_method:
            self.next_processor.on_end(span)
            return

        # Check if route is in no_log list
        if http_route:
            # Extract route handler name from the route pattern
            route_name = http_route.strip('/').replace('/', '_').replace('{', '').replace('}', '')
            if any(no_log in http_route or no_log in route_name for no_log in no_log_routes):
                # Drop this span - don't forward to next processor
                return

            # Check if route is force logged
            if any(force in http_route or force in route_name for force in force_log_routes):
                # Always log force-logged routes
                self.next_processor.on_end(span)
                return

        # Apply filtering rules based on HTTP method and status
        should_log = False

        if http_method == 'GET':
            # Only log unsuccessful GET requests
            should_log = http_status and http_status >= 400
        elif http_method in ['POST', 'PUT', 'PATCH', 'DELETE']:
            # Always log these methods
            should_log = True
        else:
            # Log other methods by default
            should_log = True

        if should_log:
            self.next_processor.on_end(span)
        # Otherwise, drop the span (don't call next_processor.on_end)

        print(should_log)
        print(f'http_method: {http_method}, http_status: {http_status}, http_route: {http_route}')

    def shutdown(self):
        """Called on shutdown"""
        self.next_processor.shutdown()

    def force_flush(self, timeout_millis: int = 30000):
        """Called to force flush"""
        return self.next_processor.force_flush(timeout_millis)


def configure_logging():
    """Configure Logfire with selective span filtering"""

    exporter = OTLPSpanExporter()
    processor = CustomFilterSpanProcessor(SimpleSpanProcessor(exporter))

    # First, configure Logfire normally
    logfire.configure(
        token=settings.LOGFIRE_TOKEN,
        send_to_logfire='if-token-present',
        additional_span_processors=[processor],
    )

    logfire.instrument_fastapi(
        app,
        capture_headers=True,
    )

Is there better way than this?