Thread-safe metric storage with ring buffer buffering. More...

#include <metric_storage.h>

Collaboration diagram for kcenon::monitoring::metric_storage:

Public Member Functions
	metric_storage (const metric_storage_config &config={})
	Constructor with configuration.

	~metric_storage ()
	Destructor.

	metric_storage (const metric_storage &)=delete

metric_storage &	operator= (const metric_storage &)=delete

common::VoidResult	store_metric (const std::string &name, double value, metric_type type=metric_type::gauge)
	Store a single metric value.

size_t	store_metrics_batch (const metric_batch &batch)
	Store a batch of metrics.

void	flush ()
	Flush buffered metrics to time series.

common::Result< double >	get_latest_value (const std::string &name) const
	Get the latest value for a metric.

std::vector< std::string >	get_metric_names () const
	Get all metric names.

common::Result< aggregation_result >	query_metric (const std::string &name, const time_series_query &query) const
	Query metric data.

const metric_storage_stats &	get_stats () const noexcept
	Get storage statistics.

const metric_storage_config &	get_config () const noexcept
	Get configuration.

void	clear ()
	Clear all stored metrics.

size_t	series_count () const
	Get number of active metric series.

size_t	memory_footprint () const
	Get memory footprint estimate.

Private Member Functions
void	background_processor ()
	Background processing loop.

time_series *	get_or_create_series (const std::string &name)
	Get or create time series for a metric.

Private Attributes
std::shared_mutex	mutex_

metric_storage_config	config_

metric_storage_stats	stats_

std::unique_ptr< ring_buffer< compact_metric_value > >	incoming_buffer_

std::unordered_map< std::string, std::unique_ptr< time_series > >	time_series_map_

std::unordered_map< uint32_t, std::string >	hash_to_name_

std::atomic< bool >	running_ {false}

std::thread	background_thread_

Detailed Description

Thread-safe metric storage with ring buffer buffering.

Provides efficient metric storage using ring buffers for incoming data and time series for historical queries. Supports background processing for automatic flushing.

Definition at line 95 of file metric_storage.h.

Constructor & Destructor Documentation

◆ metric_storage() [1/2]

kcenon::monitoring::metric_storage::metric_storage ( const metric_storage_config & config = {} )

inlineexplicit

Constructor with configuration.

Parameters

config Metric storage configuration options

Exceptions

std::invalid_argument if configuration validation fails

Definition at line 166 of file metric_storage.h.

                                                                  {})
        : config_(config) {
 
        auto validation = config_.validate();
        if (validation.is_err()) {
            throw std::invalid_argument("Invalid metric storage configuration: " +
                                       validation.error().message);
        }
 
        // Initialize ring buffer
        ring_buffer_config rb_config;
        rb_config.capacity = config_.ring_buffer_capacity;
        rb_config.overwrite_old = true;
        rb_config.batch_size = (std::min)(rb_config.capacity / 2, size_t(64));
        if (rb_config.batch_size == 0) rb_config.batch_size = 1;
        incoming_buffer_ = std::make_unique<ring_buffer<compact_metric_value>>(rb_config);
 
        // Start background processing if enabled
        if (config_.enable_background_processing) {
            running_.store(true, std::memory_order_release);
            background_thread_ = std::thread(&metric_storage::background_processor, this);
        }
    }

Referenced by memory_footprint().

Here is the caller graph for this function:

◆ ~metric_storage()

kcenon::monitoring::metric_storage::~metric_storage ( )

inline

Destructor.

Definition at line 193 of file metric_storage.h.

                      {
        if (running_.load(std::memory_order_acquire)) {
            running_.store(false, std::memory_order_release);
            if (background_thread_.joinable()) {
                background_thread_.join();
            }
        }
    }

References background_thread_, and running_.

◆ metric_storage() [2/2]

kcenon::monitoring::metric_storage::metric_storage ( const metric_storage & )

delete

Member Function Documentation

◆ background_processor()

void kcenon::monitoring::metric_storage::background_processor ( )

inlineprivate

Background processing loop.

Definition at line 117 of file metric_storage.h.

                                {
        while (running_.load(std::memory_order_acquire)) {
            std::this_thread::sleep_for(config_.flush_interval);
            if (running_.load(std::memory_order_acquire)) {
                flush();
            }
        }
    }

References config_, flush(), kcenon::monitoring::metric_storage_config::flush_interval, and running_.

Here is the call graph for this function:

◆ clear()

void kcenon::monitoring::metric_storage::clear ( )

inline

Clear all stored metrics.

Definition at line 366 of file metric_storage.h.

                 {
        std::unique_lock<std::shared_mutex> lock(mutex_);
        incoming_buffer_->clear();
        time_series_map_.clear();
        hash_to_name_.clear();
        stats_.active_metric_series.store(0, std::memory_order_relaxed);
    }

References kcenon::monitoring::metric_storage_stats::active_metric_series, hash_to_name_, incoming_buffer_, mutex_, stats_, and time_series_map_.

◆ flush()

void kcenon::monitoring::metric_storage::flush ( )

inline

Flush buffered metrics to time series.

Definition at line 260 of file metric_storage.h.

                 {
        std::vector<compact_metric_value> flushed_metrics;
        flushed_metrics.reserve(config_.ring_buffer_capacity);
 
        // Read all available metrics from buffer
        incoming_buffer_->read_batch(flushed_metrics, config_.ring_buffer_capacity);
 
        if (flushed_metrics.empty()) {
            return;
        }
 
        std::unique_lock<std::shared_mutex> lock(mutex_);
 
        for (auto& metric : flushed_metrics) {
            // Find metric name from hash
            auto name_it = hash_to_name_.find(metric.metadata.name_hash);
            if (name_it == hash_to_name_.end()) {
                continue;
            }
 
            const std::string& name = name_it->second;
 
            // Get or create time series
            auto* series = get_or_create_series(name);
            if (series == nullptr) {
                stats_.failed_flushes.fetch_add(1, std::memory_order_relaxed);
                continue;
            }
 
            // Add data point to time series
            series->add_point(metric.as_double(), metric.get_timestamp());
        }
 
        stats_.flush_count.fetch_add(1, std::memory_order_relaxed);
    }

References config_, kcenon::monitoring::metric_storage_stats::failed_flushes, kcenon::monitoring::metric_storage_stats::flush_count, get_or_create_series(), hash_to_name_, incoming_buffer_, mutex_, kcenon::monitoring::metric_storage_config::ring_buffer_capacity, and stats_.

Referenced by background_processor(), and TEST_F().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ get_config()

const metric_storage_config & kcenon::monitoring::metric_storage::get_config ( ) const

inlinenoexcept

Get configuration.

Definition at line 359 of file metric_storage.h.

                                                             {
        return config_;
    }

References config_.

◆ get_latest_value()

common::Result< double > kcenon::monitoring::metric_storage::get_latest_value ( const std::string & name ) const

inline

Get the latest value for a metric.

Parameters

name	Metric name

Returns: Optional containing the latest value if available

Definition at line 301 of file metric_storage.h.

                                                                       {
        std::shared_lock<std::shared_mutex> lock(mutex_);
 
        auto it = time_series_map_.find(name);
        if (it == time_series_map_.end()) {
            return common::Result<double>::err(error_info(monitoring_error_code::collection_failed,
                                      "Metric not found: " + name, "monitoring_system").to_common_error());
        }
 
        return it->second->get_latest_value();
    }

References kcenon::monitoring::collection_failed, mutex_, and time_series_map_.

◆ get_metric_names()

std::vector< std::string > kcenon::monitoring::metric_storage::get_metric_names ( ) const

inline

Get all metric names.

Returns: Vector of metric names

Definition at line 317 of file metric_storage.h.

                                                  {
        std::shared_lock<std::shared_mutex> lock(mutex_);
 
        std::vector<std::string> names;
        names.reserve(time_series_map_.size());
 
        for (const auto& pair : time_series_map_) {
            names.push_back(pair.first);
        }
 
        return names;
    }

References mutex_, and time_series_map_.

◆ get_or_create_series()

time_series * kcenon::monitoring::metric_storage::get_or_create_series ( const std::string & name )

inlineprivate

Get or create time series for a metric.

Definition at line 129 of file metric_storage.h.

                                                             {
        auto it = time_series_map_.find(name);
        if (it != time_series_map_.end()) {
            return it->second.get();
        }
 
        // Check if we're at capacity
        if (time_series_map_.size() >= config_.max_metrics) {
            return nullptr;
        }
 
        // Create new time series
        time_series_config ts_config;
        ts_config.max_points = config_.time_series_max_points;
        ts_config.retention_period = config_.retention_period;
 
        auto result = time_series::create(name, ts_config);
        if (result.is_err()) {
            return nullptr;
        }
 
        auto* ptr = result.value().get();
        time_series_map_[name] = std::move(result.value());
        stats_.active_metric_series.fetch_add(1, std::memory_order_relaxed);
 
        // Store hash mapping
        hash_to_name_[hash_metric_name(name)] = name;
 
        return ptr;
    }

Referenced by flush().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ get_stats()

const metric_storage_stats & kcenon::monitoring::metric_storage::get_stats ( ) const

inlinenoexcept

Get storage statistics.

Definition at line 352 of file metric_storage.h.

                                                           {
        return stats_;
    }

References stats_.

◆ memory_footprint()

size_t kcenon::monitoring::metric_storage::memory_footprint ( ) const

inline

Get memory footprint estimate.

Definition at line 385 of file metric_storage.h.

                                    {
        std::shared_lock<std::shared_mutex> lock(mutex_);
 
        size_t total = sizeof(metric_storage);
        total += config_.ring_buffer_capacity * sizeof(compact_metric_value);
 
        for (const auto& pair : time_series_map_) {
            total += pair.first.capacity();
            total += pair.second->memory_footprint();
        }
 
        return total;
    }

References config_, metric_storage(), mutex_, kcenon::monitoring::metric_storage_config::ring_buffer_capacity, and time_series_map_.

Here is the call graph for this function:

◆ operator=()

metric_storage & kcenon::monitoring::metric_storage::operator= ( const metric_storage & )

delete

◆ query_metric()

common::Result< aggregation_result > kcenon::monitoring::metric_storage::query_metric	(	const std::string &	name,
		const time_series_query &	query ) const

inline

Query metric data.

Parameters

name	Metric name
query	Query parameters

Returns: Aggregation result

Definition at line 336 of file metric_storage.h.

                                                                                  {
        std::shared_lock<std::shared_mutex> lock(mutex_);
 
        auto it = time_series_map_.find(name);
        if (it == time_series_map_.end()) {
            return common::Result<aggregation_result>::err(error_info(monitoring_error_code::collection_failed,
                                                 "Metric not found: " + name, "monitoring_system").to_common_error());
        }
 
        return it->second->query(query);
    }

References kcenon::monitoring::collection_failed, mutex_, and time_series_map_.

◆ series_count()

size_t kcenon::monitoring::metric_storage::series_count ( ) const

inline

Get number of active metric series.

Definition at line 377 of file metric_storage.h.

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

References mutex_, and time_series_map_.

◆ store_metric()

common::VoidResult kcenon::monitoring::metric_storage::store_metric	(	const std::string &	name,
		double	value,
		metric_type	type = metric_type::gauge )

inline

Store a single metric value.

Parameters

name	Metric name
value	Metric value
type	Metric type (default: gauge)

Returns: Result indicating success or failure

Definition at line 213 of file metric_storage.h.

                                                                 {
        auto metadata = create_metric_metadata(name, type);
        compact_metric_value metric(metadata, value);
 
        auto result = incoming_buffer_->write(std::move(metric));
        if (result.is_ok()) {
            stats_.total_metrics_stored.fetch_add(1, std::memory_order_relaxed);
 
            // Store name mapping
            std::unique_lock<std::shared_mutex> lock(mutex_);
            if (hash_to_name_.find(metadata.name_hash) == hash_to_name_.end()) {
                hash_to_name_[metadata.name_hash] = name;
            }
        } else {
            stats_.total_metrics_dropped.fetch_add(1, std::memory_order_relaxed);
        }
 
        return result;
    }

References kcenon::monitoring::create_metric_metadata(), hash_to_name_, incoming_buffer_, mutex_, stats_, kcenon::monitoring::metric_storage_stats::total_metrics_dropped, and kcenon::monitoring::metric_storage_stats::total_metrics_stored.

Here is the call graph for this function:

◆ store_metrics_batch()

size_t kcenon::monitoring::metric_storage::store_metrics_batch ( const metric_batch & batch )

inline

Store a batch of metrics.

Parameters

batch Metric batch to store

Returns: Number of metrics successfully stored

Definition at line 240 of file metric_storage.h.

                                                          {
        size_t stored = 0;
 
        for (const auto& metric : batch.metrics) {
            compact_metric_value copy = metric;
            auto result = incoming_buffer_->write(std::move(copy));
            if (result.is_ok()) {
                stored++;
                stats_.total_metrics_stored.fetch_add(1, std::memory_order_relaxed);
            } else {
                stats_.total_metrics_dropped.fetch_add(1, std::memory_order_relaxed);
            }
        }
 
        return stored;
    }

References incoming_buffer_, kcenon::monitoring::metric_batch::metrics, stats_, kcenon::monitoring::metric_storage_stats::total_metrics_dropped, and kcenon::monitoring::metric_storage_stats::total_metrics_stored.

Member Data Documentation

◆ background_thread_

std::thread kcenon::monitoring::metric_storage::background_thread_

private

Definition at line 112 of file metric_storage.h.

Referenced by ~metric_storage().

◆ config_

metric_storage_config kcenon::monitoring::metric_storage::config_

private

Definition at line 98 of file metric_storage.h.

Referenced by background_processor(), flush(), get_config(), get_or_create_series(), and memory_footprint().

◆ hash_to_name_

std::unordered_map<uint32_t, std::string> kcenon::monitoring::metric_storage::hash_to_name_

private

Definition at line 108 of file metric_storage.h.

Referenced by clear(), flush(), get_or_create_series(), and store_metric().

◆ incoming_buffer_

std::unique_ptr<ring_buffer<compact_metric_value> > kcenon::monitoring::metric_storage::incoming_buffer_

private

Definition at line 102 of file metric_storage.h.

Referenced by clear(), flush(), store_metric(), and store_metrics_batch().

◆ mutex_

std::shared_mutex kcenon::monitoring::metric_storage::mutex_

mutableprivate

Definition at line 97 of file metric_storage.h.

Referenced by clear(), flush(), get_latest_value(), get_metric_names(), memory_footprint(), query_metric(), series_count(), and store_metric().

◆ running_

std::atomic<bool> kcenon::monitoring::metric_storage::running_ {false}

private

Definition at line 111 of file metric_storage.h.

111{false};

Referenced by background_processor(), and ~metric_storage().

◆ stats_

metric_storage_stats kcenon::monitoring::metric_storage::stats_

mutableprivate

Definition at line 99 of file metric_storage.h.

Referenced by clear(), flush(), get_or_create_series(), get_stats(), store_metric(), and store_metrics_batch().

◆ time_series_map_

std::unordered_map<std::string, std::unique_ptr<time_series> > kcenon::monitoring::metric_storage::time_series_map_

private

Definition at line 105 of file metric_storage.h.

Referenced by clear(), get_latest_value(), get_metric_names(), get_or_create_series(), memory_footprint(), query_metric(), and series_count().

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

include/kcenon/monitoring/utils/metric_storage.h

Public Member Functions

Private Member Functions

Private Attributes

Detailed Description

Constructor & Destructor Documentation

◆ metric_storage() [1/2]

◆ ~metric_storage()

◆ metric_storage() [2/2]

Member Function Documentation

◆ background_processor()

◆ clear()

◆ flush()

◆ get_config()

◆ get_latest_value()

◆ get_metric_names()

◆ get_or_create_series()

◆ get_stats()

◆ memory_footprint()

◆ operator=()

◆ query_metric()

◆ series_count()

◆ store_metric()

◆ store_metrics_batch()

Member Data Documentation

◆ background_thread_

◆ config_

◆ hash_to_name_

◆ incoming_buffer_

◆ mutex_

◆ running_

◆ stats_

◆ time_series_map_