I am using Sentry and sentry_sdk to monitor errors and traces in my Python application. I want to configure the error and trace rates for different routes in my FastAPI API. To do this, I want to write two decorators called sentry_error_rate and sentry_trace_rate that will allow me to set the sample rates for errors and traces, respectively.
The sentry_error_rate decorator should take a single argument errors_sample_rate (a float between 0 and 1) and apply it to a specific route.
The sentry_trace_rate decorator should take a single argument traces_sample_rate (also a float between 0 and 1) and apply it to a specific route.
def sentry_trace_rate(traces_sample_rate: float = 0.0) -> callable:
""" Decorator to set the traces_sample_rate for a specific route.
This is useful for routes that are called very frequently, but we
want to sample them to reduce the amount of data we send to Sentry.
Args:
traces_sample_rate (float): The sample rate to use for this route.
"""
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
# Do something here ?
return await func(*args, **kwargs)
return wrapper
return decorator
def sentry_error_rate(errors_sample_rate: float = 0.0) -> callable:
""" Decorator to set the errors_sample_rate for a specific route.
This is useful for routes that are called very frequently, but we
want to sample them to reduce the amount of data we send to Sentry.
Args:
errors_sample_rate (float): The sample rate to use for this route.
"""
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
# Do something here ?
return await func(*args, **kwargs)
return wrapper
return decorator
Does someone have an idea if this is possible and how it could be done ?
I finally managed to do it using a registry mechanism. Each route with a decorator are registred in a dictionary with their trace/error rate. I then used a trace_sampler/before_send function as indicated here:
Here's my sentry_wrapper.py:
import asyncio
import random
from functools import wraps
from typing import Callable, Union
from fastapi import APIRouter
_route_traces_entrypoints = {}
_route_errors_entrypoints = {}
_fn_traces_entrypoints = {}
_fn_errors_entrypoints = {}
_fn_to_route_entrypoints = {}
def sentry_trace_rate(trace_sample_rate: float = 0.0) -> Callable:
"""Decorator to set the sentry trace rate for a specific endpoint.
This is useful for endpoints that are called very frequently,
and we don't want to report all traces.
Args:
trace_sample_rate (float): The rate to sample traces. 0.0 to disable traces.
"""
def decorator(fn: Callable) -> Callable:
# Assert there is not twice function with the same nam
if fn.__name__ in _fn_traces_entrypoints:
raise ValueError(f"Two function have the same name: {fn.__name__} | {fn.__file__}")
# Add fn entrypoint
_fn_traces_entrypoints[fn.__name__] = trace_sample_rate
# Check for coroutines and return the right wrapper
if asyncio.iscoroutinefunction(fn):
@wraps(fn)
async def wrapper(*args, **kwargs) -> Callable:
return await fn(*args, **kwargs)
return wrapper
else:
@wraps(fn)
def wrapper(*args, **kwargs) -> Callable:
return fn(*args, **kwargs)
return wrapper
return decorator
def sentry_error_rate(error_sample_rate: float = 0.0) -> Callable:
"""Decorator to set the sentry error rate for a specific endpoint.
This is useful for endpoints that are called very frequently,
and we don't want to report all errors.
Args:
error_sample_rate (float): The rate to sample errors. 0.0 to disable errors.
"""
def decorator(fn: Callable) -> Callable:
# Assert there is not twice function with the same nam
if fn.__name__ in _fn_errors_entrypoints:
raise ValueError(f"Two function have the same name: {fn.__name__} | {fn.__file__}")
# Add fn entrypoint
_fn_errors_entrypoints[fn.__name__] = error_sample_rate
# Check for coroutines and return the right wrapper
if asyncio.iscoroutinefunction(fn):
@wraps(fn)
async def wrapper(*args, **kwargs) -> Callable:
return await fn(*args, **kwargs)
return wrapper
else:
@wraps(fn)
def wrapper(*args, **kwargs) -> Callable:
return fn(*args, **kwargs)
return wrapper
return decorator
def register_traces_disabler(router: APIRouter) -> None:
"""Register all the entrypoints for the traces disabler
Args:
router (APIRouter): The router to register
"""
for route in router.routes:
if route.name in _fn_traces_entrypoints:
_route_traces_entrypoints[route.path] = _fn_traces_entrypoints[route.name]
def register_errors_disabler(router: APIRouter) -> None:
"""Register all the entrypoints for the errors disabler
Args:
router (APIRouter): The router to register
"""
for route in router.routes:
if route.name in _fn_errors_entrypoints:
_route_errors_entrypoints[route.path] = _fn_errors_entrypoints[route.name]
class TracesSampler:
"""Class to sample traces for sentry
Args:
default_traces_sample_rate (float, optional): The default sample rate for traces.
Defaults to 1.0.
"""
def __init__(self, default_traces_sample_rate: float = 1.0) -> None:
self.default_traces_sample_rate = default_traces_sample_rate
def __call__(self, sampling_context) -> float:
return _route_traces_entrypoints.get(sampling_context["asgi_scope"]["path"], self.default_traces_sample_rate)
class BeforeSend:
"""Class to sample event before sending them to sentry
Args:
default_errors_sample_rate (float, optional): The default sample rate for errors.
Defaults to 1.0.
"""
def __init__(self, default_errors_sample_rate: float = 1.0) -> None:
self.default_errors_sample_rate = default_errors_sample_rate
def __call__(self, event: dict, hint: dict) -> Union[dict, None]:
# Get the sample rate for this route, or use the default if it's not defined
sample_rate = _route_errors_entrypoints.get(event["transaction"], self.default_errors_sample_rate)
# Generate a random number between 0 and 1, and discard the event if it's greater than the sample rate
if random.random() > sample_rate:
return None
# Return the event if it should be captured
return event
I have then ro register some routes:
@router.get("/route")
@sentry_wrapper.sentry_trace_rate(trace_sample_rate=0.5) # limit traces to 50%
@sentry_wrapper.sentry_error_rate(error_sample_rate=0.25) # limit error to 25%
def route_fn():
pass
And don't forget to register each route at the end of the file:
from app.services.sentry_wrapper import register_errors_disabler, register_traces_disabler
register_traces_disabler(router)
register_errors_disabler(router)