OTLP log exporter for OpenTelemetry integration. More...

#include <otlp_writer.h>

Inheritance diagram for kcenon::logger::otlp_writer:

Collaboration diagram for kcenon::logger::otlp_writer:

Classes
struct	config
	Configuration for OTLP writer. More...

struct	export_stats
	Statistics snapshot for OTLP export (copyable) More...

struct	internal_stats
	Internal atomic statistics (non-copyable) More...

Public Types
enum class	protocol_type { http , grpc }
	Transport protocol for OTLP export. More...

Public Member Functions
	otlp_writer (const config &cfg)
	Constructor with configuration.

	~otlp_writer () override
	Destructor - flushes remaining logs.

	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
	Write a log entry with OTEL context.

common::VoidResult	flush () override
	Flush pending logs.

std::string	get_name () const override
	Get writer name.

bool	is_healthy () const override
	Check if writer is healthy.

export_stats	get_stats () const
	Get export statistics.

void	force_export ()
	Force immediate export of current batch.

Public Member Functions inherited from kcenon::logger::base_writer
	base_writer (std::unique_ptr< log_formatter_interface > formatter=nullptr)
	Constructor with optional formatter.

virtual	~base_writer ()=default

virtual void	set_use_color (bool use_color)
	Set whether to use color output (if supported)

bool	use_color () const
	Get current color output setting.

log_formatter_interface *	get_formatter () const
	Get the current formatter.

Public Member Functions inherited from kcenon::logger::log_writer_interface
virtual	~log_writer_interface ()=default

virtual common::VoidResult	close ()
	Close the writer and release resources.

virtual auto	is_open () const -> bool
	Check if writer is open and ready.

Private Member Functions
void	export_thread_func ()

bool	export_batch (const std::vector< log_entry > &batch)

bool	export_with_http (const std::vector< log_entry > &batch)

Static Private Member Functions
static int	to_otlp_severity (common::interfaces::log_level level)

static std::string	escape_json (const std::string &str)

Private Attributes
config	config_

internal_stats	stats_

std::queue< log_entry >	queue_

std::mutex	queue_mutex_

std::condition_variable	queue_cv_

std::unique_ptr< std::thread >	export_thread_

std::atomic< bool >	running_ {false}

std::atomic< bool >	healthy_ {true}

Additional Inherited Members
Static Public Attributes inherited from kcenon::logger::async_writer_tag
static constexpr writer_category	category = writer_category::asynchronous

Protected Member Functions inherited from kcenon::logger::base_writer
std::string	format_log_entry (const log_entry &entry) const
	Format a log entry using the current formatter.

Detailed Description

OTLP log exporter for OpenTelemetry integration.

Exports log records to OpenTelemetry collectors using the OTLP protocol. Supports:

HTTP and gRPC transport
Batch export for efficiency
Automatic retry with exponential backoff
Graceful degradation when collector unavailable

Note: This writer batches logs for network efficiency. Logs may be delayed by up to flush_interval before being sent.

Warning: Requires OpenTelemetry collector endpoint to be available. Logs are dropped if collector is unavailable and queue is full.

Category: Asynchronous (batched network export with background thread)

Since: 3.0.0; 1.4.0 Added async_writer_tag for category classification

Definition at line 74 of file otlp_writer.h.

Member Enumeration Documentation

◆ protocol_type

enum class kcenon::logger::otlp_writer::protocol_type

strong

Transport protocol for OTLP export.

Enumerator
http	OTLP/HTTP (port 4318 by default)
grpc	OTLP/gRPC (port 4317 by default)

Definition at line 80 of file otlp_writer.h.

Constructor & Destructor Documentation

◆ otlp_writer() [1/3]

kcenon::logger::otlp_writer::otlp_writer ( const config & cfg )

explicit

Constructor with configuration.

Parameters

cfg	Configuration for OTLP export

Definition at line 113 of file otlp_writer.cpp.

    : config_(cfg) {
 
#ifdef LOGGER_HAS_OTLP
    otel_impl_ = std::make_unique<otel_impl>(cfg);
#endif
 
    // Start background export thread
    running_.store(true);
    export_thread_ = std::make_unique<std::thread>([this]() {
        export_thread_func();
    });
}

References export_thread_, export_thread_func(), and running_.

Here is the call graph for this function:

◆ ~otlp_writer()

kcenon::logger::otlp_writer::~otlp_writer ( )

override

Destructor - flushes remaining logs.

Definition at line 127 of file otlp_writer.cpp.

                          {
    // Stop background thread
    running_.store(false);
 
    // Wake up export thread
    {
        std::lock_guard<std::mutex> lock(queue_mutex_);
        queue_cv_.notify_one();
    }
 
    // Wait for thread to finish
    if (export_thread_ && export_thread_->joinable()) {
        export_thread_->join();
    }
 
    // Export remaining logs
    std::vector<log_entry> remaining;
    {
        std::lock_guard<std::mutex> lock(queue_mutex_);
        while (!queue_.empty()) {
            remaining.push_back(std::move(queue_.front()));
            queue_.pop();
        }
    }
 
    if (!remaining.empty()) {
        export_batch(remaining);
    }
}

References export_batch(), export_thread_, queue_, queue_cv_, queue_mutex_, and running_.

Here is the call graph for this function:

◆ otlp_writer() [2/3]

kcenon::logger::otlp_writer::otlp_writer ( const otlp_writer & )

delete

◆ otlp_writer() [3/3]

kcenon::logger::otlp_writer::otlp_writer ( otlp_writer && )

delete

Member Function Documentation

◆ escape_json()

std::string kcenon::logger::otlp_writer::escape_json ( const std::string & str )

staticprivate

Definition at line 457 of file otlp_writer.cpp.

                                                       {
    std::ostringstream result;
    for (char c : str) {
        switch (c) {
            case '"': result << "\\\""; break;
            case '\\': result << "\\\\"; break;
            case '\b': result << "\\b"; break;
            case '\f': result << "\\f"; break;
            case '\n': result << "\\n"; break;
            case '\r': result << "\\r"; break;
            case '\t': result << "\\t"; break;
            default:
                if (static_cast<unsigned char>(c) < 0x20) {
                    result << "\\u" << std::hex << std::setfill('0')
                           << std::setw(4) << static_cast<int>(c);
                } else {
                    result << c;
                }
        }
    }
    return result.str();
}

Referenced by export_with_http().

Here is the caller graph for this function:

◆ export_batch()

bool kcenon::logger::otlp_writer::export_batch ( const std::vector< log_entry > & batch )

private

Definition at line 262 of file otlp_writer.cpp.

                                                                {
    bool success = false;
    std::size_t retries = 0;
    auto delay = config_.retry_delay;
 
    while (!success && retries <= config_.max_retries) {
        if (retries > 0) {
            stats_.retries.fetch_add(1, std::memory_order_relaxed);
            std::this_thread::sleep_for(delay);
            delay *= 2;  // Exponential backoff
        }
 
#ifdef LOGGER_HAS_OTLP
        success = export_with_otel_sdk(batch);
#else
        success = export_with_http(batch);
#endif
 
        ++retries;
    }
 
    if (success) {
        stats_.logs_exported.fetch_add(batch.size(), std::memory_order_relaxed);
        stats_.export_success.fetch_add(1, std::memory_order_relaxed);
        stats_.last_export = std::chrono::system_clock::now();
        healthy_.store(true, std::memory_order_release);
    } else {
        stats_.logs_dropped.fetch_add(batch.size(), std::memory_order_relaxed);
        stats_.export_failures.fetch_add(1, std::memory_order_relaxed);
        stats_.last_error = std::chrono::system_clock::now();
        healthy_.store(false, std::memory_order_release);
    }
 
    return success;
}

Referenced by export_thread_func(), force_export(), and ~otlp_writer().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ export_thread_func()

void kcenon::logger::otlp_writer::export_thread_func ( )

private

Definition at line 231 of file otlp_writer.cpp.

                                     {
    while (running_.load(std::memory_order_acquire)) {
        std::vector<log_entry> batch;
 
        {
            std::unique_lock<std::mutex> lock(queue_mutex_);
 
            // Wait for batch size or timeout
            queue_cv_.wait_for(lock, config_.flush_interval, [this] {
                return !running_.load(std::memory_order_acquire) ||
                       queue_.size() >= config_.max_batch_size;
            });
 
            if (!running_.load(std::memory_order_acquire) && queue_.empty()) {
                break;
            }
 
            // Collect batch
            while (!queue_.empty() && batch.size() < config_.max_batch_size) {
                batch.push_back(std::move(queue_.front()));
                queue_.pop();
            }
        }
 
        // Export batch
        if (!batch.empty()) {
            export_batch(batch);
        }
    }
}

References config_, export_batch(), kcenon::logger::otlp_writer::config::flush_interval, kcenon::logger::otlp_writer::config::max_batch_size, queue_, queue_cv_, queue_mutex_, and running_.

Referenced by otlp_writer().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ export_with_http()

bool kcenon::logger::otlp_writer::export_with_http ( const std::vector< log_entry > & batch )

private

Definition at line 376 of file otlp_writer.cpp.

                                                                    {
    // Fallback implementation without OpenTelemetry SDK
    // This is a minimal HTTP export implementation for demonstration
    // In production, you would use a proper HTTP client library
 
    // Build JSON payload following OTLP/HTTP JSON format
    std::ostringstream json;
    json << R"({"resourceLogs":[{"resource":{"attributes":[)";
 
    // Add service name attribute
    json << R"({"key":"service.name","value":{"stringValue":")"
         << config_.service_name << R"("}}]},)";
 
    json << R"("scopeLogs":[{"logRecords":[)";
 
    bool first = true;
    for (const auto& entry : batch) {
        if (!first) json << ",";
        first = false;
 
        // Convert log_level from logger_system to common::interfaces
        auto level = static_cast<common::interfaces::log_level>(static_cast<int>(entry.level));
 
        // Convert timestamp to nanoseconds
        auto ns = std::chrono::duration_cast<std::chrono::nanoseconds>(
            entry.timestamp.time_since_epoch()).count();
 
        json << R"({"timeUnixNano":")" << ns << R"(",)";
        json << R"("severityNumber":)" << to_otlp_severity(level) << ",";
        json << R"("body":{"stringValue":")" << escape_json(entry.message.to_string()) << R"("},)";
 
        // Add attributes
        json << R"("attributes":[)";
        bool first_attr = true;
 
        if (entry.location) {
            std::string file = entry.location->file.to_string();
 
            if (!file.empty()) {
                if (!first_attr) json << ",";
                first_attr = false;
                json << R"({"key":"code.filepath","value":{"stringValue":")"
                     << escape_json(file) << R"("}})";
            }
 
            if (entry.location->line > 0) {
                if (!first_attr) json << ",";
                first_attr = false;
                json << R"({"key":"code.lineno","value":{"intValue":")"
                     << entry.location->line << R"("}})";
            }
        }
 
        if (entry.otel_ctx) {
            if (!first_attr) json << ",";
            first_attr = false;
            json << R"({"key":"trace_id","value":{"stringValue":")"
                 << entry.otel_ctx->trace_id << R"("}},)";
            json << R"({"key":"span_id","value":{"stringValue":")"
                 << entry.otel_ctx->span_id << R"("}})";
        }
 
        json << "]}";
    }
 
    json << "]}]}]}";
 
    // OTLP HTTP transport is not available (LOGGER_HAS_OTLP not defined).
    // The JSON payload was built but cannot be sent without an HTTP client.
    // Return false so callers know export did not succeed.
    static bool warned = false;
    if (!warned)
    {
        std::cerr << "[logger_system] WARNING: OTLP export configured but no HTTP transport "
                  << "available. Build with LOGGER_ENABLE_OTLP=ON and OpenTelemetry SDK, "
                  << "or provide an HTTP client library. Log data is NOT being exported.\n";
        warned = true;
    }
    return false;
}

References config_, escape_json(), kcenon::logger::json, kcenon::logger::otlp_writer::config::service_name, and to_otlp_severity().

Referenced by export_batch().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ flush()

common::VoidResult kcenon::logger::otlp_writer::flush ( )

overridevirtual

Flush pending logs.

Implements kcenon::logger::base_writer.

Definition at line 194 of file otlp_writer.cpp.

                                    {
    force_export();
    return common::ok();
}

References force_export(), and kcenon::common::ok().

Here is the call graph for this function:

◆ force_export()

void kcenon::logger::otlp_writer::force_export ( )

Force immediate export of current batch.

Definition at line 215 of file otlp_writer.cpp.

                               {
    std::vector<log_entry> batch;
 
    {
        std::lock_guard<std::mutex> lock(queue_mutex_);
        while (!queue_.empty()) {
            batch.push_back(std::move(queue_.front()));
            queue_.pop();
        }
    }
 
    if (!batch.empty()) {
        export_batch(batch);
    }
}

References export_batch(), queue_, and queue_mutex_.

Referenced by flush().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ get_name()

std::string kcenon::logger::otlp_writer::get_name ( ) const

inlineoverridevirtual

Get writer name.

Implements kcenon::logger::base_writer.

Definition at line 230 of file otlp_writer.h.

230{ return "otlp"; }

◆ get_stats()

otlp_writer::export_stats kcenon::logger::otlp_writer::get_stats ( ) const

Get export statistics.

Definition at line 203 of file otlp_writer.cpp.

                                                     {
    return export_stats{
        stats_.logs_exported.load(std::memory_order_relaxed),
        stats_.logs_dropped.load(std::memory_order_relaxed),
        stats_.export_success.load(std::memory_order_relaxed),
        stats_.export_failures.load(std::memory_order_relaxed),
        stats_.retries.load(std::memory_order_relaxed),
        stats_.last_export,
        stats_.last_error
    };
}

References kcenon::logger::otlp_writer::internal_stats::export_failures, kcenon::logger::otlp_writer::internal_stats::export_success, kcenon::logger::otlp_writer::internal_stats::last_error, kcenon::logger::otlp_writer::internal_stats::last_export, kcenon::logger::otlp_writer::internal_stats::logs_dropped, kcenon::logger::otlp_writer::internal_stats::logs_exported, kcenon::logger::otlp_writer::internal_stats::retries, and stats_.

◆ is_healthy()

bool kcenon::logger::otlp_writer::is_healthy ( ) const

overridevirtual

Check if writer is healthy.

Reimplemented from kcenon::logger::base_writer.

Definition at line 199 of file otlp_writer.cpp.

                                   {
    return healthy_.load(std::memory_order_acquire);
}

References healthy_.

◆ operator=() [1/2]

otlp_writer & kcenon::logger::otlp_writer::operator= ( const otlp_writer & )

delete

◆ operator=() [2/2]

otlp_writer & kcenon::logger::otlp_writer::operator= ( otlp_writer && )

delete

◆ to_otlp_severity()

int kcenon::logger::otlp_writer::to_otlp_severity ( common::interfaces::log_level level )

staticprivate

Definition at line 298 of file otlp_writer.cpp.

                                                                 {
    // OTLP Severity Numbers (per OpenTelemetry specification)
    // https://opentelemetry.io/docs/specs/otel/logs/data-model/#field-severitynumber
    switch (level) {
        case common::interfaces::log_level::trace:
            return 1;   // TRACE
        case common::interfaces::log_level::debug:
            return 5;   // DEBUG
        case common::interfaces::log_level::info:
            return 9;   // INFO
        case common::interfaces::log_level::warn:
            return 13;  // WARN
        case common::interfaces::log_level::error:
            return 17;  // ERROR
        case common::interfaces::log_level::fatal:
            return 21;  // FATAL
        default:
            return 9;   // INFO as default
    }
}

Referenced by export_with_http().

Here is the caller graph for this function:

◆ write()

common::VoidResult kcenon::logger::otlp_writer::write ( const log_entry & entry )

overridevirtual

Write a log entry with OTEL context.

Parameters

entry The log entry to write

Returns: common::VoidResult indicating success or error

Since: 3.5.0 This is now the only write method (legacy API removed)

Implements kcenon::logger::base_writer.

Definition at line 157 of file otlp_writer.cpp.

                                                          {
    {
        std::lock_guard<std::mutex> lock(queue_mutex_);
 
        // Check queue size limit
        if (queue_.size() >= config_.max_queue_size) {
            stats_.logs_dropped.fetch_add(1, std::memory_order_relaxed);
            return common::ok();  // Drop silently to avoid backpressure
        }
 
        // Create a copy of the entry since log_entry is move-only
        if (entry.location) {
            queue_.emplace(entry.level,
                          entry.message.to_string(),
                          entry.location->file.to_string(),
                          entry.location->line,
                          entry.location->function.to_string(),
                          entry.timestamp);
        } else {
            queue_.emplace(entry.level,
                          entry.message.to_string(),
                          entry.timestamp);
        }
 
        // Copy OTEL context to the queued entry
        auto& queued_entry = queue_.back();
        queued_entry.otel_ctx = entry.otel_ctx.has_value() ? entry.otel_ctx : otlp::otel_context_storage::get();
 
        // Wake up export thread if batch size reached
        if (queue_.size() >= config_.max_batch_size) {
            queue_cv_.notify_one();
        }
    }
 
    return common::ok();
}

References config_, kcenon::logger::otlp::otel_context_storage::get(), kcenon::logger::log_entry::level, kcenon::logger::log_entry::location, kcenon::logger::otlp_writer::internal_stats::logs_dropped, kcenon::logger::otlp_writer::config::max_batch_size, kcenon::logger::otlp_writer::config::max_queue_size, kcenon::logger::log_entry::message, kcenon::common::ok(), kcenon::logger::log_entry::otel_ctx, queue_, queue_cv_, queue_mutex_, stats_, kcenon::logger::log_entry::timestamp, and kcenon::logger::small_string< SSO_SIZE >::to_string().

Here is the call graph for this function:

Member Data Documentation

◆ config_

config kcenon::logger::otlp_writer::config_

private

Definition at line 269 of file otlp_writer.h.

Referenced by export_batch(), export_thread_func(), export_with_http(), and write().

◆ export_thread_

std::unique_ptr<std::thread> kcenon::logger::otlp_writer::export_thread_

private

Definition at line 278 of file otlp_writer.h.

Referenced by otlp_writer(), and ~otlp_writer().

◆ healthy_

std::atomic<bool> kcenon::logger::otlp_writer::healthy_ {true}

private

Definition at line 280 of file otlp_writer.h.

280{true};

Referenced by export_batch(), and is_healthy().

◆ queue_

std::queue<log_entry> kcenon::logger::otlp_writer::queue_

private

Definition at line 273 of file otlp_writer.h.

Referenced by export_thread_func(), force_export(), write(), and ~otlp_writer().

◆ queue_cv_

std::condition_variable kcenon::logger::otlp_writer::queue_cv_

private

Definition at line 275 of file otlp_writer.h.

Referenced by export_thread_func(), write(), and ~otlp_writer().

◆ queue_mutex_

std::mutex kcenon::logger::otlp_writer::queue_mutex_

mutableprivate

Definition at line 274 of file otlp_writer.h.

Referenced by export_thread_func(), force_export(), write(), and ~otlp_writer().

◆ running_

std::atomic<bool> kcenon::logger::otlp_writer::running_ {false}

private

Definition at line 279 of file otlp_writer.h.

279{false};

Referenced by export_thread_func(), otlp_writer(), and ~otlp_writer().

◆ stats_

internal_stats kcenon::logger::otlp_writer::stats_

private

Definition at line 270 of file otlp_writer.h.

Referenced by export_batch(), get_stats(), and write().

The documentation for this class was generated from the following files:

include/kcenon/logger/writers/otlp_writer.h
src/impl/writers/otlp_writer.cpp

Classes

Public Types

Public Member Functions

Private Member Functions

Static Private Member Functions

Private Attributes

Additional Inherited Members

Detailed Description

Member Enumeration Documentation

◆ protocol_type

Constructor & Destructor Documentation

◆ otlp_writer() [1/3]

◆ ~otlp_writer()

◆ otlp_writer() [2/3]

◆ otlp_writer() [3/3]

Member Function Documentation

◆ escape_json()

◆ export_batch()

◆ export_thread_func()

◆ export_with_http()

◆ flush()

◆ force_export()

◆ get_name()

◆ get_stats()

◆ is_healthy()

◆ operator=() [1/2]

◆ operator=() [2/2]

◆ to_otlp_severity()

◆ write()

Member Data Documentation

◆ config_

◆ export_thread_

◆ healthy_

◆ queue_

◆ queue_cv_

◆ queue_mutex_

◆ running_

◆ stats_