Sampling

Types for configuring sampling. See the sampling guide.

SamplingOptions dataclass

SamplingOptions(
    head: float | Sampler = 1.0,
    tail: (
        Callable[[TailSamplingSpanInfo], float] | None
    ) = None,
)

Options for logfire.configure(sampling=...).

See the sampling guide.

head class-attribute instance-attribute

head: float | Sampler = 1.0

Head sampling options.

If it's a float, it should be a number between 0.0 and 1.0. This is the probability that an entire trace will randomly included.

Alternatively you can pass a custom OpenTelemetry Sampler.

tail class-attribute instance-attribute

tail: Callable[[TailSamplingSpanInfo], float] | None = None

An optional tail sampling callback which will be called for every span.

It should return a number between 0.0 and 1.0, the probability that the entire trace will be included. Use SamplingOptions.level_or_duration for a common use case.

Every span in a trace will be stored in memory until either the trace is included by tail sampling or it's completed and discarded, so large traces may consume a lot of memory.

level_or_duration classmethod

level_or_duration(
    *,
    head: float | Sampler = 1.0,
    level_threshold: LevelName | None = "notice",
    duration_threshold: float | None = 5.0,
    background_rate: float = 0.0
) -> Self

Returns a SamplingOptions instance that tail samples traces based on their log level and duration.

If a trace has at least one span/log that has a log level greater than or equal to level_threshold, or if the duration of the whole trace is greater than duration_threshold seconds, then the whole trace will be included. Otherwise, the probability is background_rate.

The head parameter is the same as in the SamplingOptions constructor.

Source code in logfire/sampling/_tail_sampling.py

@classmethod
def level_or_duration(
    cls,
    *,
    head: float | Sampler = 1.0,
    level_threshold: LevelName | None = 'notice',
    duration_threshold: float | None = 5.0,
    background_rate: float = 0.0,
) -> Self:
    """Returns a `SamplingOptions` instance that tail samples traces based on their log level and duration.

    If a trace has at least one span/log that has a log level greater than or equal to `level_threshold`,
    or if the duration of the whole trace is greater than `duration_threshold` seconds,
    then the whole trace will be included.
    Otherwise, the probability is `background_rate`.

    The `head` parameter is the same as in the `SamplingOptions` constructor.
    """
    head_sample_rate = head if isinstance(head, (float, int)) else 1.0

    if not (0.0 <= background_rate <= head_sample_rate <= 1.0):
        raise ValueError('Invalid sampling rates, must be 0.0 <= background_rate <= head <= 1.0')

    def get_tail_sample_rate(span_info: TailSamplingSpanInfo) -> float:
        if duration_threshold is not None and span_info.duration > duration_threshold:
            return 1.0

        if level_threshold is not None and span_info.level >= level_threshold:
            return 1.0

        return background_rate

    return cls(head=head, tail=get_tail_sample_rate)

SpanLevel dataclass

SpanLevel(number: int)

A convenience class for comparing span/log levels.

Can be compared to log level names (strings) such as 'info' or 'error' using <, >, <=, or >=, so e.g. level >= 'error' is valid.

Will raise an exception if compared to a non-string or an invalid level name.

number instance-attribute

number: int

The raw numeric value of the level. Higher values are more severe.

name property

name: LevelName | None

The human-readable name of the level, or None if the number is invalid.

from_span classmethod

from_span(span: ReadableSpan) -> SpanLevel

Create a SpanLevel from an OpenTelemetry span.

If the span has no level set, defaults to 'info'.

Source code in logfire/types.py

@classmethod
def from_span(cls, span: ReadableSpan) -> SpanLevel:
    """Create a SpanLevel from an OpenTelemetry span.

    If the span has no level set, defaults to 'info'.
    """
    attributes = span.attributes or {}
    level = attributes.get(ATTRIBUTES_LOG_LEVEL_NUM_KEY)
    if not isinstance(level, int):
        level = LEVEL_NUMBERS['info']
    return cls(level)

TailSamplingSpanInfo dataclass

TailSamplingSpanInfo(
    span: ReadableSpan,
    context: Context | None,
    event: Literal["start", "end"],
    buffer: TraceBuffer,
)

Argument passed to the SamplingOptions.tail callback.

span instance-attribute

span: ReadableSpan

Raw span object being started or ended.

context instance-attribute

context: Context | None

Second argument of SpanProcessor.on_start or None for SpanProcessor.on_end.

event instance-attribute

event: Literal['start', 'end']

'start' if the span is being started, 'end' if it's being ended.

level property

level: SpanLevel

The log level of the span.

duration property

duration: float

The time in seconds between the start of the trace and the start/end of this span.