test_stream_aggregation.cpp File Reference

#include <gtest/gtest.h>
#include <kcenon/monitoring/utils/stream_aggregator.h>
#include <kcenon/monitoring/utils/aggregation_processor.h>
#include <kcenon/monitoring/utils/metric_storage.h>
#include <chrono>
#include <thread>
#include <vector>
#include <random>
#include <cmath>

Include dependency graph for test_stream_aggregation.cpp:

Go to the source code of this file.

Classes
class	StreamAggregationTest
	Test suite for Phase 3 P2: Statistical aggregation functions. More...

Functions
	TEST_F (StreamAggregationTest, OnlineStatisticsBasic)
	TEST_F (StreamAggregationTest, OnlineStatisticsLargeDataset)
	TEST_F (StreamAggregationTest, QuantileEstimatorMedian)
	TEST_F (StreamAggregationTest, QuantileEstimatorPercentiles)
	TEST_F (StreamAggregationTest, MovingWindowBasic)
	TEST_F (StreamAggregationTest, MovingWindowExpiration)
	TEST_F (StreamAggregationTest, StreamAggregatorBasic)
	TEST_F (StreamAggregationTest, StreamAggregatorPercentiles)
	TEST_F (StreamAggregationTest, StreamAggregatorOutlierDetection)
	TEST_F (StreamAggregationTest, StreamAggregatorReset)
	TEST_F (StreamAggregationTest, AggregationProcessorBasic)
	TEST_F (StreamAggregationTest, AggregationProcessorMultipleMetrics)
	TEST_F (StreamAggregationTest, AggregationProcessorForceAggregation)
	TEST_F (StreamAggregationTest, AggregationProcessorInvalidRule)
	TEST_F (StreamAggregationTest, PearsonCorrelation)
	TEST_F (StreamAggregationTest, StandardAggregationRules)
	TEST_F (StreamAggregationTest, ConfigurationValidation)
	TEST_F (StreamAggregationTest, StreamAggregatorThreadSafety)

Function Documentation

TEST_F() [1/18]

TEST_F	(	StreamAggregationTest	,
		AggregationProcessorBasic	)

Definition at line 279 of file test_stream_aggregation.cpp.

                                                         {
    auto storage = std::make_shared<metric_storage>();
    aggregation_processor processor(storage);
    
    // Create aggregation rule
    aggregation_rule rule;
    rule.source_metric = "test_metric";
    rule.target_metric_prefix = "test_metric_stats";
    rule.aggregation_interval = std::chrono::milliseconds(1000);
    rule.percentiles = {0.5, 0.95};
    rule.compute_rate = true;
    rule.detect_outliers = false;
    
    auto add_result = processor.add_aggregation_rule(rule);
    EXPECT_TRUE(add_result.is_ok());
    
    // Add observations
    for (int i = 1; i <= 100; ++i) {
        auto result = processor.process_observation("test_metric", static_cast<double>(i));
        EXPECT_TRUE(result.is_ok());
    }
    
    // Get current statistics
    auto stats_result = processor.get_current_statistics("test_metric");
    EXPECT_TRUE(stats_result.is_ok());
    
    auto stats = stats_result.value();
    EXPECT_EQ(stats.count, 100);
    EXPECT_GT(stats.mean, 0.0);
}

References kcenon::monitoring::aggregation_processor::add_aggregation_rule(), kcenon::monitoring::aggregation_rule::aggregation_interval, kcenon::monitoring::aggregation_rule::compute_rate, kcenon::monitoring::aggregation_rule::detect_outliers, kcenon::monitoring::aggregation_processor::get_current_statistics(), kcenon::monitoring::aggregation_rule::percentiles, kcenon::monitoring::aggregation_processor::process_observation(), kcenon::monitoring::aggregation_rule::source_metric, kcenon::monitoring::storage, and kcenon::monitoring::aggregation_rule::target_metric_prefix.

Here is the call graph for this function:

TEST_F() [2/18]

TEST_F	(	StreamAggregationTest	,
		AggregationProcessorForceAggregation	)

Definition at line 343 of file test_stream_aggregation.cpp.

                                                                    {
    auto storage = std::make_shared<metric_storage>();
    aggregation_processor processor(storage);
    
    aggregation_rule rule;
    rule.source_metric = "response_time";
    rule.target_metric_prefix = "response_time_agg";
    rule.aggregation_interval = std::chrono::hours(1);  // Long interval
    rule.percentiles = {0.5, 0.9, 0.95, 0.99};
    
    processor.add_aggregation_rule(rule);
    
    // Add observations
    auto samples = generate_normal_samples(1000, 100.0, 20.0);
    for (double sample : samples) {
        processor.process_observation("response_time", sample);
    }
    
    // Force aggregation before interval
    auto result = processor.force_aggregation("response_time");
    EXPECT_TRUE(result.is_ok());
    
    auto agg_result = result.value();
    EXPECT_EQ(agg_result.source_metric, "response_time");
    EXPECT_EQ(agg_result.samples_processed, 1000);
    EXPECT_GE(agg_result.processing_duration.count(), 0);  // May be 0 on fast systems
    
    // Check that aggregated metrics were stored
    storage->flush();
    auto latest = storage->get_latest_value("response_time_agg.mean");
    EXPECT_TRUE(latest.is_ok());
    EXPECT_NEAR(latest.value(), 100.0, 10.0);
}

References kcenon::monitoring::aggregation_processor::add_aggregation_rule(), kcenon::monitoring::aggregation_rule::aggregation_interval, kcenon::monitoring::aggregation_processor::force_aggregation(), kcenon::monitoring::aggregation_rule::percentiles, kcenon::monitoring::aggregation_processor::process_observation(), kcenon::monitoring::aggregation_rule::source_metric, kcenon::monitoring::storage, and kcenon::monitoring::aggregation_rule::target_metric_prefix.

Here is the call graph for this function:

TEST_F() [3/18]

TEST_F	(	StreamAggregationTest	,
		AggregationProcessorInvalidRule	)

Definition at line 377 of file test_stream_aggregation.cpp.

                                                               {
    aggregation_processor processor;
 
    // Test invalid rule (empty source metric)
    aggregation_rule invalid_rule;
    invalid_rule.source_metric = "";  // Invalid
    invalid_rule.target_metric_prefix = "test";
 
    auto result = processor.add_aggregation_rule(invalid_rule);
    EXPECT_FALSE(result.is_ok());
 
    // Test duplicate rule
    aggregation_rule valid_rule;
    valid_rule.source_metric = "test_metric";
    valid_rule.target_metric_prefix = "test_stats";
 
    auto result1 = processor.add_aggregation_rule(valid_rule);
    EXPECT_TRUE(result1.is_ok());
 
    auto result2 = processor.add_aggregation_rule(valid_rule);  // Duplicate
    EXPECT_FALSE(result2.is_ok());
}

References kcenon::monitoring::aggregation_processor::add_aggregation_rule(), kcenon::monitoring::aggregation_rule::source_metric, and kcenon::monitoring::aggregation_rule::target_metric_prefix.

Here is the call graph for this function:

TEST_F() [4/18]

TEST_F	(	StreamAggregationTest	,
		AggregationProcessorMultipleMetrics	)

Definition at line 310 of file test_stream_aggregation.cpp.

                                                                   {
    aggregation_processor processor;
    
    // Add rules for multiple metrics
    std::vector<std::string> metric_names = {"cpu_usage", "memory_usage", "network_io"};
    
    for (const auto& metric_name : metric_names) {
        aggregation_rule rule;
        rule.source_metric = metric_name;
        rule.target_metric_prefix = metric_name + "_stats";
        rule.aggregation_interval = std::chrono::milliseconds(500);
        
        auto result = processor.add_aggregation_rule(rule);
        EXPECT_TRUE(result.is_ok());
    }
    
    // Add observations to each metric
    for (const auto& metric_name : metric_names) {
        for (int i = 1; i <= 50; ++i) {
            auto result = processor.process_observation(metric_name, static_cast<double>(i));
            EXPECT_TRUE(result.is_ok());
        }
    }
    
    // Check configured metrics
    auto configured = processor.get_configured_metrics();
    EXPECT_EQ(configured.size(), 3);
    
    for (const auto& metric_name : metric_names) {
        EXPECT_NE(std::find(configured.begin(), configured.end(), metric_name), configured.end());
    }
}

References kcenon::monitoring::aggregation_processor::add_aggregation_rule(), kcenon::monitoring::aggregation_rule::aggregation_interval, kcenon::monitoring::aggregation_processor::get_configured_metrics(), kcenon::monitoring::aggregation_processor::process_observation(), kcenon::monitoring::aggregation_rule::source_metric, and kcenon::monitoring::aggregation_rule::target_metric_prefix.

Here is the call graph for this function:

TEST_F() [5/18]

TEST_F	(	StreamAggregationTest	,
		ConfigurationValidation	)

Definition at line 458 of file test_stream_aggregation.cpp.

                                                       {
    // Test invalid stream aggregator config
    stream_aggregator_config invalid_config;
    invalid_config.window_size = 0;  // Invalid
 
    auto validation = invalid_config.validate();
    EXPECT_FALSE(validation.is_ok());
 
    // Test valid config
    stream_aggregator_config valid_config;
    valid_config.window_size = 1000;
    valid_config.window_duration = std::chrono::milliseconds(60000);
 
    validation = valid_config.validate();
    EXPECT_TRUE(validation.is_ok());
 
    // Test invalid aggregation rule
    aggregation_rule invalid_rule;
    invalid_rule.source_metric = "test";
    invalid_rule.target_metric_prefix = "";  // Invalid
 
    validation = invalid_rule.validate();
    EXPECT_FALSE(validation.is_ok());
}

References kcenon::monitoring::aggregation_rule::source_metric, kcenon::monitoring::aggregation_rule::target_metric_prefix, kcenon::monitoring::aggregation_rule::validate(), kcenon::monitoring::stream_aggregator_config::validate(), kcenon::monitoring::stream_aggregator_config::window_duration, and kcenon::monitoring::stream_aggregator_config::window_size.

Here is the call graph for this function:

TEST_F() [6/18]

TEST_F	(	StreamAggregationTest	,
		MovingWindowBasic	)

Definition at line 143 of file test_stream_aggregation.cpp.

                                                 {
    moving_window_aggregator<double> window(std::chrono::milliseconds(1000), 100);
    
    auto now = std::chrono::system_clock::now();
    
    // Add values
    for (int i = 0; i < 10; ++i) {
        window.add_value(static_cast<double>(i), now + std::chrono::milliseconds(i * 10));
    }
    
    EXPECT_EQ(window.size(), 10);
    
    auto values = window.get_values();
    EXPECT_EQ(values.size(), 10);
    
    // Check values are correct
    for (size_t i = 0; i < values.size(); ++i) {
        EXPECT_EQ(values[i], static_cast<double>(i));
    }
}

References kcenon::monitoring::moving_window_aggregator< T >::add_value(), kcenon::monitoring::moving_window_aggregator< T >::get_values(), and kcenon::monitoring::moving_window_aggregator< T >::size().

Here is the call graph for this function:

TEST_F() [7/18]

TEST_F	(	StreamAggregationTest	,
		MovingWindowExpiration	)

Definition at line 164 of file test_stream_aggregation.cpp.

                                                      {
    moving_window_aggregator<double> window(std::chrono::milliseconds(100), 1000);
    
    auto now = std::chrono::system_clock::now();
    
    // Add old values (should expire)
    for (int i = 0; i < 5; ++i) {
        window.add_value(static_cast<double>(i), now - std::chrono::milliseconds(200));
    }
    
    // Add new values (should remain)
    for (int i = 10; i < 15; ++i) {
        window.add_value(static_cast<double>(i), now);
    }
    
    auto values = window.get_values();
    
    // Should only have the new values
    EXPECT_EQ(values.size(), 5);
    for (size_t i = 0; i < values.size(); ++i) {
        EXPECT_EQ(values[i], static_cast<double>(i + 10));
    }
}

References kcenon::monitoring::moving_window_aggregator< T >::add_value(), and kcenon::monitoring::moving_window_aggregator< T >::get_values().

Here is the call graph for this function:

TEST_F() [8/18]

TEST_F	(	StreamAggregationTest	,
		OnlineStatisticsBasic	)

Definition at line 68 of file test_stream_aggregation.cpp.

                                                     {
    online_statistics stats;
    
    EXPECT_EQ(stats.count(), 0);
    EXPECT_EQ(stats.mean(), 0.0);
    EXPECT_EQ(stats.variance(), 0.0);
    
    // Add some values
    std::vector<double> values = {1.0, 2.0, 3.0, 4.0, 5.0};
    for (double value : values) {
        stats.add_value(value);
    }
    
    EXPECT_EQ(stats.count(), 5);
    EXPECT_DOUBLE_EQ(stats.mean(), 3.0);  // (1+2+3+4+5)/5 = 3
    
    auto full_stats = stats.get_statistics();
    EXPECT_EQ(full_stats.count, 5);
    EXPECT_DOUBLE_EQ(full_stats.mean, 3.0);
    EXPECT_DOUBLE_EQ(full_stats.min_value, 1.0);
    EXPECT_DOUBLE_EQ(full_stats.max_value, 5.0);
    EXPECT_GT(full_stats.variance, 0.0);
    EXPECT_GT(full_stats.std_deviation, 0.0);
}

References kcenon::monitoring::online_statistics::add_value(), kcenon::monitoring::online_statistics::count(), kcenon::monitoring::online_statistics::get_statistics(), kcenon::monitoring::online_statistics::mean(), and kcenon::monitoring::online_statistics::variance().

Here is the call graph for this function:

TEST_F() [9/18]

TEST_F	(	StreamAggregationTest	,
		OnlineStatisticsLargeDataset	)

Definition at line 93 of file test_stream_aggregation.cpp.

                                                            {
    online_statistics stats;
    
    // Generate a large dataset with known properties
    auto samples = generate_normal_samples(10000, 100.0, 15.0);
    
    for (double sample : samples) {
        stats.add_value(sample);
    }
    
    auto full_stats = stats.get_statistics();
    
    // Check that the statistics are close to the expected values
    EXPECT_NEAR(full_stats.mean, 100.0, 1.0);  // Should be close to 100
    EXPECT_NEAR(full_stats.std_deviation, 15.0, 1.0);  // Should be close to 15
    EXPECT_EQ(full_stats.count, 10000);
}

References kcenon::monitoring::online_statistics::add_value(), and kcenon::monitoring::online_statistics::get_statistics().

Here is the call graph for this function:

TEST_F() [10/18]

TEST_F	(	StreamAggregationTest	,
		PearsonCorrelation	)

Definition at line 401 of file test_stream_aggregation.cpp.

                                                  {
    // Perfect positive correlation
    std::vector<double> x1 = {1, 2, 3, 4, 5};
    std::vector<double> y1 = {2, 4, 6, 8, 10};
    
    double corr1 = pearson_correlation(x1, y1);
    EXPECT_NEAR(corr1, 1.0, 0.001);
    
    // Perfect negative correlation
    std::vector<double> x2 = {1, 2, 3, 4, 5};
    std::vector<double> y2 = {5, 4, 3, 2, 1};
    
    double corr2 = pearson_correlation(x2, y2);
    EXPECT_NEAR(corr2, -1.0, 0.001);
    
    // No correlation
    std::vector<double> x3 = {1, 2, 3, 4, 5};
    std::vector<double> y3 = {3, 3, 3, 3, 3};  // Constant
    
    double corr3 = pearson_correlation(x3, y3);
    EXPECT_NEAR(corr3, 0.0, 0.001);
    
    // Different sizes (should return 0)
    std::vector<double> x4 = {1, 2, 3};
    std::vector<double> y4 = {1, 2, 3, 4, 5};
    
    double corr4 = pearson_correlation(x4, y4);
    EXPECT_EQ(corr4, 0.0);
}

References kcenon::monitoring::pearson_correlation().

Here is the call graph for this function:

TEST_F() [11/18]

TEST_F	(	StreamAggregationTest	,
		QuantileEstimatorMedian	)

Definition at line 112 of file test_stream_aggregation.cpp.

                                                       {
    quantile_estimator median_estimator(0.5);
    
    // Add values 1 through 100
    for (int i = 1; i <= 100; ++i) {
        median_estimator.add_observation(static_cast<double>(i));
    }
    
    double estimated_median = median_estimator.get_quantile();
    
    // For 1-100, median should be around 50.5
    EXPECT_NEAR(estimated_median, 50.5, 5.0);  // Allow some tolerance for P² algorithm
}

References kcenon::monitoring::quantile_estimator::add_observation(), and kcenon::monitoring::quantile_estimator::get_quantile().

Here is the call graph for this function:

TEST_F() [12/18]

TEST_F	(	StreamAggregationTest	,
		QuantileEstimatorPercentiles	)

Definition at line 126 of file test_stream_aggregation.cpp.

                                                            {
    quantile_estimator p95_estimator(0.95);
    
    // Add uniform samples 0-100
    auto samples = generate_uniform_samples(1000, 0.0, 100.0);
    
    for (double sample : samples) {
        p95_estimator.add_observation(sample);
    }
    
    double p95 = p95_estimator.get_quantile();
    
    // 95th percentile should be around 95
    EXPECT_NEAR(p95, 95.0, 10.0);  // P² algorithm has some approximation error
}

References kcenon::monitoring::quantile_estimator::add_observation(), and kcenon::monitoring::quantile_estimator::get_quantile().

Here is the call graph for this function:

TEST_F() [13/18]

TEST_F	(	StreamAggregationTest	,
		StandardAggregationRules	)

Definition at line 431 of file test_stream_aggregation.cpp.

                                                        {
    auto rules = create_standard_aggregation_rules();
    
    EXPECT_GT(rules.size(), 0);
    
    // Validate all rules
    for (const auto& rule : rules) {
        auto validation = rule.validate();
        EXPECT_TRUE(validation.is_ok()) << "Rule validation failed for: " << rule.source_metric;
    }
    
    // Check that standard metrics are included
    std::vector<std::string> expected_metrics = {"response_time", "request_count", "error_count"};
    
    for (const auto& expected : expected_metrics) {
        bool found = false;
        for (const auto& rule : rules) {
            if (rule.source_metric == expected) {
                found = true;
                break;
            }
        }
        EXPECT_TRUE(found) << "Expected metric not found: " << expected;
    }
}

References kcenon::monitoring::create_standard_aggregation_rules().

Here is the call graph for this function:

TEST_F() [14/18]

TEST_F	(	StreamAggregationTest	,
		StreamAggregatorBasic	)

Definition at line 189 of file test_stream_aggregation.cpp.

                                                     {
    stream_aggregator_config config;
    config.window_size = 1000;
    config.enable_outlier_detection = false;  // Disable for predictable testing
    
    stream_aggregator aggregator(config);
    
    // Add observations
    std::vector<double> values = {1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0, 10.0};
    for (double value : values) {
        auto result = aggregator.add_observation(value);
        EXPECT_TRUE(result.is_ok());
    }
    
    auto stats = aggregator.get_statistics();
    
    EXPECT_EQ(stats.count, 10);
    EXPECT_DOUBLE_EQ(stats.mean, 5.5);
    EXPECT_DOUBLE_EQ(stats.min_value, 1.0);
    EXPECT_DOUBLE_EQ(stats.max_value, 10.0);
    EXPECT_GT(stats.std_deviation, 0.0);
    
    // Check percentiles
    EXPECT_GT(stats.percentiles.size(), 0);
}

References kcenon::monitoring::stream_aggregator::add_observation(), kcenon::monitoring::stream_aggregator_config::enable_outlier_detection, kcenon::monitoring::stream_aggregator::get_statistics(), and kcenon::monitoring::stream_aggregator_config::window_size.

Here is the call graph for this function:

TEST_F() [15/18]

TEST_F	(	StreamAggregationTest	,
		StreamAggregatorOutlierDetection	)

Definition at line 238 of file test_stream_aggregation.cpp.

                                                                {
    stream_aggregator_config config;
    config.enable_outlier_detection = true;
    config.outlier_threshold = 2.0;  // 2 standard deviations
    
    stream_aggregator aggregator(config);
    
    // Add normal values around 50
    for (int i = 45; i <= 55; ++i) {
        aggregator.add_observation(static_cast<double>(i));
    }
    
    // Add outliers
    aggregator.add_observation(100.0);  // Should be detected as outlier
    aggregator.add_observation(0.0);    // Should be detected as outlier
    
    auto stats = aggregator.get_statistics();
    
    EXPECT_GT(stats.outlier_count, 0);
    EXPECT_GT(stats.outliers.size(), 0);
}

References kcenon::monitoring::stream_aggregator::add_observation(), kcenon::monitoring::stream_aggregator_config::enable_outlier_detection, kcenon::monitoring::stream_aggregator::get_statistics(), and kcenon::monitoring::stream_aggregator_config::outlier_threshold.

Here is the call graph for this function:

TEST_F() [16/18]

TEST_F	(	StreamAggregationTest	,
		StreamAggregatorPercentiles	)

Definition at line 215 of file test_stream_aggregation.cpp.

                                                           {
    stream_aggregator aggregator;
    
    // Add 100 values from 1 to 100
    for (int i = 1; i <= 100; ++i) {
        aggregator.add_observation(static_cast<double>(i));
    }
    
    // Get specific percentiles
    auto p50_result = aggregator.get_percentile(0.5);
    auto p95_result = aggregator.get_percentile(0.95);
    auto p99_result = aggregator.get_percentile(0.99);
    
    EXPECT_TRUE(p50_result);
    EXPECT_TRUE(p95_result);
    EXPECT_TRUE(p99_result);
    
    // Check approximate values
    EXPECT_NEAR(p50_result.value(), 50.0, 10.0);
    EXPECT_NEAR(p95_result.value(), 95.0, 10.0);
    EXPECT_NEAR(p99_result.value(), 99.0, 10.0);
}

References kcenon::monitoring::stream_aggregator::add_observation(), and kcenon::monitoring::stream_aggregator::get_percentile().

Here is the call graph for this function:

TEST_F() [17/18]

TEST_F	(	StreamAggregationTest	,
		StreamAggregatorReset	)

Definition at line 260 of file test_stream_aggregation.cpp.

                                                     {
    stream_aggregator aggregator;
    
    // Add some observations
    for (int i = 1; i <= 10; ++i) {
        aggregator.add_observation(static_cast<double>(i));
    }
    
    EXPECT_EQ(aggregator.count(), 10);
    
    // Reset
    aggregator.reset();
    
    EXPECT_EQ(aggregator.count(), 0);
    EXPECT_EQ(aggregator.mean(), 0.0);
    EXPECT_EQ(aggregator.variance(), 0.0);
}

References kcenon::monitoring::stream_aggregator::add_observation(), kcenon::monitoring::stream_aggregator::count(), kcenon::monitoring::stream_aggregator::mean(), kcenon::monitoring::stream_aggregator::reset(), and kcenon::monitoring::stream_aggregator::variance().

Here is the call graph for this function:

TEST_F() [18/18]

TEST_F	(	StreamAggregationTest	,
		StreamAggregatorThreadSafety	)

Definition at line 484 of file test_stream_aggregation.cpp.

                                                            {
    stream_aggregator aggregator;
    
    const int num_threads = 4;
    const int observations_per_thread = 1000;
    std::vector<std::thread> threads;
    
    // Launch multiple threads adding observations
    for (int t = 0; t < num_threads; ++t) {
        threads.emplace_back([&aggregator, t, observations_per_thread]() {
            std::random_device rd;
            std::mt19937 gen(rd());
            std::uniform_real_distribution<> dis(0.0, 100.0);
            
            for (int i = 0; i < observations_per_thread; ++i) {
                double value = dis(gen);
                aggregator.add_observation(value);
                
                // Small delay to increase chance of contention
                std::this_thread::sleep_for(std::chrono::microseconds(1));
            }
        });
    }
    
    // Wait for all threads to complete
    for (auto& thread : threads) {
        thread.join();
    }
    
    // Verify we processed all observations
    EXPECT_EQ(aggregator.count(), num_threads * observations_per_thread);
    
    auto stats = aggregator.get_statistics();
    EXPECT_GT(stats.mean, 0.0);
    EXPECT_GT(stats.std_deviation, 0.0);
}

References kcenon::monitoring::stream_aggregator::add_observation(), kcenon::monitoring::stream_aggregator::count(), and kcenon::monitoring::stream_aggregator::get_statistics().

Here is the call graph for this function: