Integrate https://pola.rs/

[jeremyandrews]

Would there be benefits integrating Goose with https://pola.rs/. For example, for logging of requests, and then allowing more complex analysis of the results. Rather than maintaining counters, averages, mins and maxes, the data could be generated as needed from logs or databases.

[j4qfrost]

I think integrating Apache Arrow is a more flexible option. There seems to be more streaming utilities and polars uses it under the hood. I'm reading through. Hopefully I can get a PR in soon.

[jeremyandrews]

Polars Integration Plan: Stream-Only Data Export

Executive Summary

Feature-flagged integration of Polars for raw metrics data export, enabling external analysis while preserving existing real-time metrics system. Stream-only approach minimizes complexity and risk while maximizing analytical flexibility.

Technical Rationale: Polars vs Apache Arrow

Polars Selection Justification

• Higher-level API: DataFrame operations vs low-level memory management
• Rust-native: Better integration with Goose codebase, no FFI overhead
• Built-in I/O: Native Parquet/CSV export without additional dependencies
• Lazy evaluation: Memory-efficient processing of large datasets
• Performance: SIMD optimizations and parallel processing built-in
• Ecosystem: Can export to Arrow format when needed, maintains interoperability

Apache Arrow Disadvantages

• Lower-level complexity: Requires building DataFrame abstractions from scratch
• Additional dependencies: Need separate libraries for file I/O and operations
• Implementation overhead: More code to maintain for basic functionality
• Learning curve: More complex API for basic use cases

Implementation Plan

Phase 1: Core Infrastructure

Feature Flag Configuration

# Cargo.toml
[features]
polars = ["dep:polars", "dep:polars-lazy"]

[dependencies]
polars = { version = "0.33", features = ["lazy", "parquet", "csv-file"], optional = true }

Data Schema

#[cfg(feature = "polars")]
#[derive(Clone)]
pub struct PolarsRequestRecord {
    pub timestamp: i64,           // Unix timestamp (ms)
    pub elapsed: u64,             // Test elapsed time (ms)
    pub scenario_index: u32,      // Scenario identifier
    pub scenario_name: String,    // Human-readable scenario name
    pub transaction_index: String, // Transaction identifier
    pub transaction_name: String, // Human-readable transaction name
    pub method: String,           // HTTP method
    pub endpoint: String,         // Request path/name
    pub response_time: u32,       // Response time (ms)
    pub status_code: u16,         // HTTP status code
    pub success: bool,            // Request success/failure
    pub user_id: u32,             // User thread identifier
    pub final_url: String,        // Final URL after redirects
    pub redirected: bool,         // Redirect occurred
    pub error: String,            // Error message (empty if none)
}

Memory Management Strategy

#[cfg(feature = "polars")]
pub struct PolarsMetrics {
    buffer: Vec<PolarsRequestRecord>,
    memory_limit: usize,
    export_path: PathBuf,
    export_format: ExportFormat,
    records_written: u64,
}

pub enum ExportFormat {
    Parquet,
    Csv,
}

Phase 2: Integration Points

Configuration Extension

pub struct GooseConfiguration {
    // Existing fields...
    #[cfg(feature = "polars")]
    pub polars_export: Option<PathBuf>,
    #[cfg(feature = "polars")]
    pub polars_format: ExportFormat,
    #[cfg(feature = "polars")]
    pub polars_memory_limit: usize, // Default: 100MB
    #[cfg(feature = "polars")]
    pub polars_buffer_size: usize,  // Default: 10,000 records
}

CLI Options

--polars-export=PATH          # Export file path
--polars-format=FORMAT        # parquet|csv (default: parquet)
--polars-memory-limit=SIZE    # Memory limit before flush (default: 100MB)
--polars-buffer-size=COUNT    # Buffer size in records (default: 10000)

Metrics Collection Integration

impl GooseAttack {
    #[cfg(feature = "polars")]
    async fn record_polars_metric(&mut self, metric: &GooseRequestMetric) -> Result<(), GooseError> {
        let record = PolarsRequestRecord::from(metric);
        self.polars_metrics.add_record(record).await?;
        Ok(())
    }
    
    #[cfg(feature = "polars")]
    async fn finalize_polars_export(&mut self) -> Result<(), GooseError> {
        self.polars_metrics.flush_and_export().await
    }
}

Phase 3: Export Implementation

Streaming Export Strategy

• Buffered writes: Accumulate records in memory buffer
• Periodic flushes: Write to disk when buffer full or memory limit reached
• Append mode: Support incremental writes for long-running tests
• Error handling: Graceful degradation if export fails

File Format Support

• Parquet (default): Columnar, compressed, efficient for analytics
• CSV: Human-readable, universal compatibility
• Future: JSON, Arrow IPC format

Phase 4: Memory Management

Buffer Management

impl PolarsMetrics {
    async fn add_record(&mut self, record: PolarsRequestRecord) -> Result<(), GooseError> {
        self.buffer.push(record);
        
        if self.buffer.len() >= self.buffer_size {
            self.flush_buffer().await?;
        }
        
        if self.estimated_memory_usage() > self.memory_limit {
            self.flush_buffer().await?;
        }
        
        Ok(())
    }
    
    async fn flush_buffer(&mut self) -> Result<(), GooseError> {
        if self.buffer.is_empty() {
            return Ok(());
        }
        
        let df = DataFrame::new(vec![
            Series::new("timestamp", self.buffer.iter().map(|r| r.timestamp).collect::<Vec<_>>()),
            Series::new("response_time", self.buffer.iter().map(|r| r.response_time).collect::<Vec<_>>()),
            // ... other columns
        ])?;
        
        match self.export_format {
            ExportFormat::Parquet => self.append_parquet(df).await?,
            ExportFormat::Csv => self.append_csv(df).await?,
        }
        
        self.records_written += self.buffer.len() as u64;
        self.buffer.clear();
        Ok(())
    }
}

Performance Considerations

Memory Usage

• Current system: ~1-2KB per endpoint (aggregated)
• Polars system: ~40 bytes per request (raw data)
• Impact: 20-40x memory increase for raw data storage
• Mitigation: Configurable memory limits, periodic flushing

CPU Overhead

• Data conversion: Minimal overhead converting GooseRequestMetric to PolarsRequestRecord
• Serialization: Parquet compression CPU cost during flushes
• Impact: <5% CPU overhead during normal operation, higher during flushes
• Mitigation: Async flushes, configurable buffer sizes

I/O Impact

• Disk writes: Periodic large writes vs continuous small writes
• Network impact: None (local file system only)
• Impact: Minimal during test, burst I/O during flushes
• Mitigation: Async I/O, SSD-optimized write patterns

Risk Assessment

Low Risk (Feature-flagged)

• Backwards compatibility: No impact when feature disabled
• Performance: Minimal overhead when enabled, configurable limits
• Dependencies: Optional dependency, doesn't affect core build

Medium Risk

• Memory usage: Could impact long-running tests
• Disk space: Large files for high-volume tests
• Complexity: Additional code paths to maintain

Mitigation Strategies

• Memory limits: Hard limits with graceful degradation
• Disk monitoring: Warn users about disk space requirements
• Documentation: Clear guidance on when to use feature
• Testing: Comprehensive testing under various load conditions

External Analysis Workflow

Data Export

# Run test with Polars export
goose --polars-export=results.parquet --host=http://example.com run

# Verify export
ls -lh results.parquet

External Analysis Examples

# Python analysis
import polars as pl

df = pl.read_parquet("results.parquet")

# Basic analysis
print(df.describe())
print(df.group_by("endpoint").agg([
    pl.col("response_time").mean().alias("avg_response_time"),
    pl.col("response_time").quantile(0.95).alias("p95_response_time"),
    pl.col("success").mean().alias("success_rate")
]))

# Time series analysis
df.with_columns([
    (pl.col("timestamp") // 30000).alias("time_bucket")  # 30-second buckets
]).group_by(["time_bucket", "endpoint"]).agg([
    pl.col("response_time").mean().alias("avg_response_time"),
    pl.col("response_time").count().alias("request_count")
])

// Rust analysis
use polars::prelude::*;

fn analyze_results() -> PolarsResult<()> {
    let df = LazyFrame::scan_parquet("results.parquet", ScanArgsParquet::default())?;
    
    let summary = df
        .group_by([col("endpoint")])
        .agg([
            col("response_time").mean().alias("avg_response_time"),
            col("response_time").quantile(lit(0.95), QuantileInterpolOptions::default()).alias("p95"),
            col("success").mean().alias("success_rate"),
        ])
        .collect()?;
    
    println!("{}", summary);
    Ok(())
}

Implementation

A. Core Infrastructure

• Feature flag setup
• Data schema definition
• Basic PolarsMetrics struct
• Configuration integration

B. Export Implementation

• Buffer management
• Parquet export functionality
• Memory limit enforcement
• Error handling

C. Integration & Testing

• Integration with existing metrics system
• Performance testing
• Memory usage validation
• Documentation

D: Polish & Documentation

• CLI help text
• Usage examples
• Performance tuning
• Code review

Success Criteria

Functional Requirements

• Export raw request data to Parquet/CSV
• Configurable memory limits and buffer sizes
• No impact on existing metrics when disabled
• Graceful handling of export failures

Performance Requirements

• <5% CPU overhead during normal operation
• Memory usage within configured limits
• No impact on request processing latency

Quality Requirements

• Comprehensive test coverage
• Clear documentation and examples
• Backwards compatibility maintained
• Clean feature flag implementation

Future Considerations

Potential Enhancements

• Streaming formats: Arrow IPC for real-time analysis
• Compression options: Different compression algorithms
• Partitioning: Time-based file partitioning for large datasets
• Schema evolution: Versioned schema for backwards compatibility

Integration Opportunities

• Built-in analysis: Common analysis functions within Goose
• Real-time streaming: WebSocket or gRPC streaming for live analysis
• Database export: Direct export to analytical databases
• Visualization: Integration with plotting libraries

This plan provides a solid foundation for Polars integration while maintaining Goose's core strengths and minimizing risk through feature flags and careful design.