otlp_writer.h

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.
 
#pragma once
 
#include "base_writer.h"
#include "../interfaces/writer_category.h"
#include "../otlp/otel_context.h"
 
#include <kcenon/logger/logger_export.h>
 
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <memory>
#include <mutex>
#include <queue>
#include <string>
#include <thread>
#include <unordered_map>
#include <vector>
 
namespace kcenon::logger {
 
class LOGGER_SYSTEM_API otlp_writer : public base_writer, public async_writer_tag {
public:
    enum class protocol_type {
        http,   
        grpc    
    };
 
    struct config {
        std::string endpoint = "http://localhost:4318/v1/logs";
 
        protocol_type protocol = protocol_type::http;
 
        std::chrono::milliseconds timeout{5000};
 
        bool use_tls = false;
 
        std::string service_name;
 
        std::string service_version;
 
        std::string service_namespace;
 
        std::string service_instance_id;
 
        std::unordered_map<std::string, std::string> resource_attributes;
 
        std::size_t max_batch_size = 512;
 
        std::chrono::milliseconds flush_interval{5000};
 
        std::size_t max_queue_size = 10000;
 
        std::size_t max_retries = 3;
 
        std::chrono::milliseconds retry_delay{100};
 
        std::unordered_map<std::string, std::string> headers;
    };
 
    struct export_stats {
        uint64_t logs_exported{0};
        uint64_t logs_dropped{0};
        uint64_t export_success{0};
        uint64_t export_failures{0};
        uint64_t retries{0};
        std::chrono::system_clock::time_point last_export;
        std::chrono::system_clock::time_point last_error;
    };
 
    struct internal_stats {
        std::atomic<uint64_t> logs_exported{0};
        std::atomic<uint64_t> logs_dropped{0};
        std::atomic<uint64_t> export_success{0};
        std::atomic<uint64_t> export_failures{0};
        std::atomic<uint64_t> retries{0};
        std::chrono::system_clock::time_point last_export;
        std::chrono::system_clock::time_point last_error;
    };
 
    explicit otlp_writer(const config& cfg);
 
    ~otlp_writer() override;
 
    // Non-copyable, non-movable
    otlp_writer(const otlp_writer&) = delete;
    otlp_writer& operator=(const otlp_writer&) = delete;
    otlp_writer(otlp_writer&&) = delete;
    otlp_writer& operator=(otlp_writer&&) = delete;
 
    common::VoidResult write(const log_entry& entry) override;
 
    common::VoidResult flush() override;
 
    std::string get_name() const override { return "otlp"; }
 
    bool is_healthy() const override;
 
    export_stats get_stats() const;
 
    void force_export();
 
private:
    // Background export thread
    void export_thread_func();
 
    // Export batch to collector
    bool export_batch(const std::vector<log_entry>& batch);
 
    // Convert log level to OTLP severity
    static int to_otlp_severity(common::interfaces::log_level level);
 
#ifdef LOGGER_HAS_OTLP
    // Export using OpenTelemetry SDK
    bool export_with_otel_sdk(const std::vector<log_entry>& batch);
#else
    // Fallback: Export using HTTP directly
    bool export_with_http(const std::vector<log_entry>& batch);
 
    // JSON string escaping helper
    static std::string escape_json(const std::string& str);
#endif
 
private:
    config config_;
    internal_stats stats_;
 
    // Queue management
    std::queue<log_entry> queue_;
    mutable std::mutex queue_mutex_;
    std::condition_variable queue_cv_;
 
    // Background thread
    std::unique_ptr<std::thread> export_thread_;
    std::atomic<bool> running_{false};
    std::atomic<bool> healthy_{true};
 
#ifdef LOGGER_HAS_OTLP
    // OpenTelemetry SDK components (forward declared, implemented in cpp)
    class otel_impl;
    std::unique_ptr<otel_impl> otel_impl_;
#endif
};
 
} // namespace kcenon::logger