network_writer.cpp

Go to the documentation of this file.

// BSD 3-Clause License
// Copyright (c) 2025, 🍀☀🌕🌥 🌊
// See the LICENSE file in the project root for full license information.
 
#include <kcenon/logger/writers/network_writer.h>
#include <kcenon/logger/security/integrity_policy.h>
#include <kcenon/logger/utils/error_handling_utils.h>
#include <kcenon/logger/utils/string_utils.h>
#include "../async/jthread_compat.h"
 
#ifdef _WIN32
    #include <winsock2.h>
    #include <ws2tcpip.h>
    #pragma comment(lib, "ws2_32.lib")
    typedef int socklen_t;
    typedef SSIZE_T ssize_t;  // Define ssize_t for Windows
    #define close closesocket
#else
    #include <sys/socket.h>
    #include <netinet/in.h>
    #include <arpa/inet.h>
    #include <netdb.h>
    #include <unistd.h>
#endif
 
#include <cerrno>
#include <cstring>
#include <functional>
#include <iomanip>
#include <sstream>
#include <thread>
 
namespace kcenon::logger {
 
using namespace async;
 
class network_send_jthread_worker {
public:
    using send_callback = std::function<void()>;
 
    explicit network_send_jthread_worker(send_callback callback)
        : callback_(std::move(callback))
    {}
 
    ~network_send_jthread_worker() {
        stop();
    }
 
    void start() {
        if (running_.exchange(true, std::memory_order_acq_rel)) {
            return;  // Already started
        }
 
#if LOGGER_HAS_JTHREAD
        auto callback = callback_;
        thread_ = compat_jthread([callback](std::stop_token stop_token) {
            while (!stop_token.stop_requested()) {
                if (callback) {
                    callback();
                }
                std::this_thread::sleep_for(std::chrono::milliseconds(10));
            }
        });
#else
        // Create worker thread with manual stop source
        // Note: We use the stop_source created by compat_jthread to ensure
        // request_stop() correctly signals the worker loop
        auto callback = callback_;
        thread_ = compat_jthread([callback](simple_stop_source& stop) {
            while (!stop.stop_requested()) {
                if (callback) {
                    callback();
                }
                std::this_thread::sleep_for(std::chrono::milliseconds(10));
            }
        });
#endif
    }
 
    void stop() {
        if (!running_.exchange(false, std::memory_order_acq_rel)) {
            return;  // Already stopped
        }
 
        // Request stop and join thread
        thread_.request_stop();
        thread_.join();
    }
 
    void notify_work() {
        has_work_.store(true, std::memory_order_release);
    }
 
private:
    send_callback callback_;
    compat_jthread thread_;
    std::atomic<bool> running_{false};
    std::atomic<bool> has_work_{false};
};
 
class network_reconnect_jthread_worker {
public:
    using reconnect_callback = std::function<void()>;
 
    explicit network_reconnect_jthread_worker(reconnect_callback callback,
                                              std::chrono::seconds interval)
        : callback_(std::move(callback))
        , reconnect_interval_(interval)
    {}
 
    ~network_reconnect_jthread_worker() {
        stop();
    }
 
    void start() {
        if (running_.exchange(true, std::memory_order_acq_rel)) {
            return;  // Already started
        }
 
#if LOGGER_HAS_JTHREAD
        auto callback = callback_;
        auto interval = reconnect_interval_;
        thread_ = compat_jthread([callback, interval](std::stop_token stop_token) {
            while (!stop_token.stop_requested()) {
                if (callback) {
                    callback();
                }
                // Sleep with stop checking
                for (auto elapsed = std::chrono::milliseconds{0};
                     elapsed < std::chrono::duration_cast<std::chrono::milliseconds>(interval)
                        && !stop_token.stop_requested();
                     elapsed += std::chrono::milliseconds(100)) {
                    std::this_thread::sleep_for(std::chrono::milliseconds(100));
                }
            }
        });
#else
        // Create worker thread with manual stop source
        // Note: We use the stop_source created by compat_jthread to ensure
        // request_stop() correctly signals the worker loop
        auto callback = callback_;
        auto interval = reconnect_interval_;
        thread_ = compat_jthread([callback, interval](simple_stop_source& stop) {
            while (!stop.stop_requested()) {
                if (callback) {
                    callback();
                }
                // Sleep with stop checking
                for (auto elapsed = std::chrono::milliseconds{0};
                     elapsed < std::chrono::duration_cast<std::chrono::milliseconds>(interval)
                        && !stop.stop_requested();
                     elapsed += std::chrono::milliseconds(100)) {
                    std::this_thread::sleep_for(std::chrono::milliseconds(100));
                }
            }
        });
#endif
    }
 
    void stop() {
        if (!running_.exchange(false, std::memory_order_acq_rel)) {
            return;  // Already stopped
        }
 
        // Request stop and join thread
        thread_.request_stop();
        thread_.join();
    }
 
private:
    reconnect_callback callback_;
    std::chrono::seconds reconnect_interval_;
    compat_jthread thread_;
    std::atomic<bool> running_{false};
};
 
network_writer::network_writer(const std::string& host,
                               uint16_t port,
                               protocol_type protocol,
                               size_t buffer_size,
                               std::chrono::seconds reconnect_interval)
    : host_(host)
    , port_(port)
    , protocol_(protocol)
    , buffer_size_(buffer_size)
    , reconnect_interval_(reconnect_interval)
    , socket_fd_(-1) {
 
#ifdef _WIN32
    // Initialize Winsock
    WSADATA wsaData;
    if (WSAStartup(MAKEWORD(2, 2), &wsaData) != 0) {
        throw std::runtime_error("Failed to initialize Winsock");
    }
#endif
 
    running_ = true;
 
    // Create and start send worker
    send_worker_ = std::make_unique<network_send_jthread_worker>(
        [this] { process_buffer(); });
    send_worker_->start();
 
    // Create and start reconnect worker for TCP
    if (protocol_ == protocol_type::tcp) {
        reconnect_worker_ = std::make_unique<network_reconnect_jthread_worker>(
            [this] { attempt_reconnect(); }, reconnect_interval_);
        reconnect_worker_->start();
    }
 
    // Initial connection attempt
    connect();
}
 
network_writer::~network_writer() {
    running_ = false;
 
    // Stop workers with error handling
    // IMPORTANT: Reset workers after stop to join their threads before accessing any members
    utils::safe_destructor_operation("send_worker_stop", [this]() {
        if (send_worker_) {
            send_worker_->stop();
            send_worker_.reset();
        }
    });
 
    utils::safe_destructor_operation("reconnect_worker_stop", [this]() {
        if (reconnect_worker_) {
            reconnect_worker_->stop();
            reconnect_worker_.reset();
        }
    });
 
    // Disconnect with error handling
    utils::safe_destructor_operation("network_disconnect", [this]() {
        disconnect();
    });
 
#ifdef _WIN32
    // Cleanup Winsock with error handling
    utils::safe_destructor_operation("winsock_cleanup", []() {
        WSACleanup();
    });
#endif
}
 
common::VoidResult network_writer::write(const log_entry& entry) {
 
    std::lock_guard<std::mutex> lock(buffer_mutex_);
 
    // Check buffer size
    if (buffer_.size() >= buffer_size_) {
        // Drop oldest message
        buffer_.pop();
        std::lock_guard<std::mutex> stats_lock(stats_mutex_);
        stats_.send_failures++;
        // Note: We still accept the new message after dropping the oldest
    }
 
    // Create a copy of the entry since log_entry is move-only
    if (entry.location) {
        buffer_.emplace(entry.level,
                       entry.message.to_string(),
                       entry.location->file.to_string(),
                       entry.location->line,
                       entry.location->function.to_string(),
                       entry.timestamp);
    } else {
        buffer_.emplace(entry.level,
                       entry.message.to_string(),
                       entry.timestamp);
    }
 
    // Notify send worker
    if (send_worker_) {
        send_worker_->notify_work();
    }
 
    return common::ok();
}
 
common::VoidResult network_writer::flush() {
    std::unique_lock<std::mutex> lock(buffer_mutex_);
    auto start = std::chrono::steady_clock::now();
    auto timeout = std::chrono::seconds(5); // 5 second timeout
 
    while (!buffer_.empty()) {
        if (buffer_cv_.wait_for(lock, timeout, [this] { return buffer_.empty() || !running_; })) {
            if (!buffer_.empty() && !running_) {
                return make_logger_void_result(logger_error_code::flush_timeout,
                                               "Network writer stopped before flush completed");
            }
        } else {
            return make_logger_void_result(logger_error_code::flush_timeout,
                                           "Network flush timeout");
        }
 
        if (std::chrono::steady_clock::now() - start > timeout) {
            return make_logger_void_result(logger_error_code::flush_timeout,
                                           "Network flush exceeded timeout");
        }
    }
    return common::ok();
}
 
network_writer::connection_stats network_writer::get_stats() const {
    std::lock_guard<std::mutex> lock(stats_mutex_);
    return stats_;
}
 
bool network_writer::connect() {
    if (connected_) {
        return true;
    }
 
    // Resolve hostname using getaddrinfo (thread-safe, IPv4/IPv6)
    int sock_type = (protocol_ == protocol_type::tcp) ? SOCK_STREAM : SOCK_DGRAM;
 
    struct addrinfo hints{}, *result = nullptr;
    hints.ai_family = AF_UNSPEC;
    hints.ai_socktype = sock_type;
 
    auto port_str = std::to_string(port_);
    int rv = getaddrinfo(host_.c_str(), port_str.c_str(), &hints, &result);
    if (rv != 0) {
        std::lock_guard<std::mutex> lock(stats_mutex_);
        stats_.connection_failures++;
        stats_.last_error = std::chrono::system_clock::now();
        return false;
    }
 
    // Try each resolved address until one succeeds
    for (auto* rp = result; rp != nullptr; rp = rp->ai_next) {
        socket_fd_ = socket(rp->ai_family, rp->ai_socktype, rp->ai_protocol);
        if (socket_fd_ < 0) {
            continue;
        }
 
        if (::connect(socket_fd_, rp->ai_addr, static_cast<socklen_t>(rp->ai_addrlen)) == 0) {
            freeaddrinfo(result);
            connected_ = true;
 
            std::lock_guard<std::mutex> lock(stats_mutex_);
            stats_.last_connected = std::chrono::system_clock::now();
            return true;
        }
 
        ::close(socket_fd_);
        socket_fd_ = -1;
    }
 
    freeaddrinfo(result);
 
    std::lock_guard<std::mutex> lock(stats_mutex_);
    stats_.connection_failures++;
    stats_.last_error = std::chrono::system_clock::now();
    return false;
}
 
void network_writer::disconnect() {
    if (socket_fd_ >= 0) {
        ::close(socket_fd_);
        socket_fd_ = -1;
    }
    connected_ = false;
}
 
bool network_writer::send_data(const std::string& data) {
    if (!connected_ || socket_fd_ < 0) {
        return false;
    }
 
#ifdef _WIN32
    int sent = ::send(socket_fd_, data.c_str(), static_cast<int>(data.length()), 0);
#else
    ssize_t sent = ::send(socket_fd_, data.c_str(), data.length(), 0);
#endif
    if (sent < 0) {
        if (protocol_ == protocol_type::tcp) {
            // TCP connection lost
            disconnect();
 
            std::lock_guard<std::mutex> lock(stats_mutex_);
            stats_.send_failures++;
            stats_.last_error = std::chrono::system_clock::now();
        }
        return false;
    }
 
    std::lock_guard<std::mutex> lock(stats_mutex_);
    stats_.messages_sent++;
    stats_.bytes_sent += sent;
 
    return true;
}
 
void network_writer::process_buffer() {
    std::unique_lock<std::mutex> lock(buffer_mutex_);
 
    // Process buffered logs
    while (!buffer_.empty() && running_) {
        auto entry = std::move(buffer_.front());
        buffer_.pop();
        lock.unlock();
 
        // Format and send
        std::string formatted = format_for_network(entry);
        send_data(formatted);
 
        lock.lock();
    }
}
 
void network_writer::attempt_reconnect() {
    if (!connected_ && running_) {
        connect();
    }
}
 
std::string network_writer::format_for_network(const log_entry& entry) {
    // Format as JSON for network transmission
    std::ostringstream oss;
    oss << "{";
 
    // Timestamp
    auto time_t = std::chrono::system_clock::to_time_t(entry.timestamp);
    oss << "\"@timestamp\":\"";
    oss << std::put_time(std::gmtime(&time_t), "%Y-%m-%dT%H:%M:%SZ") << "\",";
 
    // Level - convert logger_system::log_level to common::interfaces::log_level
    auto level = static_cast<common::interfaces::log_level>(static_cast<int>(entry.level));
    oss << "\"level\":\"" << utils::string_utils::level_to_string(level) << "\",";
 
    // Message
    oss << "\"message\":\"" << escape_json(entry.message.to_string()) << "\"";
 
    // Optional fields from source_location
    if (entry.location) {
        std::string file = entry.location->file.to_string();
        std::string function = entry.location->function.to_string();
 
        if (!file.empty()) {
            oss << ",\"file\":\"" << escape_json(file) << "\"";
            oss << ",\"line\":" << entry.location->line;
        }
 
        if (!function.empty()) {
            oss << ",\"function\":\"" << escape_json(function) << "\"";
        }
    }
 
    // Add hostname
    char hostname[256];
    if (gethostname(hostname, sizeof(hostname)) == 0) {
        oss << ",\"host\":\"" << hostname << "\"";
    }
 
    // Tamper-evident signature (Issue #612). The signature covers the
    // unsigned JSON body (closed with '}'); verifiers recompute by
    // stripping the trailing ,"sig_alg":"..","signature":".." suffix,
    // which is a stable canonical form on the wire.
    if (integrity_policy_) {
        std::string unsigned_body = oss.str() + "}";
        std::string sig = integrity_policy_->sign(unsigned_body);
        if (!sig.empty()) {
            oss << ",\"sig_alg\":\"" << integrity_policy_->name() << "\"";
            oss << ",\"signature\":\"" << sig << "\"";
        }
    }
 
    oss << "}\n";
    return oss.str();
}
 
void network_writer::set_integrity_policy(
    std::shared_ptr<security::integrity_policy> policy) {
    std::lock_guard<std::mutex> lock(buffer_mutex_);
    integrity_policy_ = std::move(policy);
}
 
std::string network_writer::escape_json(const std::string& str) const {
    std::string escaped;
    for (char c : str) {
        if (c == '"') escaped += "\\\"";
        else if (c == '\\') escaped += "\\\\";
        else if (c == '\n') escaped += "\\n";
        else if (c == '\r') escaped += "\\r";
        else if (c == '\t') escaped += "\\t";
        else escaped += c;
    }
    return escaped;
}
 
} // namespace kcenon::logger