aggregation_processor.h

Go to the documentation of this file.

// BSD 3-Clause License
// Copyright (c) 2021-2025, 🍀☀🌕🌥 🌊
// See the LICENSE file in the project root for full license information.
 
#pragma once
 
#include "../core/result_types.h"
#include "../core/error_codes.h"
#include "stream_aggregator.h"
#include "metric_storage.h"
 
#include <chrono>
#include <functional>
#include <memory>
#include <mutex>
#include <shared_mutex>
#include <string>
#include <unordered_map>
#include <vector>
 
namespace kcenon::monitoring {
 
struct aggregation_rule {
    std::string source_metric;                                  
    std::string target_metric_prefix;                           
    std::chrono::milliseconds aggregation_interval{60000};      
    std::vector<double> percentiles = {0.5, 0.9, 0.95, 0.99};   
    bool compute_rate = false;                                  
    bool detect_outliers = true;                                
    double outlier_threshold = 3.0;                             
 
    common::VoidResult validate() const {
        if (source_metric.empty()) {
            return common::VoidResult::err(error_info(monitoring_error_code::invalid_configuration,
                                   "Source metric name cannot be empty").to_common_error());
        }
        if (target_metric_prefix.empty()) {
            return common::VoidResult::err(error_info(monitoring_error_code::invalid_configuration,
                                   "Target metric prefix cannot be empty").to_common_error());
        }
        if (aggregation_interval.count() <= 0) {
            return common::VoidResult::err(error_info(monitoring_error_code::invalid_configuration,
                                   "Aggregation interval must be positive").to_common_error());
        }
        return common::ok();
    }
};
 
struct stream_aggregation_result {
    std::string source_metric;                              
    size_t samples_processed = 0;                           
    std::chrono::microseconds processing_duration{0};       
    streaming_statistics statistics;                        
    std::chrono::system_clock::time_point timestamp;        
    bool stored_successfully = false;                       
};
 
// Alias for backward compatibility
using aggregation_result_stream = stream_aggregation_result;
 
class aggregation_processor {
public:
    aggregation_processor() : storage_(nullptr) {}
 
    explicit aggregation_processor(std::shared_ptr<metric_storage> storage)
        : storage_(std::move(storage)) {}
 
    common::VoidResult add_aggregation_rule(const aggregation_rule& rule) {
        auto validation = rule.validate();
        if (validation.is_err()) {
            return validation;
        }
 
        std::unique_lock<std::shared_mutex> lock(mutex_);
 
        // Check for duplicate
        if (aggregators_.find(rule.source_metric) != aggregators_.end()) {
            return common::VoidResult::err(error_info(monitoring_error_code::already_exists,
                                   "Aggregation rule already exists for metric: " +
                                   rule.source_metric).to_common_error());
        }
 
        // Create stream aggregator config
        stream_aggregator_config config;
        config.enable_outlier_detection = rule.detect_outliers;
        config.outlier_threshold = rule.outlier_threshold;
        config.percentiles_to_track = rule.percentiles;
 
        // Create aggregator
        aggregator_entry entry;
        entry.rule = rule;
        entry.aggregator = std::make_unique<stream_aggregator>(config);
        entry.last_aggregation = std::chrono::system_clock::now();
 
        aggregators_.emplace(rule.source_metric, std::move(entry));
 
        return common::ok();
    }
 
    common::VoidResult process_observation(const std::string& metric_name, double value) {
        std::shared_lock<std::shared_mutex> lock(mutex_);
 
        auto it = aggregators_.find(metric_name);
        if (it == aggregators_.end()) {
            // No rule for this metric, silently accept
            return common::ok();
        }
 
        return it->second.aggregator->add_observation(value);
    }
 
    common::Result<streaming_statistics> get_current_statistics(const std::string& metric_name) const {
        std::shared_lock<std::shared_mutex> lock(mutex_);
 
        auto it = aggregators_.find(metric_name);
        if (it == aggregators_.end()) {
            return common::make_error<streaming_statistics>(
                static_cast<int>(monitoring_error_code::metric_not_found),
                "No aggregator found for metric: " + metric_name);
        }
 
        return common::ok(it->second.aggregator->get_statistics());
    }
 
    std::vector<std::string> get_configured_metrics() const {
        std::shared_lock<std::shared_mutex> lock(mutex_);
 
        std::vector<std::string> metrics;
        metrics.reserve(aggregators_.size());
 
        for (const auto& [name, entry] : aggregators_) {
            metrics.push_back(name);
        }
 
        return metrics;
    }
 
    common::Result<stream_aggregation_result> force_aggregation(const std::string& metric_name) {
        auto start_time = std::chrono::steady_clock::now();
 
        std::unique_lock<std::shared_mutex> lock(mutex_);
 
        auto it = aggregators_.find(metric_name);
        if (it == aggregators_.end()) {
            return common::make_error<stream_aggregation_result>(
                static_cast<int>(monitoring_error_code::metric_not_found),
                "No aggregator found for metric: " + metric_name);
        }
 
        auto& entry = it->second;
        auto stats = entry.aggregator->get_statistics();
 
        auto end_time = std::chrono::steady_clock::now();
        auto duration = std::chrono::duration_cast<std::chrono::microseconds>(
            end_time - start_time);
 
        stream_aggregation_result agg_result;
        agg_result.source_metric = metric_name;
        agg_result.samples_processed = stats.count;
        agg_result.processing_duration = duration;
        agg_result.statistics = stats;
        agg_result.timestamp = std::chrono::system_clock::now();
 
        // Store aggregated metrics if storage is available
        if (storage_) {
            const auto& prefix = entry.rule.target_metric_prefix;
 
            storage_->store_metric(prefix + ".mean", stats.mean, metric_type::gauge);
            storage_->store_metric(prefix + ".min", stats.min_value, metric_type::gauge);
            storage_->store_metric(prefix + ".max", stats.max_value, metric_type::gauge);
            storage_->store_metric(prefix + ".stddev", stats.std_deviation, metric_type::gauge);
            storage_->store_metric(prefix + ".count",
                                   static_cast<double>(stats.count), metric_type::counter);
 
            for (const auto& [p, value] : stats.percentiles) {
                std::string suffix = ".p" + std::to_string(static_cast<int>(p * 100));
                storage_->store_metric(prefix + suffix, value, metric_type::gauge);
            }
 
            agg_result.stored_successfully = true;
        }
 
        // Reset the aggregator for next interval
        entry.aggregator->reset();
        entry.last_aggregation = std::chrono::system_clock::now();
 
        return common::ok(agg_result);
    }
 
    common::VoidResult remove_aggregation_rule(const std::string& metric_name) {
        std::unique_lock<std::shared_mutex> lock(mutex_);
 
        auto it = aggregators_.find(metric_name);
        if (it == aggregators_.end()) {
            return common::VoidResult::err(error_info(monitoring_error_code::metric_not_found,
                                   "No aggregator found for metric: " + metric_name).to_common_error());
        }
 
        aggregators_.erase(it);
        return common::ok();
    }
 
    bool has_rule(const std::string& metric_name) const {
        std::shared_lock<std::shared_mutex> lock(mutex_);
        return aggregators_.find(metric_name) != aggregators_.end();
    }
 
    size_t rule_count() const {
        std::shared_lock<std::shared_mutex> lock(mutex_);
        return aggregators_.size();
    }
 
    void clear() {
        std::unique_lock<std::shared_mutex> lock(mutex_);
        aggregators_.clear();
    }
 
private:
    struct aggregator_entry {
        aggregation_rule rule;
        std::unique_ptr<stream_aggregator> aggregator;
        std::chrono::system_clock::time_point last_aggregation;
    };
 
    mutable std::shared_mutex mutex_;
    std::shared_ptr<metric_storage> storage_;
    std::unordered_map<std::string, aggregator_entry> aggregators_;
};
 
inline std::vector<aggregation_rule> create_standard_aggregation_rules() {
    std::vector<aggregation_rule> rules;
 
    // Response time metrics
    {
        aggregation_rule rule;
        rule.source_metric = "response_time";
        rule.target_metric_prefix = "response_time_agg";
        rule.aggregation_interval = std::chrono::milliseconds(60000);
        rule.percentiles = {0.5, 0.9, 0.95, 0.99};
        rule.compute_rate = false;
        rule.detect_outliers = true;
        rules.push_back(rule);
    }
 
    // Request count metrics
    {
        aggregation_rule rule;
        rule.source_metric = "request_count";
        rule.target_metric_prefix = "request_count_agg";
        rule.aggregation_interval = std::chrono::milliseconds(60000);
        rule.percentiles = {0.5, 0.9, 0.95, 0.99};
        rule.compute_rate = true;
        rule.detect_outliers = false;
        rules.push_back(rule);
    }
 
    // Error count metrics
    {
        aggregation_rule rule;
        rule.source_metric = "error_count";
        rule.target_metric_prefix = "error_count_agg";
        rule.aggregation_interval = std::chrono::milliseconds(60000);
        rule.percentiles = {0.5, 0.9, 0.95, 0.99};
        rule.compute_rate = true;
        rule.detect_outliers = true;
        rule.outlier_threshold = 2.0;  // More sensitive for errors
        rules.push_back(rule);
    }
 
    // CPU usage metrics
    {
        aggregation_rule rule;
        rule.source_metric = "cpu_usage";
        rule.target_metric_prefix = "cpu_usage_agg";
        rule.aggregation_interval = std::chrono::milliseconds(60000);
        rule.percentiles = {0.5, 0.9, 0.95, 0.99};
        rule.compute_rate = false;
        rule.detect_outliers = true;
        rules.push_back(rule);
    }
 
    // Memory usage metrics
    {
        aggregation_rule rule;
        rule.source_metric = "memory_usage";
        rule.target_metric_prefix = "memory_usage_agg";
        rule.aggregation_interval = std::chrono::milliseconds(60000);
        rule.percentiles = {0.5, 0.9, 0.95, 0.99};
        rule.compute_rate = false;
        rule.detect_outliers = true;
        rules.push_back(rule);
    }
 
    return rules;
}
 
} // namespace kcenon::monitoring