Skip to content

[EXPORTER] Refactor ElasticSearchRecordable #3164

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 7 commits into from
Dec 2, 2024
Merged

[EXPORTER] Refactor ElasticSearchRecordable #3164

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
db6398a
[EXPORTER] Refactor `ElasticSearchRecordable`
sjinks Nov 22, 2024
f789345
fix: format string in maintainer mode
sjinks Nov 23, 2024
aed45b2
fix: add used includes
sjinks Nov 23, 2024
59f1a5b
fix: typo: add missing `[`
sjinks Nov 23, 2024
b72c41b
refactor: faster version of `SetTimestamp()`
sjinks Nov 25, 2024
e4b0920
Merge branch 'main' into refactor-es-exporter
marcalff Dec 2, 2024
d49d0ec
Merge branch 'main' into refactor-es-exporter
marcalff Dec 2, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion exporters/elasticsearch/src/es_log_record_exporter.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,7 +63,7 @@ class ResponseHandler : public http_client::EventHandler
{
log_message = BuildResponseLogMessage(response, body_);

OTEL_INTERNAL_LOG_ERROR("ES Log Exporter] Export failed, " << log_message);
OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Export failed, " << log_message);
}

if (console_debug_)
Expand Down
193 changes: 39 additions & 154 deletions exporters/elasticsearch/src/es_log_recordable.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,14 +1,41 @@
// Copyright The OpenTelemetry Authors
// SPDX-License-Identifier: Apache-2.0

#include <chrono>
#include <ctime>
#include <nlohmann/json.hpp>
#include <string>

#include "opentelemetry/exporters/elasticsearch/es_log_recordable.h"
#include "opentelemetry/logs/severity.h"
#include "opentelemetry/nostd/variant.h"
#include "opentelemetry/sdk/instrumentationscope/instrumentation_scope.h"
#include "opentelemetry/sdk/resource/resource.h"
#include "opentelemetry/trace/span_id.h"
#include "opentelemetry/trace/trace_flags.h"
#include "opentelemetry/trace/trace_id.h"

namespace nlohmann
{
template <>
struct adl_serializer<opentelemetry::sdk::common::OwnedAttributeValue>
{
static void to_json(json &j, const opentelemetry::sdk::common::OwnedAttributeValue &v)
{
opentelemetry::nostd::visit([&j](const auto &value) { j = value; }, v);
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

A possible breaking change here is that the original WriteValue() did not handle spans. If this is important, I can add an if condition to follow the old behavior.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I just bumped into your changes and I was wondering if this could help putting json objects within the root json ?
The modifications I did was kind of a workaround so we could add keys like user.id that would be visible within ELK like the documentation asks for but the key was still user.id and not an object user under root that would have a key id

My hope would be to be able to be able to create an attribute or resource that contains a json object for some keys:

auto attr = { "user":  R"({"id": 1234}"_json};

}
};

template <>
struct adl_serializer<opentelemetry::common::AttributeValue>
{
static void to_json(json &j, const opentelemetry::common::AttributeValue &v)
{
opentelemetry::nostd::visit([&j](const auto &value) { j = value; }, v);
}
};
} // namespace nlohmann

OPENTELEMETRY_BEGIN_NAMESPACE
namespace exporter
{
Expand All @@ -18,146 +45,13 @@ void ElasticSearchRecordable::WriteValue(
const opentelemetry::sdk::common::OwnedAttributeValue &value,
const std::string &name)
{
namespace common = opentelemetry::sdk::common;
switch (value.index())
{
case common::kTypeBool:
json_[name] = opentelemetry::nostd::get<bool>(value) ? true : false;
return;
case common::kTypeInt:
json_[name] = opentelemetry::nostd::get<int>(value);
return;
case common::kTypeInt64:
json_[name] = opentelemetry::nostd::get<int64_t>(value);
return;
case common::kTypeUInt:
json_[name] = opentelemetry::nostd::get<unsigned int>(value);
return;
case common::kTypeUInt64:
json_[name] = opentelemetry::nostd::get<uint64_t>(value);
return;
case common::kTypeDouble:
json_[name] = opentelemetry::nostd::get<double>(value);
return;
case common::kTypeString:
json_[name] = opentelemetry::nostd::get<std::string>(value).data();
return;
default:
return;
}
json_[name] = value;
}

void ElasticSearchRecordable::WriteValue(const opentelemetry::common::AttributeValue &value,
const std::string &name)
{

// Assert size of variant to ensure that this method gets updated if the variant
// definition changes

if (nostd::holds_alternative<bool>(value))
{
json_[name] = opentelemetry::nostd::get<bool>(value) ? true : false;
}
else if (nostd::holds_alternative<int>(value))
{
json_[name] = opentelemetry::nostd::get<int>(value);
}
else if (nostd::holds_alternative<int64_t>(value))
{
json_[name] = opentelemetry::nostd::get<int64_t>(value);
}
else if (nostd::holds_alternative<unsigned int>(value))
{
json_[name] = opentelemetry::nostd::get<unsigned int>(value);
}
else if (nostd::holds_alternative<uint64_t>(value))
{
json_[name] = opentelemetry::nostd::get<uint64_t>(value);
}
else if (nostd::holds_alternative<double>(value))
{
json_[name] = opentelemetry::nostd::get<double>(value);
}
else if (nostd::holds_alternative<const char *>(value))
{
json_[name] = std::string(nostd::get<const char *>(value));
}
else if (nostd::holds_alternative<nostd::string_view>(value))
{
json_[name] = static_cast<std::string>(opentelemetry::nostd::get<nostd::string_view>(value));
}
else if (nostd::holds_alternative<nostd::span<const uint8_t>>(value))
{
nlohmann::json array_value = nlohmann::json::array();
for (const auto &val : nostd::get<nostd::span<const uint8_t>>(value))
{
array_value.push_back(val);
}
json_[name] = array_value;
}
else if (nostd::holds_alternative<nostd::span<const bool>>(value))
{
nlohmann::json array_value = nlohmann::json::array();
for (const auto &val : nostd::get<nostd::span<const bool>>(value))
{
array_value.push_back(val);
}
json_[name] = array_value;
}
else if (nostd::holds_alternative<nostd::span<const int>>(value))
{
nlohmann::json array_value = nlohmann::json::array();
for (const auto &val : nostd::get<nostd::span<const int>>(value))
{
array_value.push_back(val);
}
json_[name] = array_value;
}
else if (nostd::holds_alternative<nostd::span<const int64_t>>(value))
{
nlohmann::json array_value = nlohmann::json::array();
for (const auto &val : nostd::get<nostd::span<const int64_t>>(value))
{
array_value.push_back(val);
}
json_[name] = array_value;
}
else if (nostd::holds_alternative<nostd::span<const unsigned int>>(value))
{
nlohmann::json array_value = nlohmann::json::array();
for (const auto &val : nostd::get<nostd::span<const unsigned int>>(value))
{
array_value.push_back(val);
}
json_[name] = array_value;
}
else if (nostd::holds_alternative<nostd::span<const uint64_t>>(value))
{
nlohmann::json array_value = nlohmann::json::array();
for (const auto &val : nostd::get<nostd::span<const uint64_t>>(value))
{
array_value.push_back(val);
}
json_[name] = array_value;
}
else if (nostd::holds_alternative<nostd::span<const double>>(value))
{
nlohmann::json array_value = nlohmann::json::array();
for (const auto &val : nostd::get<nostd::span<const double>>(value))
{
array_value.push_back(val);
}
json_[name] = array_value;
}
else if (nostd::holds_alternative<nostd::span<const nostd::string_view>>(value))
{
nlohmann::json array_value = nlohmann::json::array();
for (const auto &val : nostd::get<nostd::span<const nostd::string_view>>(value))
{
array_value.push_back(static_cast<std::string>(val));
}
json_[name] = array_value;
}
json_[name] = value;
}

ElasticSearchRecordable::ElasticSearchRecordable() noexcept : sdk::logs::Recordable()
Expand All @@ -180,27 +74,19 @@ void ElasticSearchRecordable::SetTimestamp(
#if __cplusplus >= 202002L
const std::string dateStr = std::format("{:%FT%T%Ez}", timePoint);
#else
const static int dateToSecondsSize = 19;
const static int millisecondsSize = 8;
const static int timeZoneSize = 1;
const static int dateSize = dateToSecondsSize + millisecondsSize + timeZoneSize;

std::time_t time = std::chrono::system_clock::to_time_t(timePoint);
std::tm tm = *std::gmtime(&time);

char bufferDate[dateSize]; // example: 2024-10-18T07:26:00.123456Z
std::strftime(bufferDate, sizeof(bufferDate), "%Y-%m-%dT%H:%M:%S", &tm);
auto microseconds =
std::chrono::duration_cast<std::chrono::microseconds>(timePoint.time_since_epoch()) %
std::chrono::seconds(1);

char bufferMilliseconds[millisecondsSize];
std::snprintf(bufferMilliseconds, sizeof(bufferMilliseconds), ".%06ld",
// `sizeof()` includes the null terminator
constexpr auto dateSize = sizeof("YYYY-MM-DDTHH:MM:SS.uuuuuuZ");
char bufferDate[dateSize];
auto offset = std::strftime(bufferDate, sizeof(bufferDate), "%Y-%m-%dT%H:%M:%S", &tm);
std::snprintf(bufferDate + offset, sizeof(bufferDate) - offset, ".%06ldZ",
static_cast<long>(microseconds.count()));

std::strcat(bufferDate, bufferMilliseconds);
std::strcat(bufferDate, "Z");

const std::string dateStr(bufferDate);
#endif

Expand All @@ -221,9 +107,8 @@ void ElasticSearchRecordable::SetSeverity(opentelemetry::logs::Severity severity
std::uint32_t severity_index = static_cast<std::uint32_t>(severity);
if (severity_index >= std::extent<decltype(opentelemetry::logs::SeverityNumToText)>::value)
{
std::stringstream sout;
sout << "Invalid severity(" << severity_index << ")";
severityField = sout.str();
severityField =
std::string("Invalid severity(").append(std::to_string(severity_index)).append(")");
}
else
{
Expand All @@ -240,7 +125,7 @@ void ElasticSearchRecordable::SetTraceId(const opentelemetry::trace::TraceId &tr
{
if (trace_id.IsValid())
{
char trace_buf[32];
char trace_buf[opentelemetry::trace::TraceId::kSize * 2];
Comment on lines -243 to +128
Copy link
Contributor Author

@sjinks sjinks Nov 22, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This approach is safer: should kSize change someday, the buffer's size will be automatically updated.

trace_id.ToLowerBase16(trace_buf);
json_["traceid"] = std::string(trace_buf, sizeof(trace_buf));
}
Expand All @@ -254,7 +139,7 @@ void ElasticSearchRecordable::SetSpanId(const opentelemetry::trace::SpanId &span
{
if (span_id.IsValid())
{
char span_buf[16];
char span_buf[opentelemetry::trace::SpanId::kSize * 2];
span_id.ToLowerBase16(span_buf);
json_["spanid"] = std::string(span_buf, sizeof(span_buf));
}
Expand Down Expand Up @@ -282,7 +167,7 @@ void ElasticSearchRecordable::SetAttribute(
void ElasticSearchRecordable::SetResource(
const opentelemetry::sdk::resource::Resource &resource) noexcept
{
for (auto &attribute : resource.GetAttributes())
for (const auto &attribute : resource.GetAttributes())
{
WriteValue(attribute.second, attribute.first);
}
Expand Down
Loading