Schema Registry Analytics Engine

Production-ready analytics engine for the LLM Schema Registry, providing comprehensive tracking, aggregation, and reporting capabilities for schema usage and performance metrics.

Features

• Real-time Event Streaming: In-memory event bus with broadcast channels for distributing events to multiple subscribers
• Time-Series Aggregation: Automatic aggregation at 1-minute, 5-minute, 1-hour, and 1-day intervals
• Schema Usage Tracking: Track reads, writes, validations, compatibility checks, and more
• Performance Analytics: Latency percentiles (p50, p95, p99), success rates, and operation breakdowns
• Popular Schema Identification: Top schemas by operation type, trending schemas, zombie detection
• Health Scorecards: Automated schema health scoring based on success rate, performance, and activity
• Anomaly Detection: Threshold-based anomaly detection for error rates and latency spikes
• Comprehensive Reporting: Daily, weekly, and monthly usage reports with JSON export

Quick Start

use schema_registry_analytics::{AnalyticsEngine, SchemaUsageEvent, Operation};
use uuid::Uuid;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Create and start the analytics engine
    let engine = AnalyticsEngine::new();
    engine.start().await?;

    // Record schema usage events
    let event = SchemaUsageEvent::new(
        Uuid::new_v4(),
        Operation::Read,
        "my-service".to_string(),
        "us-west-1".to_string(),
        125, // latency in ms
        true, // success
    );
    engine.record_event(event)?;

    // Query analytics
    let top_schemas = engine.get_top_schemas(Some(Operation::Read), 10);
    println!("Top 10 schemas: {:?}", top_schemas);

    // Get schema health
    let schema_id = Uuid::new_v4();
    if let Some(health) = engine.get_schema_health(&schema_id.into()) {
        println!("Health score: {}", health.overall_score);
    }

    // Graceful shutdown
    engine.shutdown().await?;
    Ok(())
}

Architecture

Components

1. Event Bus (event_bus.rs): Real-time event distribution using tokio broadcast channels
2. Aggregator (aggregator.rs): Time-series data aggregation with configurable windows
3. Storage (storage.rs): In-memory storage with retention policies (prepared for TimescaleDB)
4. Query Executor (query.rs): High-level query interface with filtering and pagination
5. Report Generator (reports.rs): Automated reporting and anomaly detection
6. Analytics Engine (engine.rs): Main orchestrator coordinating all components

Data Flow

Event → Event Bus → [Event Consumer] → Storage
                                    → Aggregator
                                         ↓
                                    Query Executor
                                         ↓
                                    Report Generator

Usage Examples

Recording Events

// Simple event recording
let event = SchemaUsageEvent::new(
    schema_id,
    Operation::Validate,
    "my-service".to_string(),
    "us-east-1".to_string(),
    50,
    true,
);
engine.record_event(event)?;

// Failed event with error message
let failed_event = SchemaUsageEvent::failed(
    schema_id,
    Operation::Validate,
    "my-service".to_string(),
    "us-east-1".to_string(),
    100,
    "Schema validation failed".to_string(),
);
engine.record_event(failed_event)?;

Querying Analytics

use chrono::{Utc, Duration};

// Get usage stats for the last 24 hours
let now = Utc::now();
let start = now - Duration::hours(24);

let stats = engine.get_usage_stats(
    TimePeriod::Hour1,
    start,
    now,
    None, // Global stats
)?;

// Query with filters using the query builder
let query = QueryBuilder::last_days(7)
    .operation(Operation::Read)
    .region("us-west-1")
    .success_only()
    .limit(100);

let results = query.execute(engine.query_executor().as_ref())?;

Generating Reports

let report_gen = engine.report_generator();

// Daily summary
let daily = report_gen.generate_daily_summary(Utc::now())?;
println!("Total operations: {}", daily.total_operations);
println!("Success rate: {:.2}%", daily.success_rate * 100.0);

// Weekly trends
let weekly = report_gen.generate_weekly_report(Utc::now())?;
println!("Trending up: {:?}", weekly.trending_up);

// Anomaly detection
let anomalies = report_gen.detect_anomalies(24)?;
for anomaly in anomalies {
    println!("Anomaly detected: {}", anomaly.description);
}

// Export to JSON
let json = report_gen.export_to_json(&daily)?;

Health Monitoring

// Get schema health scorecard
if let Some(health) = engine.get_schema_health(&schema_id.into()) {
    println!("Overall health: {}/100", health.overall_score);
    println!("Success rate: {}/100", health.success_rate_score);
    println!("Performance: {}/100", health.performance_score);
    println!("Activity: {}/100", health.activity_score);
    println!("Is zombie: {}", health.is_zombie);

    for recommendation in health.recommendations {
        println!("Recommendation: {}", recommendation);
    }
}

Configuration

Customize the analytics engine with AnalyticsConfig:

use schema_registry_analytics::{AnalyticsEngine, AnalyticsConfig, StorageConfig, TimePeriod};

let config = AnalyticsConfig {
    storage_config: StorageConfig {
        detailed_retention_days: 90,
        max_events_per_day: 1_000_000,
        store_raw_events: true,
    },
    event_bus_capacity: 50_000,
    auto_cleanup: true,
    cleanup_interval_seconds: 1800, // 30 minutes
    aggregation_periods: vec![
        TimePeriod::Minute1,
        TimePeriod::Minute5,
        TimePeriod::Hour1,
        TimePeriod::Day1,
    ],
};

let engine = AnalyticsEngine::with_config(config);

Performance Characteristics

• Event Processing: <1ms latency
• Query Response: <100ms for typical queries
• Memory Efficient: Automatic retention-based cleanup
• Thread-Safe: Concurrent access via Arc + RwLock
• Scalable: Designed for high-throughput event streams

Storage Strategy

In-Memory (Current)

• Fast access and aggregation
• Configurable retention periods
• Automatic cleanup
• Suitable for up to 10M events/day

Future: TimescaleDB

The storage layer is designed to be easily swapped with TimescaleDB for:

• Persistent storage
• Larger data volumes
• Advanced time-series queries
• Long-term historical analysis

Integration with Observability

The analytics engine integrates with the existing observability metrics:

// Record metrics alongside analytics
metrics::counter!("schema.operations.total").increment(1);
engine.record_event(event)?;

Testing

Run the comprehensive test suite:

cargo test -p schema-registry-analytics

All 43+ unit tests and 10+ integration tests verify:

• Event bus functionality
• Aggregation accuracy
• Storage and retrieval
• Query filtering
• Report generation
• Concurrent access
• Memory management

Future Enhancements

1. Kafka Integration: Replace in-memory event bus with Kafka for distributed processing
2. TimescaleDB Storage: Persistent time-series database for historical data
3. ML-based Anomaly Detection: Advanced anomaly detection using machine learning
4. Real-time Dashboards: WebSocket-based real-time analytics dashboards
5. Custom Metrics: User-defined custom metrics and aggregations
6. Multi-region Aggregation: Cross-region analytics and comparison
7. Prometheus Export: Direct metrics export to Prometheus

License

Apache-2.0