Real-Time Metrics Aggregation Engine

A high-performance, fault-tolerant metrics aggregation engine built with Elixir/OTP patterns. Designed to handle real-time metrics collection with multiple time windows, tag-based grouping, and concurrent processing.

• Real-time Aggregation: Collect and aggregate metrics across 1-minute, 5-minute, and 15-minute rolling windows
• Tag-based Grouping: Group metrics by arbitrary tag combinations for detailed analysis
• High Concurrency: Handle thousands of metrics per second with isolated worker processes
• Fault Tolerance: OTP supervision trees ensure system resilience against process failures
• Memory Efficient: Automatic cleanup of expired data with bounded memory usage
• ETS Caching: High-performance storage layer for fast data access

1. Clone the repository:

git clone <repository-url>
cd matrix_aggregation_engine

2. Install dependencies:

mix deps.get

3. Start the application:

iex -S mix

# Record a single metric
MatrixAggregationEngine.record_metric(%{
  metric_name: "api.response_time",
  value: 150.5,
  timestamp: DateTime.utc_now(),
  tags: %{"service" => "web", "environment" => "prod"}
})

# Query aggregations for all tags
aggregations = MatrixAggregationEngine.get_aggregations("api.response_time", :one_minute)
# => %{"service:web,environment:prod" => %{count: 10, sum: 1505.0, avg: 150.5, min: 100.0, max: 200.0}}

# Query with tag filters
filtered = MatrixAggregationEngine.get_aggregations("api.response_time", :five_minute, %{"service" => "web"})

# List supported time windows
MatrixAggregationEngine.list_supported_windows()
# => [:one_minute, :five_minute, :fifteen_minute]

# Get system statistics
MatrixAggregationEngine.get_worker_stats()
# => %{total_workers: 15, workers_by_window: %{one_minute: 5, five_minute: 5, fifteen_minute: 5}, active_metrics: 5}

Generate sample data for testing and exploration:

# Generate comprehensive sample dataset
MatrixAggregationEngine.SampleData.generate_all_sample_data()

# Generate specific metric types
MatrixAggregationEngine.SampleData.generate_api_metrics(100)
MatrixAggregationEngine.SampleData.generate_database_metrics(50)
MatrixAggregationEngine.SampleData.generate_error_metrics(25)

# Simulate continuous load
MatrixAggregationEngine.SampleData.simulate_continuous_load(30)

• MetricsEngine: Main interface for recording and querying metrics
• AggregationWorker: GenServer processes handling specific metric/time-window combinations
• ProcessRegistry: ETS-based registry for efficient worker process routing
• MetricsSupervisor: Dynamic supervisor managing worker processes
• Storage: ETS-based caching layer for aggregated results

1. Metric Ingestion: MetricsEngine.record_metric/1 validates and routes metrics
2. Worker Routing: ProcessRegistry finds or creates appropriate worker processes
3. Aggregation: AggregationWorker processes update in-memory aggregations
4. Querying: MetricsEngine.get_aggregations/3 retrieves current aggregations
5. Cleanup: Automatic expiration of old data maintains bounded memory usage

• 1 minute: Most recent 1-minute of data
• 5 minutes: Most recent 5-minute window
• 15 minutes: Most recent 15-minute window

Data is organized into time buckets and automatically expires when outside the current window.

The system is designed for high throughput and low latency:

• Throughput: >1000 metrics/second per node
• Write Latency: <1ms for metric ingestion
• Read Latency: <10ms for aggregation queries
• Memory: ~50KB overhead per unique metric/window combination

Run the comprehensive test suite:

# All tests
mix test

# Exclude performance tests for faster runs
mix test --exclude performance

# Run only performance tests
mix test --only performance

# Run with coverage
mix test --cover

• Unit Tests: Core data structures and business logic
• Integration Tests: End-to-end workflows and process interactions
• Performance Tests: High-throughput scenarios and memory usage
• Concurrent Tests: Race conditions and process coordination

Main module providing the public API.

Record a new metric data point.

Parameters:

• metric_data (map): Metric data with required fields:
  • metric_name (string): Name of the metric
  • value (number): Numeric value
  • timestamp (DateTime): When the metric occurred
  • tags (map): Key-value tags for grouping

Returns: :ok or {:error, reason}

Retrieve current aggregations for a metric.

Parameters:

• metric_name (string): Name of the metric to query
• time_window (atom): Time window (:one_minute, :five_minute, :fifteen_minute)
• tag_filters (map, optional): Filter results by specific tag values

Returns: Map of tag_key => aggregation_data

Get system statistics including worker counts and memory usage.

Returns: Map with worker and system statistics

• CLEANUP_INTERVAL: Cleanup interval in milliseconds (default: 30000)
• MAX_WORKERS: Maximum number of worker processes (default: unlimited)

config :matrix_aggregation_engine,
  cleanup_interval: 30_000,
  supported_windows: [:one_minute, :five_minute, :fifteen_minute]

• Erlang/OTP 24+
• Elixir 1.15+
• 2GB+ RAM recommended for high-throughput scenarios

Monitor key metrics in production:

• Process count and memory usage
• Metric ingestion rate and latency
• Aggregation query performance
• Error rates and process restarts

The system scales both vertically and horizontally:

• Vertical: More CPU cores = more concurrent workers
• Horizontal: Distribute workers across multiple nodes (future enhancement)

1. High Memory Usage: Check for too many unique metric/tag combinations
2. Slow Queries: Verify time windows and tag filters are appropriate
3. Process Crashes: Monitor supervision tree for systematic failures

# Check worker statistics
MatrixAggregationEngine.get_worker_stats()

# List all workers
MatrixAggregationEngine.ProcessRegistry.list_all_workers()

# Observe process supervision tree
Observer.start()

# Start your system
{:ok, _pid} = MatrixAggregationEngine.start_link()

# Record some metrics
MatrixAggregationEngine.record_metric(%{
  metric_name: "api.response_time",
  value: 150.5,
  timestamp: DateTime.utc_now(),
  tags: %{"service" => "web"}
})

# Get aggregations
MatrixAggregationEngine.get_aggregations("api.response_time", :one_minute)

[Add your license information here]

[Add contributing guidelines here]