Thread-safe time series data storage. More...

#include <time_series.h>

Collaboration diagram for kcenon::monitoring::time_series:

Public Member Functions
common::VoidResult	add_point (double value, std::chrono::system_clock::time_point timestamp=std::chrono::system_clock::now())
	Add a data point.

common::VoidResult	add_points (const std::vector< time_point_data > &points)
	Add multiple data points.

common::Result< aggregation_result >	query (const time_series_query &query) const
	Query data for a time range.

size_t	size () const
	Get current number of data points.

bool	empty () const
	Check if series is empty.

const std::string &	name () const noexcept
	Get series name.

const time_series_config &	get_config () const noexcept
	Get configuration.

void	clear ()
	Clear all data.

common::Result< double >	get_latest_value () const
	Get latest value.

size_t	memory_footprint () const
	Get memory footprint in bytes.

Static Public Member Functions
static common::Result< std::unique_ptr< time_series > >	create (const std::string &name, const time_series_config &config={})
	Factory method to create time_series with validation.

Private Member Functions
void	cleanup_old_data ()
	Cleanup old data points.

void	compress_data ()
	Compress data if enabled.

void	enforce_size_limit ()
	Ensure data size doesn't exceed maximum.

	time_series (const std::string &name, const time_series_config &config)
	Private constructor (use create() factory method)

Private Attributes
std::mutex	mutex_

std::vector< time_point_data >	data_

time_series_config	config_

std::string	series_name_

size_t	insertion_count_ = 0

Detailed Description

Thread-safe time series data storage.

Definition at line 209 of file time_series.h.

Constructor & Destructor Documentation

◆ time_series()

kcenon::monitoring::time_series::time_series	(	const std::string &	name,
		const time_series_config &	config )

inlineprivate

Private constructor (use create() factory method)

Definition at line 282 of file time_series.h.

        : config_(config), series_name_(name) {
        data_.reserve(config_.max_points);
    }

References config_, data_, and kcenon::monitoring::time_series_config::max_points.

Referenced by memory_footprint().

Here is the caller graph for this function:

Member Function Documentation

◆ add_point()

common::VoidResult kcenon::monitoring::time_series::add_point	(	double	value,
		std::chrono::system_clock::time_point	timestamp = std::chrono::system_clock::now() )

inline

Add a data point.

Definition at line 308 of file time_series.h.

                                                             {
        std::lock_guard<std::mutex> lock(mutex_);
 
        time_point_data point(timestamp, value);
 
        // Optimize: if timestamp is newest, append directly (O(1) vs O(n))
        if (data_.empty() || point.timestamp >= data_.back().timestamp) {
            data_.push_back(point);
        } else {
            // Insert in chronological order only when necessary
            auto it = std::upper_bound(data_.begin(), data_.end(), point,
                [](const time_point_data& a, const time_point_data& b) {
                    return a.timestamp < b.timestamp;
                });
 
            data_.insert(it, point);
        }
 
        // Perform maintenance periodically (every 100 insertions) instead of every time
        ++insertion_count_;
        if (insertion_count_ % 100 == 0) {
            cleanup_old_data();
            compress_data();
            enforce_size_limit();
        }
 
        return common::ok();
    }

References cleanup_old_data(), compress_data(), data_, enforce_size_limit(), insertion_count_, mutex_, and kcenon::monitoring::time_point_data::timestamp.

Here is the call graph for this function:

◆ add_points()

common::VoidResult kcenon::monitoring::time_series::add_points ( const std::vector< time_point_data > & points )

inline

Add multiple data points.

Definition at line 342 of file time_series.h.

                                                                          {
        std::lock_guard<std::mutex> lock(mutex_);
 
        for (const auto& point : points) {
            // Optimize: if timestamp is newest, append directly
            if (data_.empty() || point.timestamp >= data_.back().timestamp) {
                data_.push_back(point);
            } else {
                auto it = std::upper_bound(data_.begin(), data_.end(), point,
                    [](const time_point_data& a, const time_point_data& b) {
                        return a.timestamp < b.timestamp;
                    });
 
                data_.insert(it, point);
            }
        }
 
        // Perform maintenance after batch insert
        cleanup_old_data();
        compress_data();
        enforce_size_limit();
 
        return common::ok();
    }

References cleanup_old_data(), compress_data(), data_, enforce_size_limit(), and mutex_.

Here is the call graph for this function:

◆ cleanup_old_data()

void kcenon::monitoring::time_series::cleanup_old_data ( )

inlineprivate

Cleanup old data points.

Definition at line 220 of file time_series.h.

                            {
        auto cutoff = std::chrono::system_clock::now() - config_.retention_period;
        
        auto it = std::remove_if(data_.begin(), data_.end(),
            [cutoff](const time_point_data& point) {
                return !point.is_valid(cutoff);
            });
        
        data_.erase(it, data_.end());
    }

References config_, data_, and kcenon::monitoring::time_series_config::retention_period.

Referenced by add_point(), and add_points().

Here is the caller graph for this function:

◆ clear()

void kcenon::monitoring::time_series::clear ( )

inline

Clear all data.

Definition at line 468 of file time_series.h.

                 {
        std::lock_guard<std::mutex> lock(mutex_);
        data_.clear();
    }

References data_, and mutex_.

◆ compress_data()

void kcenon::monitoring::time_series::compress_data ( )

inlineprivate

Compress data if enabled.

Definition at line 234 of file time_series.h.

                         {
        if (!config_.enable_compression || data_.size() < 3) {
            return;
        }
        
        // Simple compression: remove points that don't add significant information
        std::vector<time_point_data> compressed;
        compressed.reserve(data_.size());
        
        if (!data_.empty()) {
            compressed.push_back(data_[0]);
            
            for (size_t i = 1; i < data_.size() - 1; ++i) {
                const auto& prev = data_[i - 1];
                const auto& curr = data_[i];
                const auto& next = data_[i + 1];
                
                // Check if current point is significantly different from linear interpolation
                double expected = prev.value + (next.value - prev.value) * 
                    (std::chrono::duration<double>(curr.timestamp - prev.timestamp).count() /
                     std::chrono::duration<double>(next.timestamp - prev.timestamp).count());
                
                if (std::abs(curr.value - expected) > config_.compression_threshold) {
                    compressed.push_back(curr);
                }
            }
            
            if (data_.size() > 1) {
                compressed.push_back(data_.back());
            }
        }
        
        data_ = std::move(compressed);
    }

References kcenon::monitoring::time_series_config::compression_threshold, config_, data_, and kcenon::monitoring::time_series_config::enable_compression.

Referenced by add_point(), and add_points().

Here is the caller graph for this function:

◆ create()

static common::Result< std::unique_ptr< time_series > > kcenon::monitoring::time_series::create	(	const std::string &	name,
		const time_series_config &	config = {} )

inlinestatic

Factory method to create time_series with validation.

Definition at line 291 of file time_series.h.

                                           {}) {
 
        auto validation = config.validate();
        if (validation.is_err()) {
            return common::Result<std::unique_ptr<time_series>>::err(
                error_info(monitoring_error_code::invalid_configuration,
                          validation.error().message, "monitoring_system").to_common_error());
        }
 
        return common::ok(std::unique_ptr<time_series>(new time_series(name, config)));
    }

Referenced by demonstrate_aggregations(), demonstrate_basic_operations(), demonstrate_batch_operations(), demonstrate_retention_and_downsampling(), kcenon::monitoring::metric_storage::get_or_create_series(), TEST_F(), and TEST_F().

Here is the caller graph for this function:

◆ empty()

bool kcenon::monitoring::time_series::empty ( ) const

inline

Check if series is empty.

Definition at line 446 of file time_series.h.

                       {
        std::lock_guard<std::mutex> lock(mutex_);
        return data_.empty();
    }

References data_, and mutex_.

◆ enforce_size_limit()

void kcenon::monitoring::time_series::enforce_size_limit ( )

inlineprivate

Ensure data size doesn't exceed maximum.

Definition at line 272 of file time_series.h.

                              {
        if (data_.size() > config_.max_points) {
            size_t remove_count = data_.size() - config_.max_points;
            data_.erase(data_.begin(), data_.begin() + remove_count);
        }
    }

References config_, data_, and kcenon::monitoring::time_series_config::max_points.

Referenced by add_point(), and add_points().

Here is the caller graph for this function:

◆ get_config()

const time_series_config & kcenon::monitoring::time_series::get_config ( ) const

inlinenoexcept

Get configuration.

Definition at line 461 of file time_series.h.

                                                          {
        return config_;
    }

References config_.

◆ get_latest_value()

common::Result< double > kcenon::monitoring::time_series::get_latest_value ( ) const

inline

Get latest value.

Definition at line 476 of file time_series.h.

                                                  {
        std::lock_guard<std::mutex> lock(mutex_);
 
        if (data_.empty()) {
            return common::Result<double>::err(
                error_info(monitoring_error_code::collection_failed,
                          "No data available", "monitoring_system").to_common_error());
        }
 
        return common::ok(data_.back().value);
    }

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

◆ memory_footprint()

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

inline

Get memory footprint in bytes.

Definition at line 491 of file time_series.h.

                                    {
        std::lock_guard<std::mutex> lock(mutex_);
        return sizeof(time_series) + 
               data_.capacity() * sizeof(time_point_data) +
               series_name_.capacity();
    }

References data_, mutex_, series_name_, and time_series().

Here is the call graph for this function:

◆ name()

const std::string & kcenon::monitoring::time_series::name ( ) const

inlinenoexcept

Get series name.

Definition at line 454 of file time_series.h.

                                           {
        return series_name_;
    }

References series_name_.

◆ query()

common::Result< aggregation_result > kcenon::monitoring::time_series::query ( const time_series_query & query ) const

inline

Query data for a time range.

Definition at line 370 of file time_series.h.

                                                                                 {
        auto validation = query.validate();
        if (validation.is_err()) {
            return common::Result<aggregation_result>::err(
                error_info(monitoring_error_code::invalid_argument,
                          validation.error().message, "monitoring_system").to_common_error());
        }
 
        std::lock_guard<std::mutex> lock(mutex_);
 
        aggregation_result result;
        result.query_start = query.start_time;
        result.query_end = query.end_time;
 
        // Find data points in the time range
        auto start_it = std::lower_bound(data_.begin(), data_.end(), query.start_time,
            [](const time_point_data& point, std::chrono::system_clock::time_point time) {
                return point.timestamp < time;
            });
 
        auto end_it = std::upper_bound(data_.begin(), data_.end(), query.end_time,
            [](std::chrono::system_clock::time_point time, const time_point_data& point) {
                return time < point.timestamp;
            });
 
        if (start_it == end_it) {
            return common::ok(std::move(result));  // No data in range
        }
 
        // Aggregate data by step size
        auto current_step_start = query.start_time;
 
        while (current_step_start < query.end_time) {
            auto current_step_end = current_step_start + query.step;
            if (current_step_end > query.end_time) {
                current_step_end = query.end_time;
            }
 
            // Find points in this step
            std::vector<time_point_data> step_points;
            for (auto it = start_it; it != end_it; ++it) {
                if (it->timestamp >= current_step_start && it->timestamp < current_step_end) {
                    step_points.push_back(*it);
                    result.total_samples += it->sample_count;
                }
            }
 
            // Aggregate points in this step
            if (!step_points.empty()) {
                time_point_data aggregated_point;
                aggregated_point.timestamp = current_step_start + query.step / 2;  // Middle of step
 
                for (const auto& point : step_points) {
                    aggregated_point.merge(point);
                }
 
                result.points.push_back(aggregated_point);
            }
 
            current_step_start = current_step_end;
        }
 
        return common::ok(std::move(result));
    }

References data_, kcenon::monitoring::invalid_argument, kcenon::monitoring::time_point_data::merge(), mutex_, kcenon::monitoring::aggregation_result::points, query(), kcenon::monitoring::aggregation_result::query_end, kcenon::monitoring::aggregation_result::query_start, kcenon::monitoring::time_point_data::timestamp, kcenon::monitoring::error_info::to_common_error(), and kcenon::monitoring::aggregation_result::total_samples.

Referenced by query().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ size()

size_t kcenon::monitoring::time_series::size ( ) const

inline

Get current number of data points.

Definition at line 438 of file time_series.h.

                        {
        std::lock_guard<std::mutex> lock(mutex_);
        return data_.size();
    }

References data_, and mutex_.

Member Data Documentation

◆ config_

time_series_config kcenon::monitoring::time_series::config_

private

Definition at line 213 of file time_series.h.

Referenced by cleanup_old_data(), compress_data(), enforce_size_limit(), get_config(), and time_series().

◆ data_

std::vector<time_point_data> kcenon::monitoring::time_series::data_

private

Definition at line 212 of file time_series.h.

Referenced by add_point(), add_points(), cleanup_old_data(), clear(), compress_data(), empty(), enforce_size_limit(), get_latest_value(), memory_footprint(), query(), size(), and time_series().

◆ insertion_count_

size_t kcenon::monitoring::time_series::insertion_count_ = 0

private

Definition at line 215 of file time_series.h.

Referenced by add_point().

◆ mutex_

std::mutex kcenon::monitoring::time_series::mutex_

mutableprivate

Definition at line 211 of file time_series.h.

Referenced by add_point(), add_points(), clear(), empty(), get_latest_value(), memory_footprint(), query(), and size().

◆ series_name_

std::string kcenon::monitoring::time_series::series_name_

private

Definition at line 214 of file time_series.h.

Referenced by memory_footprint(), and name().

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

include/kcenon/monitoring/utils/time_series.h

Public Member Functions

Static Public Member Functions

Private Member Functions

Private Attributes

Detailed Description

Constructor & Destructor Documentation

◆ time_series()

Member Function Documentation

◆ add_point()

◆ add_points()

◆ cleanup_old_data()

◆ clear()

◆ compress_data()

◆ create()

◆ empty()

◆ enforce_size_limit()

◆ get_config()

◆ get_latest_value()

◆ memory_footprint()

◆ name()

◆ query()

◆ size()

Member Data Documentation

◆ config_

◆ data_

◆ insertion_count_

◆ mutex_

◆ series_name_