distributed_tracer.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 <string>
#include <memory>
#include <chrono>
#include <vector>
#include <unordered_map>
#include <mutex>
#include <atomic>
#include <optional>
 
#include <kcenon/monitoring/core/result_types.h>
#include <kcenon/monitoring/core/error_codes.h>
#include <kcenon/monitoring/context/thread_context.h>
 
namespace kcenon { namespace monitoring {
 
// Forward declaration for trace exporter interface
class trace_exporter_interface;
 
struct trace_export_settings {
    std::size_t batch_size = 100;           
    std::chrono::milliseconds flush_interval{5000};  
    std::size_t max_queue_size = 2048;      
    bool export_on_finish = true;           
};
 
struct trace_span {
    std::string trace_id;           // Unique trace identifier
    std::string span_id;            // Unique span identifier
    std::string parent_span_id;     // Parent span ID (empty for root span)
    std::string operation_name;     // Name of the operation
    std::string service_name;       // Service executing this span
 
    // Timing information
    std::chrono::system_clock::time_point start_time;
    std::chrono::system_clock::time_point end_time;
    std::chrono::microseconds duration{0};
 
    // Context and metadata
    std::unordered_map<std::string, std::string> tags;
    std::unordered_map<std::string, std::string> baggage;
 
    // Status information
    enum class status_code {
        unset,
        ok,
        error
    };
    status_code status{status_code::unset};
    std::string status_message;
 
    bool is_finished() const {
        return end_time != std::chrono::system_clock::time_point{};
    }
 
    void calculate_duration() {
        if (is_finished()) {
            duration = std::chrono::duration_cast<std::chrono::microseconds>(
                end_time - start_time
            );
        }
    }
};
 
struct trace_context {
    std::string trace_id;
    std::string span_id;
    std::string trace_flags;
    std::string trace_state;
    std::unordered_map<std::string, std::string> baggage;
 
    std::string to_w3c_traceparent() const {
        // Version-TraceId-SpanId-TraceFlags
        return "00-" + trace_id + "-" + span_id + "-" + trace_flags;
    }
 
    static common::Result<trace_context> from_w3c_traceparent(const std::string& header) {
        // Parse format: version-traceid-spanid-traceflags
        if (header.length() < 55) {  // Minimum valid length
            error_info err(kcenon::monitoring::monitoring_error_code::invalid_argument,
                          "Invalid traceparent header length");
            return common::Result<trace_context>::err(err.to_common_error());
        }
 
        trace_context ctx;
        // Simple parsing - production code would be more robust
        size_t pos = 0;
        size_t dash1 = header.find('-', pos);
        size_t dash2 = header.find('-', dash1 + 1);
        size_t dash3 = header.find('-', dash2 + 1);
 
        if (dash1 == std::string::npos || dash2 == std::string::npos || dash3 == std::string::npos) {
            error_info err(kcenon::monitoring::monitoring_error_code::invalid_argument,
                          "Invalid traceparent header format");
            return common::Result<trace_context>::err(err.to_common_error());
        }
 
        ctx.trace_id = header.substr(dash1 + 1, dash2 - dash1 - 1);
        ctx.span_id = header.substr(dash2 + 1, dash3 - dash2 - 1);
        ctx.trace_flags = header.substr(dash3 + 1);
 
        return common::ok(ctx);
    }
};
 
class span_builder {
private:
    trace_span span_;
 
public:
    span_builder& with_trace_id(const std::string& id) {
        span_.trace_id = id;
        return *this;
    }
 
    span_builder& with_parent(const std::string& parent_id) {
        span_.parent_span_id = parent_id;
        return *this;
    }
 
    span_builder& with_operation(const std::string& name) {
        span_.operation_name = name;
        return *this;
    }
 
    span_builder& with_service(const std::string& name) {
        span_.service_name = name;
        return *this;
    }
 
    span_builder& with_tag(const std::string& key, const std::string& value) {
        span_.tags[key] = value;
        return *this;
    }
 
    span_builder& with_baggage(const std::string& key, const std::string& value) {
        span_.baggage[key] = value;
        return *this;
    }
 
    trace_span build() {
        if (span_.span_id.empty()) {
            span_.span_id = thread_context::generate_request_id();
        }
        if (span_.trace_id.empty()) {
            span_.trace_id = thread_context::generate_correlation_id();
        }
        span_.start_time = std::chrono::system_clock::now();
        return span_;
    }
};
 
class distributed_tracer {
private:
    struct tracer_impl;
    std::unique_ptr<tracer_impl> impl_;
 
public:
    distributed_tracer();
    ~distributed_tracer();
 
    // Disable copy
    distributed_tracer(const distributed_tracer&) = delete;
    distributed_tracer& operator=(const distributed_tracer&) = delete;
 
    // Enable move
    distributed_tracer(distributed_tracer&&) noexcept;
    distributed_tracer& operator=(distributed_tracer&&) noexcept;
 
    common::Result<std::shared_ptr<trace_span>> start_span(
        const std::string& operation_name,
        const std::string& service_name = "monitoring_system"
    );
 
    common::Result<std::shared_ptr<trace_span>> start_child_span(
        const trace_span& parent,
        const std::string& operation_name
    );
 
    common::Result<std::shared_ptr<trace_span>> start_span_from_context(
        const trace_context& context,
        const std::string& operation_name
    );
 
    common::Result<bool> finish_span(std::shared_ptr<trace_span> span);
 
    std::shared_ptr<trace_span> get_current_span() const;
 
    void set_current_span(std::shared_ptr<trace_span> span);
 
    trace_context extract_context(const trace_span& span) const;
 
    template<typename Carrier>
    void inject_context(const trace_context& context, Carrier& carrier) {
        carrier["traceparent"] = context.to_w3c_traceparent();
        if (!context.trace_state.empty()) {
            carrier["tracestate"] = context.trace_state;
        }
        // Inject baggage
        for (const auto& [key, value] : context.baggage) {
            carrier["baggage-" + key] = value;
        }
    }
 
    template<typename Carrier>
    common::Result<trace_context> extract_context_from_carrier(const Carrier& carrier) {
        auto traceparent_it = carrier.find("traceparent");
        if (traceparent_it == carrier.end()) {
            return common::Result<trace_context>::err(error_info(kcenon::monitoring::monitoring_error_code::not_found,
                "Traceparent not found in carrier").to_common_error());
        }
 
        auto ctx_result = trace_context::from_w3c_traceparent(traceparent_it->second);
        if (ctx_result.is_err()) {
            return common::Result<trace_context>::err(ctx_result.error());
        }
 
        auto ctx = ctx_result.value();
 
        // Extract tracestate if present
        auto tracestate_it = carrier.find("tracestate");
        if (tracestate_it != carrier.end()) {
            ctx.trace_state = tracestate_it->second;
        }
 
        // Extract baggage
        for (const auto& [key, value] : carrier) {
            if (key.find("baggage-") == 0) {  // C++17 compatible
                ctx.baggage[key.substr(8)] = value;
            }
        }
 
        return ctx;
    }
 
    common::Result<std::vector<trace_span>> get_trace(const std::string& trace_id) const;
 
    common::Result<bool> export_spans(std::vector<trace_span> spans);
 
    void set_exporter(std::shared_ptr<trace_exporter_interface> exporter);
 
    std::shared_ptr<trace_exporter_interface> get_exporter() const;
 
    void configure_export(const trace_export_settings& settings);
 
    trace_export_settings get_export_settings() const;
 
    common::VoidResult flush();
 
    std::unordered_map<std::string, std::size_t> get_export_stats() const;
};
 
class scoped_span {
private:
    std::shared_ptr<trace_span> span_;
    distributed_tracer* tracer_;
 
public:
    scoped_span(std::shared_ptr<trace_span> span, distributed_tracer* tracer)
        : span_(span), tracer_(tracer) {
        if (tracer_) {
            tracer_->set_current_span(span_);
        }
    }
 
    ~scoped_span() {
        if (span_ && tracer_) {
            tracer_->finish_span(span_);
        }
    }
 
    // Disable copy
    scoped_span(const scoped_span&) = delete;
    scoped_span& operator=(const scoped_span&) = delete;
 
    // Enable move
    scoped_span(scoped_span&& other) noexcept
        : span_(std::move(other.span_)), tracer_(other.tracer_) {
        other.tracer_ = nullptr;
    }
 
    scoped_span& operator=(scoped_span&& other) noexcept {
        if (this != &other) {
            if (span_ && tracer_) {
                tracer_->finish_span(span_);
            }
            span_ = std::move(other.span_);
            tracer_ = other.tracer_;
            other.tracer_ = nullptr;
        }
        return *this;
    }
 
    trace_span* operator->() { return span_.get(); }
    const trace_span* operator->() const { return span_.get(); }
    trace_span& operator*() { return *span_; }
    const trace_span& operator*() const { return *span_; }
};
 
distributed_tracer& global_tracer();
 
#define TRACE_SPAN(operation_name) \
    auto _span_result = kcenon::monitoring::global_tracer().start_span(operation_name); \
    kcenon::monitoring::scoped_span _scoped_span( \
        _span_result.is_ok() ? _span_result.value() : nullptr, \
        &kcenon::monitoring::global_tracer() \
    )
 
#define TRACE_CHILD_SPAN(parent, operation_name) \
    auto _child_span_result = kcenon::monitoring::global_tracer().start_child_span(parent, operation_name); \
    kcenon::monitoring::scoped_span _child_scoped_span( \
        _child_span_result.is_ok() ? _child_span_result.value() : nullptr, \
        &kcenon::monitoring::global_tracer() \
    )
 
} } // namespace kcenon::monitoring