Struct MetricsObserver 

pub struct MetricsObserver { /* private fields */ }

Metrics observer that collects and reports pipeline processing metrics to Prometheus.

MetricsObserver implements the ProcessingObserver trait to receive processing events from the pipeline and update Prometheus metrics accordingly. It provides real-time metrics collection with minimal performance overhead.

Features

Real-Time Metrics Collection

• Event-Driven: Responds to processing events in real-time
• Atomic Updates: Thread-safe metric updates using atomic operations
• Low Latency: Minimal delay between events and metric updates
• Continuous Monitoring: Provides continuous visibility into processing

Performance Tracking

• Throughput Calculation: Real-time throughput calculation in MB/s
• Duration Tracking: Tracks processing duration for chunks and overall processing
• Progress Monitoring: Monitors processing progress and completion rates
• Efficiency Metrics: Calculates processing efficiency and performance indicators

Prometheus Integration

• Metric Updates: Updates Prometheus metrics through MetricsService
• Counter Increments: Increments counters for processed chunks and bytes
• Histogram Recording: Records duration histograms for performance analysis
• Gauge Updates: Updates gauge metrics for real-time values

Usage Examples

Basic Observer Creation

Integration with Pipeline Processing

Metric Tracking

The observer tracks several key metrics:

Byte Tracking

• Total Bytes: Total bytes to be processed (set at start)
• Processed Bytes: Cumulative bytes processed (updated continuously)
• Current Chunk Size: Size of currently processing chunk

Performance Tracking

• Start Time: Processing start timestamp for throughput calculation
• Throughput: Real-time throughput calculation in MB/s
• Duration: Processing duration for performance analysis

Thread Safety

The observer is designed for concurrent use:

• Atomic Counters: All counters use atomic operations
• Immutable Service: Metrics service is shared through Arc
• No Locks: Lock-free design for minimal contention
• Safe Updates: Thread-safe metric updates

Performance Characteristics

• Low Overhead: Minimal impact on processing performance (~1-2% overhead)
• Efficient Updates: Optimized atomic operations
• Memory Efficient: Small memory footprint (~100 bytes)
• Scalable: Performance scales with number of processing threads

Implementations

impl MetricsObserver

pub fn new(metrics_service: Arc<MetricsService>) -> Self

Creates a new metrics observer with the provided metrics // service.

Initializes the observer with the given metrics service and sets up internal tracking state for processing metrics. The observer is ready to receive processing events immediately after creation.

Arguments

• metrics_service - Arc-wrapped metrics service for reporting metrics

Returns

A new MetricsObserver instance ready to collect processing metrics.

Examples

Initialization

The observer initializes with:

• Current timestamp as start time
• All byte counters set to zero
• Ready to receive processing events

Trait Implementations

impl ProcessingObserver for MetricsObserver

fn on_processing_started<'life0, 'async_trait>( &'life0 self, total_bytes: u64, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Called when processing starts

fn on_chunk_started<'life0, 'async_trait>( &'life0 self, chunk_id: u64, size: usize, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Called when a chunk starts processing

fn on_chunk_completed<'life0, 'async_trait>( &'life0 self, chunk_id: u64, duration: Duration, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Called when a chunk completes processing

fn on_progress_update<'life0, 'async_trait>( &'life0 self, bytes_processed: u64, _total_bytes: u64, throughput_mbps: f64, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Called periodically with progress updates

fn on_processing_completed<'life0, 'life1, 'async_trait>( &'life0 self, total_duration: Duration, final_metrics: Option<&'life1 ProcessingMetrics>, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Called when processing completes