Processor

provide.foundation.profiling.processor

TODO: Add module docstring.

Classes

ProfilingProcessor

ProfilingProcessor(
    sample_rate: float = DEFAULT_PROFILING_SAMPLE_RATE,
)

Structlog processor that collects performance metrics via sampling.

This processor integrates into Foundation's existing structlog pipeline to collect metrics about message processing performance, emoji overhead, and throughput with configurable sampling to minimize performance impact.

Example

processor = ProfilingProcessor(sample_rate=0.01) # 1% sampling

Add to structlog processor chain

processors.append(processor)

Later, get metrics

metrics = processor.get_metrics() print(f"Processing {metrics.messages_per_second:.0f} msg/sec")

Initialize profiling processor with sampling configuration.

Parameters:

Name	Type	Description	Default
sample_rate	float	Fraction of messages to sample (0.0 to 1.0) 0.01 = 1% sampling for minimal overhead	DEFAULT_PROFILING_SAMPLE_RATE

Source code in provide/foundation/profiling/processor.py

def __init__(self, sample_rate: float = DEFAULT_PROFILING_SAMPLE_RATE) -> None:
    """Initialize profiling processor with sampling configuration.

    Args:
        sample_rate: Fraction of messages to sample (0.0 to 1.0)
                    0.01 = 1% sampling for minimal overhead

    """
    if not 0.0 <= sample_rate <= 1.0:
        raise SamplingError("Sample rate must be between 0.0 and 1.0", sample_rate=sample_rate)

    self.sample_rate = sample_rate
    self.metrics = ProfileMetrics()
    self._enabled = True

Functions

__call__

__call__(
    logger: Any, method_name: str, event_dict: EventDict
) -> structlog.types.EventDict

Process log event and optionally collect metrics.

This is the main entry point called by structlog for each log message. Uses sampling to minimize performance overhead.

Parameters:

Name	Type	Description	Default
logger	Any	The logger instance (unused)	required
method_name	str	The logging method name (unused)	required
event_dict	EventDict	The event dictionary to process	required

Returns:

Type	Description
EventDict	The event_dict unchanged (pass-through processor)

Source code in provide/foundation/profiling/processor.py

def __call__(
    self,
    logger: Any,
    method_name: str,
    event_dict: structlog.types.EventDict,
) -> structlog.types.EventDict:
    """Process log event and optionally collect metrics.

    This is the main entry point called by structlog for each log message.
    Uses sampling to minimize performance overhead.

    Args:
        logger: The logger instance (unused)
        method_name: The logging method name (unused)
        event_dict: The event dictionary to process

    Returns:
        The event_dict unchanged (pass-through processor)

    """
    # Always return event_dict unchanged - we're just observing
    if not self._enabled:
        return event_dict

    # Use sampling to reduce overhead
    if random.random() > self.sample_rate:
        return event_dict

    # Measure processing time for this event
    start_time = time.perf_counter_ns()

    try:
        # Analyze event characteristics
        has_emoji = self._detect_emoji_processing(event_dict)
        field_count = len(event_dict)

        # Record metrics (very fast operation)
        processing_time = time.perf_counter_ns() - start_time
        self.metrics.record_message(
            duration_ns=processing_time,
            has_emoji=has_emoji,
            field_count=field_count,
        )

    except Exception:
        # Never let profiling break the logging pipeline
        # Silently ignore any profiling errors
        pass

    return event_dict

disable

disable() -> None

Disable metrics collection.

Source code in provide/foundation/profiling/processor.py

def disable(self) -> None:
    """Disable metrics collection."""
    self._enabled = False

enable

enable() -> None

Enable metrics collection.

Source code in provide/foundation/profiling/processor.py

def enable(self) -> None:
    """Enable metrics collection."""
    self._enabled = True

get_metrics

get_metrics() -> ProfileMetrics

Get current metrics.

Returns:

Type	Description
ProfileMetrics	Current ProfileMetrics instance

Source code in provide/foundation/profiling/processor.py

def get_metrics(self) -> ProfileMetrics:
    """Get current metrics.

    Returns:
        Current ProfileMetrics instance

    """
    return self.metrics

reset

reset() -> None

Reset collected metrics.

Source code in provide/foundation/profiling/processor.py

def reset(self) -> None:
    """Reset collected metrics."""
    self.metrics.reset()