Add span events and status

core/include/userver/tracing/in_place_span.hpp

@@ -37,7 +37,7 @@ class InPlaceSpan final {
 
 private:
     struct Impl;
-    utils::FastPimpl<Impl, 4224, 8> impl_;
+    utils::FastPimpl<Impl, 4240, 8> impl_;
 };
 
 }  // namespace tracing

core/include/userver/tracing/span.hpp

@@ -33,6 +33,34 @@ class Span final {
 public:
     class Impl;
 
+    /// @brief Operation status code.
+    enum class StatusCode : uint8_t {
+        kUnset,  // Default status.
+        kOk,     // Operation has completed successfully.
+        kError,  // The operation contains an error.
+    };
+
+    /// @brief Span event.
+    /// @see
+    /// https://github.com/open-telemetry/opentelemetry-proto/blob/main/opentelemetry/proto/trace/v1/trace.proto#L222.
+    /// @todo Implement attributes.
+    struct Event final {
+        /// @brief Constructor.
+        /// @param name Event name.
+        explicit Event(std::string_view name);
+
+        /// @brief Constructor.
+        /// @param name Event name.
+        /// @param time_unix_nano Event timestamp.
+        Event(std::string_view name, uint64_t time_unix_nano);
+
+        /// @brief Event name.
+        std::string name;
+
+        /// @brief Event timestamp.
+        uint64_t time_unix_nano{};
+    };
+
     explicit Span(
         TracerPtr tracer,
         std::string name,
@@ -157,6 +185,12 @@ class Span final {
     /// @overload AddNonInheritableTag
     void AddNonInheritableTags(const logging::LogExtra&);
 
+    /// Add an event to Span.
+    void AddEvent(std::string_view event_name);
+
+    /// Set span status.
+    void SetStatus(StatusCode status, const std::string_view description);
+
     /// @brief Sets level for tags logging
     void SetLogLevel(logging::Level log_level);
 

core/include/userver/utils/async.hpp

@@ -41,7 +41,7 @@ struct SpanWrapCall {
 
     struct Impl;
 
-    static constexpr std::size_t kImplSize = 4264;
+    static constexpr std::size_t kImplSize = 4280;
     static constexpr std::size_t kImplAlign = 8;
     utils::FastPimpl<Impl, kImplSize, kImplAlign> pimpl_;
 };

core/src/tracing/span.cpp

@@ -149,6 +149,7 @@ void Span::Impl::PutIntoLogger(logging::impl::TagWriter writer) && {
     }
     writer.PutLogExtra(log_extra_inheritable_);
 
+    LogEvents(writer);
     LogOpenTracing();
 }
 
@@ -368,6 +369,29 @@ void Span::AddTagFrozen(std::string key, logging::LogExtra::Value value) {
     pimpl_->log_extra_inheritable_.Extend(std::move(key), std::move(value), logging::LogExtra::ExtendType::kFrozen);
 }
 
+void Span::AddEvent(std::string_view event_name) { pimpl_->events_.emplace_back(event_name); }
+
+// Opentelemetry implementation:
+// https://github.com/open-telemetry/opentelemetry-cpp/blob/v1.18.0/exporters/zipkin/src/recordable.cc#L220
+void Span::SetStatus(StatusCode status, const std::string_view description) {
+    if (status == StatusCode::kUnset) {
+        return;
+    }
+
+    AddTag("otel.status_description", std::string{description});
+
+    if (status == StatusCode::kError) {
+        // Error description is required by OpenTelemetry
+
+        // Service data
+        AddTag("otel.status_code", "ERROR");
+        AddTag("error", "true");
+    } else {
+        AddTag("otel.status_code", "OK");
+        AddTag("error", "false");
+    }
+}
+
 void Span::SetLink(std::string link) { AddTagFrozen(kLinkTag, std::move(link)); }
 
 void Span::SetParentLink(std::string parent_link) { AddTagFrozen(kParentLinkTag, std::move(parent_link)); }
@@ -412,6 +436,11 @@ const Span::Impl* GetParentSpanImpl() {
     return !spans_ptr || spans_ptr->empty() ? nullptr : &spans_ptr->back();
 }
 
+Span::Event::Event(std::string_view name)
+    : Span::Event{name, static_cast<uint64_t>(std::chrono::system_clock::now().time_since_epoch().count())} {}
+
+Span::Event::Event(std::string_view name, uint64_t time_unix_nano) : name{name}, time_unix_nano{time_unix_nano} {}
+
 namespace impl {
 
 struct DetachLocalSpansScope::Impl {

core/src/tracing/span_impl.hpp

@@ -76,6 +76,7 @@ class Span::Impl : public boost::intrusive::list_base_hook<boost::intrusive::lin
     void AttachToCoroStack();
 
 private:
+    void LogEvents(logging::impl::TagWriter& writer) const;
     void LogOpenTracing() const;
     void DoLogOpenTracing(logging::impl::TagWriter writer) const;
     static void AddOpentracingTags(formats::json::StringBuilder& output, const logging::LogExtra& input);
@@ -105,6 +106,8 @@ class Span::Impl : public boost::intrusive::list_base_hook<boost::intrusive::lin
     const ReferenceType reference_type_;
     utils::impl::SourceLocation source_location_;
 
+    std::vector<Span::Event> events_;
+
     friend class Span;
     friend class SpanBuilder;
     friend class TagScope;

core/src/tracing/span_opentracing.cpp

@@ -79,6 +79,7 @@ constexpr std::string_view kStartTimeMillis = "start_time_millis";
 constexpr std::string_view kDuration = "duration";
 
 }  // namespace jaeger
+
 }  // namespace
 
 void Span::Impl::LogOpenTracing() const {
@@ -123,6 +124,24 @@ void Span::Impl::DoLogOpenTracing(logging::impl::TagWriter writer) const {
     writer.PutTag("tags", tags.GetStringView());
 }
 
+void Span::Impl::LogEvents(logging::impl::TagWriter& writer) const {
+    if (events_.empty()) {
+        return;
+    }
+
+    formats::json::StringBuilder events;
+    {
+        const formats::json::StringBuilder::ObjectGuard event_guard(events);
+
+        for (const auto& event : events_) {
+            events.Key(event.name);
+            events.WriteUInt64(event.time_unix_nano);
+        }
+    }
+
+    writer.PutTag("events", events.GetStringView());
+}
+
 void Span::Impl::AddOpentracingTags(formats::json::StringBuilder& output, const logging::LogExtra& input) {
     for (const auto& [key, value] : *input.extra_) {
         const auto tag_it = jaeger::kGetOpentracingTags.TryFind(key);

core/src/tracing/span_test.cpp

@@ -136,7 +136,10 @@ UTEST_F(Span, LogBufferSize) {
 }
 
 UTEST_F(Span, SourceLocation) {
+    // clang-format off
     { tracing::Span span("span_name"); }
+    // clang-format on
+
     logging::LogFlush();
     EXPECT_THAT(GetStreamString(), HasSubstr("module=TestBody ( "));
     EXPECT_THAT(GetStreamString(), HasSubstr("userver/core/src/tracing"));
@@ -204,7 +207,10 @@ UTEST_F(OpentracingSpan, Tags) {
 
 UTEST_F(OpentracingSpan, FromTracerWithServiceName) {
     auto tracer = tracing::MakeTracer("test_service", tracing::Tracer::GetTracer()->GetOptionalLogger());
+    // clang-format off
     { tracing::Span span(tracer, "span_name", nullptr, tracing::ReferenceType::kChild); }
+    // clang-format on
+
     FlushOpentracing();
     const auto log_str = GetOtStreamString();
     EXPECT_THAT(log_str, HasSubstr("service_name=test_service"));
@@ -430,6 +436,7 @@ UTEST_F(Span, NoLogPrefixes) {
     };
     tracing::Tracer::SetNoLogSpans(std::move(no_logs));
 
+    // clang-format off
     { tracing::Span a{kIgnorePrefix0 + "foo"}; }
     { tracing::Span a{kLogSpan0}; }
     { tracing::Span a{kLogSpan1}; }
@@ -442,6 +449,7 @@ UTEST_F(Span, NoLogPrefixes) {
     { tracing::Span a{kIgnorePrefix1}; }
     { tracing::Span a{kIgnorePrefix2}; }
     { tracing::Span a{kIgnoreSpan}; }
+    // clang-format on
 
     logging::LogFlush();
 
@@ -479,6 +487,7 @@ UTEST_F(Span, NoLogMixed) {
     const std::string kIgnorePrefix1 = "skip";
     const std::string kIgnorePrefix2 = "do_not_keep";
 
+    // clang-format off
     { tracing::Span a{kIgnorePrefix0 + "oops"}; }
     { tracing::Span a{kLogSpan0}; }
     { tracing::Span a{kLogSpan1}; }
@@ -492,6 +501,7 @@ UTEST_F(Span, NoLogMixed) {
     { tracing::Span a{kIgnorePrefix1}; }
     { tracing::Span a{kIgnorePrefix2}; }
     { tracing::Span a{kIgnoreSpan}; }
+    // clang-format on
 
     logging::LogFlush();
 
@@ -689,4 +699,42 @@ UTEST_F(Span, MakeSpanWithParentIdTraceIdLinkWithExisting) {
     }
 }
 
+UTEST_F(Span, MakeSpanEvent) {
+    {
+        tracing::Span root_span("root_span");
+
+        auto error_span = root_span.CreateChild("error_span");
+        error_span.AddEvent("error_event");
+    }
+
+    logging::LogFlush();
+
+    const auto logs_raw = GetStreamString();
+
+    EXPECT_THAT(logs_raw, HasSubstr("events={\"error_event\":"));
+    EXPECT_THAT(logs_raw, HasSubstr("root_span"));
+    EXPECT_THAT(logs_raw, HasSubstr("error_span"));
+}
+
+UTEST_F(Span, MakeSpanSetStatus) {
+    {
+        tracing::Span error_span("error_span");
+        error_span.SetStatus(tracing::Span::StatusCode::kError, "Error description");
+        tracing::Span ok_span("ok_span");
+        ok_span.SetStatus(tracing::Span::StatusCode::kOk, "Ok description");
+    }
+
+    logging::LogFlush();
+
+    const auto logs_raw = GetStreamString();
+
+    EXPECT_THAT(logs_raw, HasSubstr("error_span"));
+    EXPECT_THAT(logs_raw, HasSubstr("otel_status_description=Error description"));
+    EXPECT_THAT(logs_raw, HasSubstr("otel_status_code=ERROR"));
+
+    EXPECT_THAT(logs_raw, HasSubstr("ok_span"));
+    EXPECT_THAT(logs_raw, HasSubstr("otel_status_description=Ok description"));
+    EXPECT_THAT(logs_raw, HasSubstr("otel_status_code=OK"));
+}
+
 USERVER_NAMESPACE_END

otlp/src/otlp/logs/component.cpp

@@ -44,6 +44,15 @@ LoggerComponent::LoggerComponent(const components::ComponentConfig& config, cons
     logger_config.extra_attributes = config["extra-attributes"].As<std::unordered_map<std::string, std::string>>({});
     logger_config.attributes_mapping =
         config["attributes-mapping"].As<std::unordered_map<std::string, std::string>>({});
+
+    // Define error mapping
+    if (logger_config.attributes_mapping.find("otel_status_code") == logger_config.attributes_mapping.end()) {
+        logger_config.attributes_mapping["otel_status_code"] = "otel.status_code";
+    }
+    if (logger_config.attributes_mapping.find("otel_status_description") == logger_config.attributes_mapping.end()) {
+        logger_config.attributes_mapping["otel_status_description"] = "otel.status_description";
+    }
+
     logger_config.logs_sink = config["sinks"]["logs"].As<SinkType>(SinkType::kOtlp);
     logger_config.tracing_sink = config["sinks"]["tracing"].As<SinkType>(SinkType::kOtlp);
 

otlp/src/otlp/logs/logger.cpp

@@ -4,6 +4,7 @@
 #include <iostream>
 
 #include <userver/engine/async.hpp>
+#include <userver/formats/json.hpp>
 #include <userver/formats/parse/common_containers.hpp>
 #include <userver/formats/parse/to.hpp>
 #include <userver/logging/impl/tag_writer.hpp>
@@ -21,11 +22,40 @@ USERVER_NAMESPACE_BEGIN
 namespace otlp {
 
 namespace {
+
 constexpr std::string_view kTelemetrySdkLanguage = "telemetry.sdk.language";
 constexpr std::string_view kTelemetrySdkName = "telemetry.sdk.name";
 constexpr std::string_view kServiceName = "service.name";
 
 const std::string kTimestampFormat = "%Y-%m-%dT%H:%M:%E*S";
+
+std::vector<tracing::Span::Event> GetEventsFromValue(const std::string_view value) {
+    std::vector<tracing::Span::Event> events;
+
+    const auto json_value = formats::json::FromString(value);
+
+    if (!json_value.IsObject()) {
+        throw std::runtime_error("Expected JSON object in \"value\"");
+    }
+
+    for (const auto& [key, value] : formats::common::Items(json_value)) {
+        events.emplace_back(key, value.As<uint64_t>());
+    }
+
+    return events;
+}
+
+void WriteEventsFromValue(::opentelemetry::proto::trace::v1::Span& span, std::string_view value) {
+    const std::vector<tracing::Span::Event> events = GetEventsFromValue(value);
+    span.mutable_events()->Reserve(events.size());
+
+    for (const auto& event : events) {
+        auto* event_proto = span.add_events();
+        event_proto->set_name(event.name);
+        event_proto->set_time_unix_nano(event.time_unix_nano);
+    }
+}
+
 }  // namespace
 
 SinkType Parse(const yaml_config::YamlConfig& value, formats::parse::To<SinkType>) {
@@ -93,8 +123,9 @@ void Logger::Log(logging::Level level, std::string_view msg) {
 
     ++stats_.by_level[static_cast<int>(level)];
 
-    [[maybe_unused]] auto parse_ok =
-        utils::encoding::TskvReadRecord(parser, [&](std::string_view key, std::string_view value) {
+    [[maybe_unused]] auto parse_ok = utils::encoding::TskvReadRecord(
+        parser,
+        [this, &log_record, &timestamp](std::string_view key, std::string_view value) {
             if (key == "text") {
                 log_record.mutable_body()->set_string_value(grpc::string(std::string{value}));
                 return true;
@@ -120,7 +151,8 @@ void Logger::Log(logging::Level level, std::string_view msg) {
             attributes->set_key(std::string{MapAttribute(key)});
             attributes->mutable_value()->set_string_value(std::string{value});
             return true;
-        });
+        }
+    );
 
     auto nanoseconds = std::chrono::duration_cast<std::chrono::nanoseconds>(timestamp.time_since_epoch());
     log_record.set_time_unix_nano(nanoseconds.count());
@@ -147,8 +179,9 @@ void Logger::Trace(logging::Level level, std::string_view msg) {
     std::string start_timestamp;
     std::string total_time;
 
-    [[maybe_unused]] auto parse_ok =
-        utils::encoding::TskvReadRecord(parser, [&](std::string_view key, std::string_view value) {
+    [[maybe_unused]] auto parse_ok = utils::encoding::TskvReadRecord(
+        parser,
+        [this, &span, &start_timestamp, &total_time](std::string_view key, std::string_view value) {
             if (key == "trace_id") {
                 span.set_trace_id(utils::encoding::FromHex(value));
                 return true;
@@ -176,12 +209,24 @@ void Logger::Trace(logging::Level level, std::string_view msg) {
             if (key == "timestamp" || key == "text") {
                 return true;
             }
+            if (key == "events") {
+                WriteEventsFromValue(span, value);
+                return true;
+            }
+            if (key == "error") {
+                auto attributes = span.add_attributes();
+                attributes->set_key(std::string{key});
+                attributes->mutable_value()->set_bool_value(value == "true");
+
+                return true;
+            }
 
             auto attributes = span.add_attributes();
             attributes->set_key(std::string{MapAttribute(key)});
             attributes->mutable_value()->set_string_value(std::string{value});
             return true;
-        });
+        }
+    );
 
     auto start_timestamp_double = std::stod(start_timestamp);
     span.set_start_time_unix_nano(start_timestamp_double * 1'000'000'000);
@@ -226,7 +271,8 @@ void Logger::SendingLoop(Queue::Consumer& consumer, LogClient& log_client, Trace
                     [&scope_logs](const opentelemetry::proto::logs::v1::LogRecord& action) {
                         auto log_records = scope_logs->add_log_records();
                         *log_records = action;
-                    }},
+                    }
+                },
                 action
             );
         } while (consumer.Pop(action, deadline));