alert_notifiers.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 <chrono>
#include <cstring>
#include <fstream>
#include <functional>
#include <memory>
#include <mutex>
#include <queue>
#include <sstream>
#include <string>
#include <thread>
#include <vector>
 
#include "alert_manager.h"
#include "alert_types.h"
#include "../core/result_types.h"
 
namespace kcenon::monitoring {
 
struct webhook_config {
    std::string url;                                        
    std::string method = "POST";                            
    std::chrono::milliseconds timeout{30000};               
    std::unordered_map<std::string, std::string> headers;   
    size_t max_retries = 3;                                 
    std::chrono::milliseconds retry_delay{1000};            
    bool send_resolved = true;                              
    std::string content_type = "application/json";          
 
    webhook_config& add_header(const std::string& key, const std::string& value) {
        headers[key] = value;
        return *this;
    }
 
    bool validate() const {
        return !url.empty() && timeout.count() > 0;
    }
};
 
class alert_formatter {
public:
    virtual ~alert_formatter() = default;
 
    virtual std::string format(const alert& a) const = 0;
 
    virtual std::string format_group(const alert_group& group) const = 0;
};
 
class json_alert_formatter : public alert_formatter {
public:
    std::string format(const alert& a) const override {
        std::ostringstream oss;
        oss << "{";
        oss << "\"name\":\"" << escape_json(a.name) << "\",";
        oss << "\"state\":\"" << alert_state_to_string(a.state) << "\",";
        oss << "\"severity\":\"" << alert_severity_to_string(a.severity) << "\",";
        oss << "\"value\":" << a.value << ",";
        oss << "\"summary\":\"" << escape_json(a.annotations.summary) << "\",";
        oss << "\"description\":\"" << escape_json(a.annotations.description) << "\",";
        oss << "\"fingerprint\":\"" << a.fingerprint() << "\",";
        oss << "\"labels\":{";
        bool first = true;
        for (const auto& [key, value] : a.labels.labels) {
            if (!first) oss << ",";
            oss << "\"" << escape_json(key) << "\":\"" << escape_json(value) << "\"";
            first = false;
        }
        oss << "}}";
        return oss.str();
    }
 
    std::string format_group(const alert_group& group) const override {
        std::ostringstream oss;
        oss << "{";
        oss << "\"group_key\":\"" << escape_json(group.group_key) << "\",";
        oss << "\"severity\":\"" << alert_severity_to_string(group.max_severity()) << "\",";
        oss << "\"alert_count\":" << group.size() << ",";
        oss << "\"alerts\":[";
        bool first = true;
        for (const auto& a : group.alerts) {
            if (!first) oss << ",";
            oss << format(a);
            first = false;
        }
        oss << "]}";
        return oss.str();
    }
 
private:
    static std::string escape_json(const std::string& s) {
        std::ostringstream oss;
        for (char c : s) {
            switch (c) {
                case '"':  oss << "\\\""; break;
                case '\\': oss << "\\\\"; break;
                case '\n': oss << "\\n";  break;
                case '\r': oss << "\\r";  break;
                case '\t': oss << "\\t";  break;
                default:   oss << c;      break;
            }
        }
        return oss.str();
    }
};
 
class text_alert_formatter : public alert_formatter {
public:
    std::string format(const alert& a) const override {
        std::ostringstream oss;
        oss << "[" << alert_state_to_string(a.state) << "] "
            << a.name << " (" << alert_severity_to_string(a.severity) << ")\n"
            << "  Summary: " << a.annotations.summary << "\n"
            << "  Value: " << a.value << "\n"
            << "  Fingerprint: " << a.fingerprint();
        return oss.str();
    }
 
    std::string format_group(const alert_group& group) const override {
        std::ostringstream oss;
        oss << "Alert Group: " << group.group_key << "\n"
            << "  Total alerts: " << group.size() << "\n"
            << "  Max severity: " << alert_severity_to_string(group.max_severity()) << "\n"
            << "  Alerts:\n";
        for (const auto& a : group.alerts) {
            oss << "    - " << a.name << " (" << alert_state_to_string(a.state) << ")\n";
        }
        return oss.str();
    }
};
 
class webhook_notifier : public alert_notifier {
public:
    using http_sender_func = std::function<common::VoidResult(
        const std::string& url,
        const std::string& method,
        const std::unordered_map<std::string, std::string>& headers,
        const std::string& body
    )>;
 
    explicit webhook_notifier(const webhook_config& config,
                              std::shared_ptr<alert_formatter> formatter = nullptr)
        : config_(config)
        , formatter_(formatter ? formatter : std::make_shared<json_alert_formatter>()) {}
 
    std::string name() const override {
        return "webhook:" + config_.url;
    }
 
    common::VoidResult notify(const alert& a) override {
        if (!config_.send_resolved && a.state == alert_state::resolved) {
            return common::ok();
        }
 
        std::string payload = formatter_->format(a);
        return send_with_retry(payload);
    }
 
    common::VoidResult notify_group(const alert_group& group) override {
        std::string payload = formatter_->format_group(group);
        return send_with_retry(payload);
    }
 
    bool is_ready() const override {
        return config_.validate() && http_sender_ != nullptr;
    }
 
    void set_http_sender(http_sender_func sender) {
        http_sender_ = std::move(sender);
    }
 
    const webhook_config& config() const { return config_; }
 
private:
    common::VoidResult send_with_retry(const std::string& payload) {
        if (!http_sender_) {
            return common::VoidResult::err(error_info(monitoring_error_code::operation_failed, "No HTTP sender configured").to_common_error());
        }
 
        auto headers = config_.headers;
        headers["Content-Type"] = config_.content_type;
 
        for (size_t attempt = 0; attempt <= config_.max_retries; ++attempt) {
            auto result = http_sender_(config_.url, config_.method, headers, payload);
            if (result.is_ok()) {
                return result;
            }
 
            if (attempt < config_.max_retries) {
                std::this_thread::sleep_for(config_.retry_delay);
            }
        }
 
        return common::VoidResult::err(static_cast<int>(monitoring_error_code::retry_attempts_exhausted),
                               "Failed to send webhook after " +
                               std::to_string(config_.max_retries) + " retries");
    }
 
    webhook_config config_;
    std::shared_ptr<alert_formatter> formatter_;
    http_sender_func http_sender_;
};
 
class file_notifier : public alert_notifier {
public:
    explicit file_notifier(std::string file_path,
                           std::shared_ptr<alert_formatter> formatter = nullptr)
        : file_path_(std::move(file_path))
        , formatter_(formatter ? formatter : std::make_shared<text_alert_formatter>()) {}
 
    std::string name() const override {
        return "file:" + file_path_;
    }
 
    common::VoidResult notify(const alert& a) override {
        return write_to_file(formatter_->format(a));
    }
 
    common::VoidResult notify_group(const alert_group& group) override {
        return write_to_file(formatter_->format_group(group));
    }
 
    bool is_ready() const override {
        return !file_path_.empty();
    }
 
private:
    common::VoidResult write_to_file(const std::string& content) {
        std::lock_guard<std::mutex> lock(mutex_);
 
        std::ofstream file(file_path_, std::ios::app);
        if (!file) {
            return common::VoidResult::err(static_cast<int>(monitoring_error_code::storage_write_failed),
                                   "Failed to open file: " + file_path_);
        }
 
        auto now = std::chrono::system_clock::now();
        auto time_t_now = std::chrono::system_clock::to_time_t(now);
 
        char time_buf[26];
#ifdef _MSC_VER
        ctime_s(time_buf, sizeof(time_buf), &time_t_now);
#else
        std::strncpy(time_buf, std::ctime(&time_t_now), sizeof(time_buf) - 1);
        time_buf[sizeof(time_buf) - 1] = '\0';
#endif
        file << "=== " << time_buf;
        file << content << "\n\n";
 
        return common::ok();
    }
 
    std::string file_path_;
    std::shared_ptr<alert_formatter> formatter_;
    std::mutex mutex_;
};
 
class multi_notifier : public alert_notifier {
public:
    explicit multi_notifier(std::string notifier_name)
        : name_(std::move(notifier_name)) {}
 
    void add_notifier(std::shared_ptr<alert_notifier> notifier) {
        std::lock_guard<std::mutex> lock(mutex_);
        notifiers_.push_back(std::move(notifier));
    }
 
    std::string name() const override { return name_; }
 
    common::VoidResult notify(const alert& a) override {
        std::lock_guard<std::mutex> lock(mutex_);
 
        std::vector<std::string> failures;
        for (const auto& notifier : notifiers_) {
            if (notifier && notifier->is_ready()) {
                auto result = notifier->notify(a);
                if (!result.is_ok()) {
                    failures.push_back(notifier->name());
                }
            }
        }
 
        if (!failures.empty()) {
            return common::VoidResult::err(static_cast<int>(monitoring_error_code::operation_failed),
                                   "Failed notifiers: " + join_strings(failures, ", "));
        }
        return common::ok();
    }
 
    common::VoidResult notify_group(const alert_group& group) override {
        std::lock_guard<std::mutex> lock(mutex_);
 
        std::vector<std::string> failures;
        for (const auto& notifier : notifiers_) {
            if (notifier && notifier->is_ready()) {
                auto result = notifier->notify_group(group);
                if (!result.is_ok()) {
                    failures.push_back(notifier->name());
                }
            }
        }
 
        if (!failures.empty()) {
            return common::VoidResult::err(static_cast<int>(monitoring_error_code::operation_failed),
                                   "Failed notifiers: " + join_strings(failures, ", "));
        }
        return common::ok();
    }
 
    bool is_ready() const override {
        std::lock_guard<std::mutex> lock(mutex_);
        return !notifiers_.empty();
    }
 
private:
    static std::string join_strings(const std::vector<std::string>& strings,
                                    const std::string& delimiter) {
        std::ostringstream oss;
        for (size_t i = 0; i < strings.size(); ++i) {
            if (i > 0) oss << delimiter;
            oss << strings[i];
        }
        return oss.str();
    }
 
    std::string name_;
    mutable std::mutex mutex_;
    std::vector<std::shared_ptr<alert_notifier>> notifiers_;
};
 
class buffered_notifier : public alert_notifier {
public:
    buffered_notifier(std::shared_ptr<alert_notifier> inner,
                      size_t buffer_size = 100,
                      std::chrono::milliseconds flush_interval = std::chrono::seconds(30))
        : inner_(std::move(inner))
        , buffer_size_(buffer_size)
        , flush_interval_(flush_interval)
        , last_flush_(std::chrono::steady_clock::now()) {}
 
    std::string name() const override {
        return "buffered:" + (inner_ ? inner_->name() : "none");
    }
 
    common::VoidResult notify(const alert& a) override {
        std::lock_guard<std::mutex> lock(mutex_);
 
        buffer_.push_back(a);
 
        if (should_flush()) {
            return flush_internal();
        }
 
        return common::ok();
    }
 
    common::VoidResult notify_group(const alert_group& group) override {
        std::lock_guard<std::mutex> lock(mutex_);
 
        for (const auto& a : group.alerts) {
            buffer_.push_back(a);
        }
 
        if (should_flush()) {
            return flush_internal();
        }
 
        return common::ok();
    }
 
    bool is_ready() const override {
        return inner_ && inner_->is_ready();
    }
 
    common::VoidResult flush() {
        std::lock_guard<std::mutex> lock(mutex_);
        return flush_internal();
    }
 
    size_t pending_count() const {
        std::lock_guard<std::mutex> lock(mutex_);
        return buffer_.size();
    }
 
private:
    bool should_flush() const {
        if (buffer_.size() >= buffer_size_) {
            return true;
        }
        auto now = std::chrono::steady_clock::now();
        return (now - last_flush_) >= flush_interval_;
    }
 
    common::VoidResult flush_internal() {
        if (buffer_.empty() || !inner_) {
            return common::ok();
        }
 
        // Create a group from buffered alerts
        alert_group group("buffered");
        for (auto& a : buffer_) {
            group.add_alert(std::move(a));
        }
        buffer_.clear();
 
        last_flush_ = std::chrono::steady_clock::now();
 
        return inner_->notify_group(group);
    }
 
    std::shared_ptr<alert_notifier> inner_;
    size_t buffer_size_;
    std::chrono::milliseconds flush_interval_;
 
    mutable std::mutex mutex_;
    std::vector<alert> buffer_;
    std::chrono::steady_clock::time_point last_flush_;
};
 
class routing_notifier : public alert_notifier {
public:
    using route_condition = std::function<bool(const alert&)>;
 
    explicit routing_notifier(std::string notifier_name)
        : name_(std::move(notifier_name)) {}
 
    void add_route(route_condition condition,
                   std::shared_ptr<alert_notifier> notifier) {
        std::lock_guard<std::mutex> lock(mutex_);
        routes_.push_back({std::move(condition), std::move(notifier)});
    }
 
    void set_default_route(std::shared_ptr<alert_notifier> notifier) {
        std::lock_guard<std::mutex> lock(mutex_);
        default_route_ = std::move(notifier);
    }
 
    std::string name() const override { return name_; }
 
    common::VoidResult notify(const alert& a) override {
        std::lock_guard<std::mutex> lock(mutex_);
 
        for (const auto& [condition, notifier] : routes_) {
            if (condition(a) && notifier && notifier->is_ready()) {
                return notifier->notify(a);
            }
        }
 
        if (default_route_ && default_route_->is_ready()) {
            return default_route_->notify(a);
        }
 
        return common::ok();
    }
 
    common::VoidResult notify_group(const alert_group& group) override {
        // Route each alert individually
        for (const auto& a : group.alerts) {
            auto result = notify(a);
            if (!result.is_ok()) {
                return result;
            }
        }
        return common::ok();
    }
 
    bool is_ready() const override {
        std::lock_guard<std::mutex> lock(mutex_);
        return !routes_.empty() || default_route_ != nullptr;
    }
 
    // Factory methods for common routes
    void route_by_severity(alert_severity severity,
                          std::shared_ptr<alert_notifier> notifier) {
        add_route(
            [severity](const alert& a) { return a.severity == severity; },
            std::move(notifier)
        );
    }
 
    void route_by_label(const std::string& key,
                        const std::string& value,
                        std::shared_ptr<alert_notifier> notifier) {
        add_route(
            [key, value](const alert& a) { return a.labels.get(key) == value; },
            std::move(notifier)
        );
    }
 
private:
    struct route {
        route_condition condition;
        std::shared_ptr<alert_notifier> notifier;
    };
 
    std::string name_;
    mutable std::mutex mutex_;
    std::vector<route> routes_;
    std::shared_ptr<alert_notifier> default_route_;
};
 
} // namespace kcenon::monitoring