realtime_log_analyzer.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 <kcenon/common/interfaces/logger_interface.h>
#include <kcenon/logger/analysis/log_analyzer.h>
 
#include <atomic>
#include <chrono>
#include <cstdint>
#include <deque>
#include <functional>
#include <memory>
#include <mutex>
#include <regex>
#include <shared_mutex>
#include <string>
#include <unordered_set>
#include <vector>
 
namespace kcenon::logger::analysis {
 
// Type alias for log_level
using log_level = common::interfaces::log_level;
 
struct anomaly_event {
    enum class type : std::uint8_t {
        error_spike,      
        pattern_match,    
        rate_anomaly,     
        new_error_type    
    };
 
    type anomaly_type = type::error_spike;                   
    std::chrono::system_clock::time_point detected_at;       
    std::string description;                                 
    std::vector<analyzed_log_entry> related_entries;         
    std::string pattern;                                     
    size_t current_count = 0;                                
    size_t threshold = 0;                                    
};
 
struct realtime_analysis_config {
    size_t error_spike_threshold = 100;                      
    size_t rate_anomaly_high_threshold = 1000;               
    size_t rate_anomaly_low_threshold = 0;                   
    std::chrono::seconds window_duration{60};                
    std::chrono::seconds baseline_duration{300};             
    bool track_new_errors = true;                            
    bool enable_rate_anomaly_detection = true;               
    double rate_deviation_factor = 2.0;                      
    size_t max_related_entries = 10;                         
};
 
struct pattern_alert {
    std::string pattern;                                     
    log_level min_level;                                     
    std::regex compiled_pattern;                             
 
    pattern_alert(const std::string& p, log_level level)
        : pattern(p), min_level(level), compiled_pattern(p, std::regex::optimize) {}
};
 
class realtime_log_analyzer {
public:
    using anomaly_callback = std::function<void(const anomaly_event&)>;
 
    realtime_log_analyzer() = default;
 
    explicit realtime_log_analyzer(const realtime_analysis_config& config)
        : config_(config) {}
 
    void set_anomaly_callback(anomaly_callback cb) {
        std::unique_lock lock(callback_mutex_);
        callback_ = std::move(cb);
    }
 
    void analyze(const analyzed_log_entry& entry) {
        auto now = std::chrono::system_clock::now();
 
        // Add to sliding window
        add_to_window(entry, now);
 
        // Check for error spike
        if (entry.level == log_level::error ||
            entry.level == log_level::fatal) {
            check_error_spike(entry, now);
        }
 
        // Check pattern alerts
        check_pattern_alerts(entry, now);
 
        // Check rate anomaly
        if (config_.enable_rate_anomaly_detection) {
            check_rate_anomaly(now);
        }
 
        // Track new error types
        if (config_.track_new_errors &&
            (entry.level == log_level::error ||
             entry.level == log_level::fatal)) {
            check_new_error_type(entry, now);
        }
    }
 
    void set_error_spike_threshold(size_t errors_per_minute) {
        config_.error_spike_threshold = errors_per_minute;
    }
 
    void add_pattern_alert(const std::string& pattern, log_level min_level) {
        std::unique_lock lock(patterns_mutex_);
        patterns_.emplace_back(pattern, min_level);
    }
 
    bool remove_pattern_alert(const std::string& pattern) {
        std::unique_lock lock(patterns_mutex_);
        auto it = std::remove_if(patterns_.begin(), patterns_.end(),
            [&pattern](const pattern_alert& alert) {
                return alert.pattern == pattern;
            });
        if (it != patterns_.end()) {
            patterns_.erase(it, patterns_.end());
            return true;
        }
        return false;
    }
 
    void clear_pattern_alerts() {
        std::unique_lock lock(patterns_mutex_);
        patterns_.clear();
    }
 
    void set_rate_thresholds(size_t high_threshold, size_t low_threshold = 0) {
        config_.rate_anomaly_high_threshold = high_threshold;
        config_.rate_anomaly_low_threshold = low_threshold;
    }
 
    void set_track_new_errors(bool enable) {
        config_.track_new_errors = enable;
    }
 
    double get_error_rate() const {
        std::shared_lock lock(window_mutex_);
        return calculate_rate(error_window_);
    }
 
    double get_log_rate() const {
        std::shared_lock lock(window_mutex_);
        return calculate_rate(log_window_);
    }
 
    struct statistics {
        size_t total_analyzed = 0;
        size_t total_errors = 0;
        size_t anomalies_detected = 0;
        size_t error_spikes = 0;
        size_t pattern_matches = 0;
        size_t rate_anomalies = 0;
        size_t new_error_types = 0;
        double current_log_rate = 0.0;
        double current_error_rate = 0.0;
    };
 
    statistics get_statistics() const {
        std::shared_lock lock(stats_mutex_);
        statistics stats;
        stats.total_analyzed = total_analyzed_.load();
        stats.total_errors = total_errors_.load();
        stats.anomalies_detected = anomalies_detected_.load();
        stats.error_spikes = error_spikes_.load();
        stats.pattern_matches = pattern_matches_.load();
        stats.rate_anomalies = rate_anomalies_.load();
        stats.new_error_types = new_error_types_.load();
        stats.current_log_rate = get_log_rate();
        stats.current_error_rate = get_error_rate();
        return stats;
    }
 
    void reset() {
        {
            std::unique_lock lock(window_mutex_);
            log_window_.clear();
            error_window_.clear();
            baseline_rates_.clear();
        }
        {
            std::unique_lock lock(errors_mutex_);
            known_errors_.clear();
        }
        {
            std::unique_lock lock(stats_mutex_);
            total_analyzed_ = 0;
            total_errors_ = 0;
            anomalies_detected_ = 0;
            error_spikes_ = 0;
            pattern_matches_ = 0;
            rate_anomalies_ = 0;
            new_error_types_ = 0;
        }
        last_rate_check_ = std::chrono::system_clock::time_point{};
        last_spike_alert_ = std::chrono::system_clock::time_point{};
    }
 
    const realtime_analysis_config& get_config() const {
        return config_;
    }
 
    void set_config(const realtime_analysis_config& config) {
        config_ = config;
    }
 
private:
    struct timestamped_entry {
        std::chrono::system_clock::time_point timestamp;
        analyzed_log_entry entry;
    };
 
    void add_to_window(const analyzed_log_entry& entry,
                       std::chrono::system_clock::time_point now) {
        std::unique_lock lock(window_mutex_);
 
        // Add to log window
        log_window_.push_back({now, entry});
 
        // Add to error window if error/fatal
        if (entry.level == log_level::error ||
            entry.level == log_level::fatal) {
            error_window_.push_back({now, entry});
        }
 
        // Clean up old entries
        auto cutoff = now - config_.window_duration;
        cleanup_window(log_window_, cutoff);
        cleanup_window(error_window_, cutoff);
 
        // Update statistics
        total_analyzed_.fetch_add(1, std::memory_order_relaxed);
        if (entry.level == log_level::error ||
            entry.level == log_level::fatal) {
            total_errors_.fetch_add(1, std::memory_order_relaxed);
        }
    }
 
    static void cleanup_window(std::deque<timestamped_entry>& window,
                               std::chrono::system_clock::time_point cutoff) {
        while (!window.empty() && window.front().timestamp < cutoff) {
            window.pop_front();
        }
    }
 
    double calculate_rate(const std::deque<timestamped_entry>& window) const {
        if (window.empty()) return 0.0;
 
        auto duration = std::chrono::duration_cast<std::chrono::seconds>(
            config_.window_duration).count();
        if (duration == 0) duration = 60;
 
        return static_cast<double>(window.size()) * 60.0 / duration;
    }
 
    void check_error_spike(const analyzed_log_entry& entry,
                          std::chrono::system_clock::time_point now) {
        std::shared_lock lock(window_mutex_);
 
        double current_rate = calculate_rate(error_window_);
 
        if (current_rate >= static_cast<double>(config_.error_spike_threshold)) {
            // Rate limit: don't alert more than once per minute
            {
                std::shared_lock rate_lock(rate_limit_mutex_);
                if (now - last_spike_alert_ < std::chrono::minutes(1)) {
                    return;
                }
            }
 
            lock.unlock();  // Release before callback
 
            // Update last alert time
            {
                std::unique_lock rate_lock(rate_limit_mutex_);
                last_spike_alert_ = now;
            }
 
            anomaly_event event;
            event.anomaly_type = anomaly_event::type::error_spike;
            event.detected_at = now;
            event.description = "Error spike detected: " +
                std::to_string(static_cast<size_t>(current_rate)) +
                " errors/minute (threshold: " +
                std::to_string(config_.error_spike_threshold) + ")";
            event.current_count = static_cast<size_t>(current_rate);
            event.threshold = config_.error_spike_threshold;
 
            // Collect related entries
            {
                std::shared_lock entries_lock(window_mutex_);
                collect_related_entries(event, error_window_);
            }
 
            notify_anomaly(event);
            error_spikes_.fetch_add(1, std::memory_order_relaxed);
        }
    }
 
    void check_pattern_alerts(const analyzed_log_entry& entry,
                             std::chrono::system_clock::time_point now) {
        std::shared_lock lock(patterns_mutex_);
 
        for (const auto& alert : patterns_) {
            // Check level threshold
            if (static_cast<int>(entry.level) < static_cast<int>(alert.min_level)) {
                continue;
            }
 
            // Check pattern match
            if (std::regex_search(entry.message, alert.compiled_pattern)) {
                lock.unlock();  // Release before callback
 
                anomaly_event event;
                event.anomaly_type = anomaly_event::type::pattern_match;
                event.detected_at = now;
                event.pattern = alert.pattern;
                event.description = "Pattern '" + alert.pattern +
                    "' matched in log message: " + entry.message;
                event.related_entries.push_back(entry);
 
                notify_anomaly(event);
                pattern_matches_.fetch_add(1, std::memory_order_relaxed);
                return;  // Only report first match per entry
            }
        }
    }
 
    void check_rate_anomaly(std::chrono::system_clock::time_point now) {
        // Rate limit rate anomaly checks to once per 10 seconds
        {
            std::shared_lock rate_lock(rate_limit_mutex_);
            if (now - last_rate_check_ < std::chrono::seconds(10)) {
                return;
            }
        }
        {
            std::unique_lock rate_lock(rate_limit_mutex_);
            last_rate_check_ = now;
        }
 
        std::shared_lock lock(window_mutex_);
        double current_rate = calculate_rate(log_window_);
        lock.unlock();
 
        // Check high rate
        if (current_rate >= static_cast<double>(config_.rate_anomaly_high_threshold)) {
            anomaly_event event;
            event.anomaly_type = anomaly_event::type::rate_anomaly;
            event.detected_at = now;
            event.description = "High log rate detected: " +
                std::to_string(static_cast<size_t>(current_rate)) +
                " logs/minute (threshold: " +
                std::to_string(config_.rate_anomaly_high_threshold) + ")";
            event.current_count = static_cast<size_t>(current_rate);
            event.threshold = config_.rate_anomaly_high_threshold;
 
            notify_anomaly(event);
            rate_anomalies_.fetch_add(1, std::memory_order_relaxed);
        }
        // Check low rate (if enabled)
        else if (config_.rate_anomaly_low_threshold > 0 &&
                 current_rate < static_cast<double>(config_.rate_anomaly_low_threshold) &&
                 total_analyzed_.load() > 100) {  // Only after enough data
            anomaly_event event;
            event.anomaly_type = anomaly_event::type::rate_anomaly;
            event.detected_at = now;
            event.description = "Low log rate detected: " +
                std::to_string(static_cast<size_t>(current_rate)) +
                " logs/minute (threshold: " +
                std::to_string(config_.rate_anomaly_low_threshold) + ")";
            event.current_count = static_cast<size_t>(current_rate);
            event.threshold = config_.rate_anomaly_low_threshold;
 
            notify_anomaly(event);
            rate_anomalies_.fetch_add(1, std::memory_order_relaxed);
        }
    }
 
    void check_new_error_type(const analyzed_log_entry& entry,
                             std::chrono::system_clock::time_point now) {
        // Normalize error message (remove numbers, timestamps, etc.)
        std::string normalized = normalize_error_message(entry.message);
 
        {
            std::shared_lock lock(errors_mutex_);
            if (known_errors_.contains(normalized)) {
                return;  // Already seen this error type
            }
        }
 
        // New error type detected
        {
            std::unique_lock lock(errors_mutex_);
            known_errors_.insert(normalized);
        }
 
        anomaly_event event;
        event.anomaly_type = anomaly_event::type::new_error_type;
        event.detected_at = now;
        event.description = "New error type detected: " + entry.message;
        event.related_entries.push_back(entry);
 
        notify_anomaly(event);
        new_error_types_.fetch_add(1, std::memory_order_relaxed);
    }
 
    static std::string normalize_error_message(const std::string& message) {
        // Remove numbers (IDs, timestamps, etc.)
        std::string normalized = std::regex_replace(message,
            std::regex(R"(\d+)"), "N");
 
        // Remove hex values
        normalized = std::regex_replace(normalized,
            std::regex(R"(0x[0-9a-fA-F]+)"), "HEX");
 
        // Remove UUIDs
        normalized = std::regex_replace(normalized,
            std::regex(R"([0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12})"),
            "UUID");
 
        return normalized;
    }
 
    void collect_related_entries(anomaly_event& event,
                                const std::deque<timestamped_entry>& window) const {
        size_t count = 0;
        for (auto it = window.rbegin();
             it != window.rend() && count < config_.max_related_entries;
             ++it, ++count) {
            event.related_entries.push_back(it->entry);
        }
    }
 
    void notify_anomaly(const anomaly_event& event) {
        anomalies_detected_.fetch_add(1, std::memory_order_relaxed);
 
        std::shared_lock lock(callback_mutex_);
        if (callback_) {
            callback_(event);
        }
    }
 
    // Configuration
    realtime_analysis_config config_;
 
    // Callback
    anomaly_callback callback_;
    mutable std::shared_mutex callback_mutex_;
 
    // Sliding windows
    std::deque<timestamped_entry> log_window_;
    std::deque<timestamped_entry> error_window_;
    std::deque<double> baseline_rates_;
    mutable std::shared_mutex window_mutex_;
 
    // Pattern alerts
    std::vector<pattern_alert> patterns_;
    mutable std::shared_mutex patterns_mutex_;
 
    // Known error types for new error detection
    std::unordered_set<std::string> known_errors_;
    mutable std::shared_mutex errors_mutex_;
 
    // Rate limiting
    std::chrono::system_clock::time_point last_rate_check_;
    std::chrono::system_clock::time_point last_spike_alert_;
    mutable std::shared_mutex rate_limit_mutex_;
 
    // Statistics (atomic for lock-free reads)
    std::atomic<size_t> total_analyzed_{0};
    std::atomic<size_t> total_errors_{0};
    std::atomic<size_t> anomalies_detected_{0};
    std::atomic<size_t> error_spikes_{0};
    std::atomic<size_t> pattern_matches_{0};
    std::atomic<size_t> rate_anomalies_{0};
    std::atomic<size_t> new_error_types_{0};
    mutable std::shared_mutex stats_mutex_;
};
 
class realtime_analyzer_factory {
public:
    static std::unique_ptr<realtime_log_analyzer> create_basic() {
        return std::make_unique<realtime_log_analyzer>();
    }
 
    static std::unique_ptr<realtime_log_analyzer> create(
        const realtime_analysis_config& config) {
        return std::make_unique<realtime_log_analyzer>(config);
    }
 
    static std::unique_ptr<realtime_log_analyzer> create_production(
        size_t error_threshold = 50,
        realtime_log_analyzer::anomaly_callback callback = nullptr) {
 
        realtime_analysis_config config;
        config.error_spike_threshold = error_threshold;
        config.rate_anomaly_high_threshold = 1000;
        config.rate_anomaly_low_threshold = 10;
        config.track_new_errors = true;
        config.enable_rate_anomaly_detection = true;
 
        auto analyzer = std::make_unique<realtime_log_analyzer>(config);
        if (callback) {
            analyzer->set_anomaly_callback(std::move(callback));
        }
        return analyzer;
    }
};
 
} // namespace kcenon::logger::analysis