kcenon::monitoring::aggregation_processor Class Reference

Processes metric streams and generates aggregated statistics. More...

#include <aggregation_processor.h>

Collaboration diagram for kcenon::monitoring::aggregation_processor:

Classes
struct	aggregator_entry

Public Member Functions
	aggregation_processor ()
	Default constructor.
	aggregation_processor (std::shared_ptr< metric_storage > storage)
	Constructor with storage backend.
common::VoidResult	add_aggregation_rule (const aggregation_rule &rule)
	Add an aggregation rule.
common::VoidResult	process_observation (const std::string &metric_name, double value)
	Process an observation for a metric.
common::Result< streaming_statistics >	get_current_statistics (const std::string &metric_name) const
	Get current statistics for a metric.
std::vector< std::string >	get_configured_metrics () const
	Get list of configured metrics.
common::Result< stream_aggregation_result >	force_aggregation (const std::string &metric_name)
	Force aggregation for a metric.
common::VoidResult	remove_aggregation_rule (const std::string &metric_name)
	Remove an aggregation rule.
bool	has_rule (const std::string &metric_name) const
	Check if a metric has an aggregation rule.
size_t	rule_count () const
	Get the number of configured rules.
void	clear ()
	Clear all aggregation rules.

Private Attributes
std::shared_mutex	mutex_
std::shared_ptr< metric_storage >	storage_
std::unordered_map< std::string, aggregator_entry >	aggregators_

Detailed Description

Processes metric streams and generates aggregated statistics.

The aggregation processor manages multiple stream aggregators, one for each configured metric, and periodically computes and stores aggregated statistics.

Definition at line 87 of file aggregation_processor.h.

Constructor & Destructor Documentation

aggregation_processor() [1/2]

kcenon::monitoring::aggregation_processor::aggregation_processor ( )

inline

Default constructor.

Definition at line 92 of file aggregation_processor.h.

: storage_(nullptr) {}

aggregation_processor() [2/2]

kcenon::monitoring::aggregation_processor::aggregation_processor ( std::shared_ptr< metric_storage > storage )

inlineexplicit

Constructor with storage backend.

Parameters

storage

Shared pointer to metric storage

Definition at line 98 of file aggregation_processor.h.

        : storage_(std::move(storage)) {}

Member Function Documentation

add_aggregation_rule()

common::VoidResult kcenon::monitoring::aggregation_processor::add_aggregation_rule ( const aggregation_rule & rule )

inline

Add an aggregation rule.

Parameters

rule	The aggregation rule to add

Returns

Result indicating success or failure

Definition at line 106 of file aggregation_processor.h.

                                                                        {
        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();
    }

References kcenon::monitoring::aggregation_processor::aggregator_entry::aggregator, aggregators_, kcenon::monitoring::already_exists, kcenon::monitoring::aggregation_rule::detect_outliers, kcenon::monitoring::stream_aggregator_config::enable_outlier_detection, kcenon::monitoring::aggregation_processor::aggregator_entry::last_aggregation, mutex_, kcenon::monitoring::aggregation_rule::outlier_threshold, kcenon::monitoring::stream_aggregator_config::outlier_threshold, kcenon::monitoring::aggregation_rule::percentiles, kcenon::monitoring::stream_aggregator_config::percentiles_to_track, kcenon::monitoring::aggregation_processor::aggregator_entry::rule, kcenon::monitoring::aggregation_rule::source_metric, kcenon::monitoring::error_info::to_common_error(), and kcenon::monitoring::aggregation_rule::validate().

Referenced by TEST_F(), TEST_F(), TEST_F(), and TEST_F().

Here is the call graph for this function:

Here is the caller graph for this function:

clear()

void kcenon::monitoring::aggregation_processor::clear ( )

inline

Clear all aggregation rules.

Definition at line 287 of file aggregation_processor.h.

                 {
        std::unique_lock<std::shared_mutex> lock(mutex_);
        aggregators_.clear();
    }

References aggregators_, and mutex_.

force_aggregation()

common::Result< stream_aggregation_result > kcenon::monitoring::aggregation_processor::force_aggregation ( const std::string & metric_name )

inline

Force aggregation for a metric.

Parameters

metric_name

The metric name

Returns

Result containing the aggregation result

Definition at line 196 of file aggregation_processor.h.

                                                                                            {
        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);
    }

References aggregators_, kcenon::monitoring::counter, kcenon::monitoring::gauge, kcenon::monitoring::metric_not_found, mutex_, kcenon::monitoring::stream_aggregation_result::processing_duration, kcenon::monitoring::stream_aggregation_result::samples_processed, kcenon::monitoring::stream_aggregation_result::source_metric, kcenon::monitoring::stream_aggregation_result::statistics, storage_, kcenon::monitoring::stream_aggregation_result::stored_successfully, and kcenon::monitoring::stream_aggregation_result::timestamp.

Referenced by TEST_F().

Here is the caller graph for this function:

get_configured_metrics()

std::vector< std::string > kcenon::monitoring::aggregation_processor::get_configured_metrics ( ) const

inline

Get list of configured metrics.

Returns

Vector of metric names

Definition at line 178 of file aggregation_processor.h.

                                                        {
        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;
    }

References aggregators_, and mutex_.

Referenced by TEST_F().

Here is the caller graph for this function:

get_current_statistics()

common::Result< streaming_statistics > kcenon::monitoring::aggregation_processor::get_current_statistics ( const std::string & metric_name ) const

inline

Get current statistics for a metric.

Parameters

metric_name

The metric name

Returns

Result containing the statistics

Definition at line 161 of file aggregation_processor.h.

                                                                                                  {
        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());
    }

References aggregators_, kcenon::monitoring::metric_not_found, and mutex_.

Referenced by TEST_F().

Here is the caller graph for this function:

has_rule()

bool kcenon::monitoring::aggregation_processor::has_rule ( const std::string & metric_name ) const

inline

Check if a metric has an aggregation rule.

Parameters

metric_name

The metric name

Returns

true if rule exists

Definition at line 271 of file aggregation_processor.h.

                                                      {
        std::shared_lock<std::shared_mutex> lock(mutex_);
        return aggregators_.find(metric_name) != aggregators_.end();
    }

References aggregators_, and mutex_.

process_observation()

common::VoidResult kcenon::monitoring::aggregation_processor::process_observation ( const std::string & metric_name, double value )

inline

Process an observation for a metric.

Parameters

metric_name	The metric name
value	The observed value

Returns

Result indicating success or failure

Definition at line 144 of file aggregation_processor.h.

                                                                                     {
        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);
    }

References aggregators_, and mutex_.

Referenced by TEST_F(), TEST_F(), and TEST_F().

Here is the caller graph for this function:

remove_aggregation_rule()

common::VoidResult kcenon::monitoring::aggregation_processor::remove_aggregation_rule ( const std::string & metric_name )

inline

Remove an aggregation rule.

Parameters

metric_name

The metric name

Returns

Result indicating success or failure

Definition at line 253 of file aggregation_processor.h.

                                                                           {
        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();
    }

References aggregators_, kcenon::monitoring::metric_not_found, and mutex_.

rule_count()

size_t kcenon::monitoring::aggregation_processor::rule_count ( ) const

inline

Get the number of configured rules.

Definition at line 279 of file aggregation_processor.h.

                              {
        std::shared_lock<std::shared_mutex> lock(mutex_);
        return aggregators_.size();
    }

References aggregators_, and mutex_.

Member Data Documentation

aggregators_

std::unordered_map<std::string, aggregator_entry> kcenon::monitoring::aggregation_processor::aggregators_

private

Definition at line 301 of file aggregation_processor.h.

Referenced by add_aggregation_rule(), clear(), force_aggregation(), get_configured_metrics(), get_current_statistics(), has_rule(), process_observation(), remove_aggregation_rule(), and rule_count().

mutex_

std::shared_mutex kcenon::monitoring::aggregation_processor::mutex_

mutableprivate

Definition at line 299 of file aggregation_processor.h.

Referenced by add_aggregation_rule(), clear(), force_aggregation(), get_configured_metrics(), get_current_statistics(), has_rule(), process_observation(), remove_aggregation_rule(), and rule_count().

storage_

std::shared_ptr<metric_storage> kcenon::monitoring::aggregation_processor::storage_

private

Definition at line 300 of file aggregation_processor.h.

Referenced by force_aggregation().

---

The documentation for this class was generated from the following file:

• include/kcenon/monitoring/utils/aggregation_processor.h