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

#include <metric_storage.h>

Collaboration diagram for kcenon::monitoring::metric_storage:

Classes
struct	validated_tag

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

	metric_storage (const metric_storage &)=delete

metric_storage &	operator= (const metric_storage &)=delete

	~metric_storage ()
	Destructor.

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.

Static Public Member Functions
static common::Result< std::unique_ptr< metric_storage > >	create (const metric_storage_config &config={})
	Create a metric storage with validated configuration.

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.

void	init_internals ()

	metric_storage (const metric_storage_config &config, validated_tag)

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/3]

kcenon::monitoring::metric_storage::metric_storage	(	const metric_storage_config &	config,
		validated_tag	)

inlineprivate

Definition at line 177 of file metric_storage.h.

        : config_(config) {
        init_internals();
    }

References init_internals().

Referenced by memory_footprint().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ metric_storage() [2/3]

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

Deprecated: Use create() for Result-based error handling

Definition at line 208 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);
        }
 
        init_internals();
    }

◆ metric_storage() [3/3]

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

delete

◆ ~metric_storage()

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

inline

Destructor.

Definition at line 227 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_.

Member Function Documentation

◆ background_processor()

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

inlineprivate

Background processing loop.

Definition at line 119 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_.

Referenced by init_internals().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ clear()

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

inline

Clear all stored metrics.

Definition at line 396 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_.

◆ create()

static common::Result< std::unique_ptr< metric_storage > > kcenon::monitoring::metric_storage::create ( const metric_storage_config & config = {} )

inlinestatic

Create a metric storage with validated configuration.

Parameters

config Metric storage configuration options

Returns: Result containing the metric storage or error

Definition at line 188 of file metric_storage.h.

                                              {}) {
        auto validation = config.validate();
        if (validation.is_err()) {
            return common::Result<std::unique_ptr<metric_storage>>::err(
                error_info(monitoring_error_code::invalid_configuration,
                           "Invalid metric storage configuration: " +
                               validation.error().message)
                    .to_common_error());
        }
        return common::ok(std::unique_ptr<metric_storage>(
            new metric_storage(config, validated_tag{})));
    }

Referenced by TEST_F(), and TEST_F().

Here is the caller graph for this function:

◆ flush()

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

inline

Flush buffered metrics to time series.

Definition at line 290 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 389 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 331 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 347 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 131 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 382 of file metric_storage.h.

                                                           {
        return stats_;
    }

References stats_.

◆ init_internals()

void kcenon::monitoring::metric_storage::init_internals ( )

inlineprivate

Definition at line 162 of file metric_storage.h.

                          {
        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);
 
        if (config_.enable_background_processing) {
            running_.store(true, std::memory_order_release);
            background_thread_ = std::thread(&metric_storage::background_processor, this);
        }
    }

References background_processor(), background_thread_, kcenon::monitoring::ring_buffer_config::batch_size, kcenon::monitoring::ring_buffer_config::capacity, config_, kcenon::monitoring::metric_storage_config::enable_background_processing, incoming_buffer_, kcenon::monitoring::ring_buffer_config::overwrite_old, kcenon::monitoring::metric_storage_config::ring_buffer_capacity, and running_.

Referenced by metric_storage().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ memory_footprint()

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

inline

Get memory footprint estimate.

Definition at line 415 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 366 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 407 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 243 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 270 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 114 of file metric_storage.h.

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

◆ config_

metric_storage_config kcenon::monitoring::metric_storage::config_

private

Definition at line 100 of file metric_storage.h.

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

◆ hash_to_name_

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

private

Definition at line 110 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 104 of file metric_storage.h.

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

◆ mutex_

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

mutableprivate

Definition at line 99 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 113 of file metric_storage.h.

113{false};

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

◆ stats_

metric_storage_stats kcenon::monitoring::metric_storage::stats_

mutableprivate

Definition at line 101 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 107 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

Classes

Public Member Functions

Static Public Member Functions

Private Member Functions

Private Attributes

Detailed Description

Constructor & Destructor Documentation

◆ metric_storage() [1/3]

◆ metric_storage() [2/3]

◆ metric_storage() [3/3]

◆ ~metric_storage()

Member Function Documentation

◆ background_processor()

◆ clear()

◆ create()

◆ flush()

◆ get_config()

◆ get_latest_value()

◆ get_metric_names()

◆ get_or_create_series()

◆ get_stats()

◆ init_internals()

◆ 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_