plugin_metric_collector.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 <algorithm>
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <functional>
#include <map>
#include <memory>
#include <mutex>
#include <optional>
#include <queue>
#include <string>
#include <thread>
#include <typeindex>
#include <unordered_map>
#include <utility>
#include <vector>
 
#include "../interfaces/metric_collector_interface.h"
#include "../interfaces/metric_types_adapter.h"
#include "../interfaces/observer_interface.h"
#include "../utils/metric_types.h"
 
namespace kcenon { namespace monitoring {
 
class metric_collector_plugin {
  public:
    virtual ~metric_collector_plugin() = default;
 
    virtual bool initialize(const std::unordered_map<std::string, std::string>& config) = 0;
 
    virtual std::vector<metric> collect() = 0;
 
    virtual std::string get_name() const = 0;
 
    virtual std::vector<std::string> get_metric_types() const = 0;
 
    virtual bool is_healthy() const = 0;
 
    virtual std::unordered_map<std::string, double> get_statistics() const = 0;
};
 
struct plugin_collector_config {
    // Collection interval in milliseconds
    std::chrono::milliseconds collection_interval{1000};
 
    // Maximum batch size for metric collection
    size_t max_batch_size{1000};
 
    // Enable caching of metrics
    bool enable_caching{true};
 
    // Cache TTL in seconds
    std::chrono::seconds cache_ttl{60};
 
    // Enable real-time streaming
    bool enable_streaming{false};
 
    // Number of worker threads for collection
    size_t worker_threads{2};
 
    // Enable metric aggregation
    bool enable_aggregation{true};
 
    // Aggregation window in seconds
    std::chrono::seconds aggregation_window{10};
};
 
class plugin_metric_collector : public interface_metric_collector {
  public:
    explicit plugin_metric_collector(const plugin_collector_config& config = {});
    ~plugin_metric_collector() override;
 
    // Disable copy
    plugin_metric_collector(const plugin_metric_collector&) = delete;
    plugin_metric_collector& operator=(const plugin_metric_collector&) = delete;
 
    // Enable move
    plugin_metric_collector(plugin_metric_collector&&) = delete;
    plugin_metric_collector& operator=(plugin_metric_collector&&) = delete;
 
    bool register_plugin(std::unique_ptr<metric_collector_plugin> plugin);
 
    bool unregister_plugin(const std::string& plugin_name);
 
    std::vector<std::string> get_registered_plugins() const;
 
    bool start();
 
    void stop();
 
    bool is_running() const;
 
    // Missing methods from interface
    common::VoidResult start_collection(const collection_config& config) override;
    common::VoidResult stop_collection() override;
    bool is_collecting() const override;
    collection_config get_config() const override;
    metric_stats get_stats() const override;
    void reset_stats() override;
 
    std::vector<metric> get_cached_metrics(const std::optional<std::string>& plugin_name = std::nullopt) const;
 
    std::vector<metric_stats> get_aggregated_metrics(size_t window_seconds = 60) const;
 
    common::Result<std::vector<metric>> force_collect() override;
 
    // interface_metric_collector implementation
    common::Result<std::vector<metric>> collect_metrics() override;
    common::VoidResult register_observer(std::shared_ptr<interface_monitoring_observer> observer) override;
    common::VoidResult unregister_observer(std::shared_ptr<interface_monitoring_observer> observer) override;
    std::vector<std::string> get_metric_types() const override;
    common::VoidResult update_config(const collection_config& config) override;
 
  private:
    // Plugin management
    mutable std::mutex plugins_mutex_;
    std::unordered_map<std::string, std::unique_ptr<metric_collector_plugin>> plugins_;
 
    // Observer management
    mutable std::mutex observers_mutex_;
    std::vector<std::weak_ptr<interface_monitoring_observer>> observers_;
 
    // Metric cache
    struct cached_metric {
        metric data;
        std::chrono::steady_clock::time_point timestamp;
        std::string plugin_name;
    };
    mutable std::mutex cache_mutex_;
    std::vector<cached_metric> metric_cache_;
 
    // Collection state
    std::atomic<bool> running_{false};
    std::vector<std::thread> worker_threads_;
    std::condition_variable work_cv_;
    mutable std::mutex work_mutex_;
 
    // Configuration
    plugin_collector_config config_;
 
    // Statistics
    std::atomic<size_t> total_metrics_collected_{0};
    std::atomic<size_t> collection_errors_{0};
    std::chrono::steady_clock::time_point start_time_;
 
    // Internal methods
    void collection_worker();
    void collect_from_plugin(const std::string& name, metric_collector_plugin* plugin);
    void notify_observers(const metric& m);
    void cleanup_cache();
    void aggregate_metrics();
};
 
class plugin_factory {
  public:
    static std::unique_ptr<metric_collector_plugin> create_system_resource_collector(
        const std::unordered_map<std::string, std::string>& config = {});
 
    static std::unique_ptr<metric_collector_plugin> create_thread_system_collector(
        const std::unordered_map<std::string, std::string>& config = {});
 
    static std::unique_ptr<metric_collector_plugin> create_logger_system_collector(
        const std::unordered_map<std::string, std::string>& config = {});
 
    static void register_factory(
        const std::string& type,
        std::function<std::unique_ptr<metric_collector_plugin>(const std::unordered_map<std::string, std::string>&)>
            factory);
 
    static std::unique_ptr<metric_collector_plugin> create(const std::string& type,
                                                             const std::unordered_map<std::string, std::string>& config = {});
 
  private:
    static std::unordered_map<
        std::string, std::function<std::unique_ptr<metric_collector_plugin>(const std::unordered_map<std::string, std::string>&)>>
        factories_;
    static std::mutex factories_mutex_;
};
 
} } // namespace kcenon::monitoring