path: root/src/zentelemetry/otlpencoder.cpp

// Copyright Epic Games, Inc. All Rights Reserved.

#include "zentelemetry/otlpencoder.h"

#include <zenbase/zenbase.h>
#include <zentelemetry/otlptrace.h>

#include <spdlog/sinks/sink.h>
#include <zencore/testing.h>

#include <protozero/buffer_string.hpp>
#include <protozero/pbf_builder.hpp>

#include "otellogprotozero.h"
#include "otelmetricsprotozero.h"
#include "otelprotozero.h"
#include "oteltraceprotozero.h"

#if ZEN_WITH_OTEL

namespace zen {

OtlpEncoder::OtlpEncoder()
{
}

OtlpEncoder::~OtlpEncoder()
{
}

static int
MapSeverity(const spdlog::level::level_enum Level)
{
	switch (Level)
	{
		case spdlog::level::critical:
			return otel::SEVERITY_NUMBER_FATAL;
		case spdlog::level::err:
			return otel::SEVERITY_NUMBER_ERROR;
		case spdlog::level::warn:
			return otel::SEVERITY_NUMBER_WARN;
		case spdlog::level::info:
			return otel::SEVERITY_NUMBER_INFO;
		case spdlog::level::debug:
			return otel::SEVERITY_NUMBER_DEBUG;
		default:
		case spdlog::level::trace:
			return otel::SEVERITY_NUMBER_TRACE;
	}
}

static const char*
MapSeverityText(const spdlog::level::level_enum Level)
{
	switch (Level)
	{
		case spdlog::level::critical:
			return "fatal";
		case spdlog::level::err:
			return "error";
		case spdlog::level::warn:
			return "warn";
		case spdlog::level::info:
			return "info";
		case spdlog::level::debug:
			return "debug";
		default:
		case spdlog::level::trace:
			return "trace";
	}
}

std::string
OtlpEncoder::FormatOtelProtobuf(const spdlog::details::log_msg& Msg) const
{
	std::string Data;

	// LogsData
	{
		protozero::pbf_builder<otel::LogsData> Builder{Data};

		// ResourceLogs
		{
			protozero::pbf_builder<otel::ResourceLogs> RlBuilder{Builder, otel::LogsData::required_repeated_ResourceLogs_resource_logs};

			// ResourceLogs / Resource
			{
				protozero::pbf_builder<otel::Resource> Res{RlBuilder, otel::ResourceLogs::optional_Resource_resource};

				AppendResourceAttributes(Res);
			}

			// ScopeLogs scope_logs
			{
				protozero::pbf_builder<otel::ScopeLogs> SlBuilder{RlBuilder, otel::ResourceLogs::required_repeated_ScopeLogs_scope_logs};

				{
					protozero::pbf_builder<otel::InstrumentationScope> IsBuilder{SlBuilder,
																				 otel::ScopeLogs::required_InstrumentationScope_scope};

					IsBuilder.add_string(otel::InstrumentationScope::string_name, Msg.logger_name.data(), Msg.logger_name.size());
				}

				// LogRecord log_records
				{
					protozero::pbf_builder<otel::LogRecord> LrBuilder{SlBuilder, otel::ScopeLogs::required_repeated_LogRecord_log_records};

					LrBuilder.add_fixed64(otel::LogRecord::required_fixed64_time_unix_nano,
										  std::chrono::duration_cast<std::chrono::nanoseconds>(Msg.time.time_since_epoch()).count());

					const int Severity = MapSeverity(Msg.level);

					LrBuilder.add_enum(otel::LogRecord::optional_SeverityNumber_severity_number, Severity);

					LrBuilder.add_string(otel::LogRecord::optional_string_severity_text, MapSeverityText(Msg.level));

					otel::TraceId	   TraceId;
					const otel::SpanId SpanId = otel::Span::GetCurrentSpanId(TraceId);

					if (SpanId && TraceId)
					{
						LrBuilder.add_bytes(otel::LogRecord::optional_bytes_trace_id, TraceId.GetData(), TraceId.kSize);
						LrBuilder.add_bytes(otel::LogRecord::optional_bytes_span_id, SpanId.GetData(), SpanId.kSize);
					}

					// body
					{
						protozero::pbf_builder<otel::AnyValue> BodyBuilder{LrBuilder, otel::LogRecord::optional_anyvalue_body};

						BodyBuilder.add_string(otel::AnyValue::string_string_value, Msg.payload.data(), Msg.payload.size());
					}

					// attributes

					{
						protozero::pbf_builder<otel::KeyValue> KvBuilder{LrBuilder, otel::LogRecord::optional_repeated_kv_attributes};
						KvBuilder.add_string(otel::KeyValue::string_key, "thread_id");

						{
							protozero::pbf_builder<otel::AnyValue> AvBuilder{KvBuilder, otel::KeyValue::AnyValue_value};

							AvBuilder.add_int64(otel::AnyValue::int64_int_value, Msg.thread_id);
						}
					}
				}
			}
		}
	}

	return Data;
}

std::string
OtlpEncoder::FormatOtelMetrics() const
{
	std::string Data;

#	if 0
	static int64_t LastNanos = 0;

	const int64_t NowNanos =
		std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::system_clock::now().time_since_epoch()).count();

	// MetricsData
	{
		protozero::pbf_builder<otel::MetricsData> Builder{Data};

		// ResourceMetrics
		protozero::pbf_builder<otel::ResourceMetrics> Rm{Builder, otel::MetricsData::repeated_ResourceMetrics_resource_metrics};

		{
			protozero::pbf_builder<otel::Resource> Res{Rm, otel::ResourceMetrics::Resource_resource};

			AppendResourceAttributes(Res);
		}

		// ScopeMetrics
		protozero::pbf_builder<otel::ScopeMetrics> Sm{Rm, otel::ResourceMetrics::repeated_ScopeMetrics_scope_metrics};

		{
			// InstrumentationScope
			protozero::pbf_builder<otel::InstrumentationScope> Is{Sm, otel::ScopeMetrics::InstrumentationScope_scope};
			Is.add_string(otel::InstrumentationScope::string_name, "scope_name");
		}

		{
			protozero::pbf_builder<otel::Metric> Metric{Sm, otel::ScopeMetrics::repeated_Metric_metrics};
			Metric.add_string(otel::Metric::string_name, "metric_name");
			Metric.add_string(otel::Metric::string_unit, "KiB");

			// Gauge
			{
				protozero::pbf_builder<otel::Gauge> Gauge{Metric, otel::Metric::oneof_data_Gauge_gauge};

				protozero::pbf_builder<otel::NumberDataPoint> Dp{Gauge, otel::Gauge::repeated_NumberDataPoint_data_points};
				Dp.add_fixed64(otel::NumberDataPoint::fixed64_time_unix_nano, NowNanos);
				Dp.add_fixed64(otel::NumberDataPoint::fixed64_start_time_unix_nano, LastNanos);
				Dp.add_sfixed64(otel::NumberDataPoint::oneof_value_sfixed64_as_int, rand() * 470 / RAND_MAX);
			}
		}

		const int RequestCount = rand() % 600;

		{
			protozero::pbf_builder<otel::Metric> Metric{Sm, otel::ScopeMetrics::repeated_Metric_metrics};
			Metric.add_string(otel::Metric::string_name, "request_count");
			Metric.add_string(otel::Metric::string_unit, "requests");

			static int SumValue = 0;
			SumValue += RequestCount;

			// Sum
			{
				protozero::pbf_builder<otel::Sum> Sum{Metric, otel::Metric::oneof_data_Sum_sum};
				Sum.add_enum(otel::Sum::AggregationTemporality_aggregation_temporality, otel::AGGREGATION_TEMPORALITY_CUMULATIVE);
				Sum.add_bool(otel::Sum::bool_is_monotonic, true);

				protozero::pbf_builder<otel::NumberDataPoint> Dp{Sum, otel::Sum::repeated_NumberDataPoint_data_points};
				Dp.add_fixed64(otel::NumberDataPoint::fixed64_time_unix_nano, NowNanos);
				Dp.add_fixed64(otel::NumberDataPoint::fixed64_start_time_unix_nano, LastNanos);
				Dp.add_double(otel::NumberDataPoint::oneof_value_double_as_double, SumValue);
			}
		}

		{
			protozero::pbf_builder<otel::Metric> Metric{Sm, otel::ScopeMetrics::repeated_Metric_metrics};
			Metric.add_string(otel::Metric::string_name, "request_latency");
			Metric.add_string(otel::Metric::string_unit, "ms");

			// Histogram
			{
				protozero::pbf_builder<otel::Histogram> Histogram{Metric, otel::Metric::oneof_data_Histogram_histogram};
				Histogram.add_enum(otel::Histogram::AggregationTemporality_aggregation_temporality,
								   otel::AGGREGATION_TEMPORALITY_CUMULATIVE);

				protozero::pbf_builder<otel::HistogramDataPoint> Dp{Histogram, otel::Histogram::repeated_HistogramDataPoint_data_points};
				Dp.add_fixed64(otel::HistogramDataPoint::fixed64_time_unix_nano, NowNanos);
				Dp.add_fixed64(otel::HistogramDataPoint::fixed64_start_time_unix_nano, LastNanos);

				// Simulated latency value

				double Sum = 0, Min = 0, Max = 0;
				int	   Buckets[6] = {0};

				for (int i = 0; i < RequestCount; ++i)
				{
					const int Latency = rand() % 250;

					if (i == 0 || Latency < Min)
					{
						Min = Latency;
					}
					if (i == 0 || Latency > Max)
					{
						Max = Latency;
					}

					Sum += Latency;

					if (Latency >= 0 && Latency < 10)
					{
						++Buckets[0];  // [0,10)
					}
					else if (Latency >= 10 && Latency < 25)
					{
						++Buckets[1];  // [10,25)
					}
					else if (Latency >= 25 && Latency < 50)
					{
						++Buckets[2];  // [25,50)
					}
					else if (Latency >= 50 && Latency < 100)
					{
						++Buckets[3];  // [50,100)
					}
					else if (Latency >= 100 && Latency < 250)
					{
						++Buckets[4];  // [100,250)
					}
					else
					{
						++Buckets[5];  // [250,+inf)
					}
				}

				Dp.add_fixed64(otel::HistogramDataPoint::fixed64_count, RequestCount);
				Dp.add_double(otel::HistogramDataPoint::optional_double_sum, Sum);
				Dp.add_double(otel::HistogramDataPoint::optional_double_min, Min);
				Dp.add_double(otel::HistogramDataPoint::optional_double_max, Max);

				// Bucket bounds
				Dp.add_double(otel::HistogramDataPoint::repeated_double_explicit_bounds, 10.0);
				Dp.add_double(otel::HistogramDataPoint::repeated_double_explicit_bounds, 25.0);
				Dp.add_double(otel::HistogramDataPoint::repeated_double_explicit_bounds, 50.0);
				Dp.add_double(otel::HistogramDataPoint::repeated_double_explicit_bounds, 100.0);
				Dp.add_double(otel::HistogramDataPoint::repeated_double_explicit_bounds, 250.0);

				// Buckets
				for (int i = 0; i < 6; ++i)
				{
					Dp.add_fixed64(otel::HistogramDataPoint::repeated_fixed64_bucket_counts, Buckets[i]);
				}
			}
		}

#		if 0
		{
			protozero::pbf_builder<otel::Metric> Metric{Sm, otel::ScopeMetrics::repeated_Metric_metrics};
			Metric.add_string(otel::Metric::string_name, "request_payload");
			Metric.add_string(otel::Metric::string_unit, "KiB");

			// ExponentialHistogram
			{
				protozero::pbf_builder<otel::ExponentialHistogram> ExponentialHistogram{
					Metric, otel::Metric::oneof_data_ExponentialHistogram_exponential_histogram};
				ExponentialHistogram.add_enum(otel::ExponentialHistogram::AggregationTemporality_aggregation_temporality,
											 otel::AGGREGATION_TEMPORALITY_CUMULATIVE);

				protozero::pbf_builder<otel::ExponentialHistogramDataPoint> Dp{
					ExponentialHistogram, otel::ExponentialHistogram::repeated_ExponentialHistogramDataPoint_data_points};
				Dp.add_fixed64(otel::ExponentialHistogramDataPoint::fixed64_time_unix_nano, NowNanos);
				Dp.add_fixed64(otel::ExponentialHistogramDataPoint::fixed64_start_time_unix_nano, LastNanos);

				// Simulated payload size values

				int64_t sum = 0;
				int		count = 0;

				for (int i = 0; i < RequestCount; ++i)
				{
					const int PayloadSize = rand() % 1'500'000;  // [0,1500000) bytes

					sum += PayloadSize;
					++count;
				}

				Dp.add_fixed64(otel::ExponentialHistogramDataPoint::fixed64_count, count);
				Dp.add_double(otel::ExponentialHistogramDataPoint::optional_double_sum, sum);
				Dp.add_sint32(otel::ExponentialHistogramDataPoint::sint32_scale, 4);
			}
		}
#		endif
	}
	LastNanos = NowNanos;
#	endif
	return Data;
}

void
OtlpEncoder::AddResourceAttribute(const std::string_view& Key, const std::string_view& Value)
{
	m_ResourceLock.WithExclusiveLock([&] { m_ResourceAttributes[std::string(Key)] = std::string(Value); });
}

void
OtlpEncoder::AddResourceAttribute(const std::string_view& Key, int64_t Value)
{
	m_ResourceLock.WithExclusiveLock([&] { m_ResourceIntAttributes[std::string(Key)] = Value; });
}

void
OtlpEncoder::AppendResourceAttributes(protozero::pbf_builder<otel::Resource>& Res) const
{
	using namespace otel;

	m_ResourceLock.WithSharedLock([&] {
		for (auto const& [K, V] : m_ResourceAttributes)
		{
			protozero::pbf_builder<otel::KeyValue> KvBuilder{Res, otel::Resource::repeated_KeyValue_attributes};
			KvBuilder.add_string(otel::KeyValue::string_key, K.c_str());
			protozero::pbf_builder<otel::AnyValue> AnyBuilder{KvBuilder, otel::KeyValue::AnyValue_value};
			AnyBuilder.add_string(otel::AnyValue::string_string_value, V.c_str());
		}

		for (auto const& [K, V] : m_ResourceIntAttributes)
		{
			protozero::pbf_builder<otel::KeyValue> KvBuilder{Res, otel::Resource::repeated_KeyValue_attributes};
			KvBuilder.add_string(otel::KeyValue::string_key, K.c_str());
			protozero::pbf_builder<otel::AnyValue> AnyBuilder{KvBuilder, otel::KeyValue::AnyValue_value};
			AnyBuilder.add_int64(otel::AnyValue::int64_int_value, V);
		}
	});
}

template<typename ParentBuilder>
static void
AppendAttributesToBuilder(ParentBuilder& Parent, auto KeyValueField, const otel::AttributePair* Attrs)
{
	for (const otel::AttributePair* Attr = Attrs; Attr != nullptr; Attr = Attr->Next)
	{
		protozero::pbf_builder<otel::KeyValue> KvBuilder{Parent, KeyValueField};
		KvBuilder.add_string(otel::KeyValue::string_key, Attr->Key);

		protozero::pbf_builder<otel::AnyValue> AnyBuilder{KvBuilder, otel::KeyValue::AnyValue_value};

		if (Attr->IsNumeric())
		{
			AnyBuilder.add_int64(otel::AnyValue::int64_int_value, Attr->GetNumericValue());
		}
		else if (Attr->IsString())
		{
			AnyBuilder.add_string(otel::AnyValue::string_string_value, Attr->GetStringValue());
		}
	}
}

std::string
OtlpEncoder::FormatOtelTrace(zen::otel::TraceId Id, std::span<const zen::otel::Span*> Spans) const
{
	std::string Data;

	using namespace otel;

	// TracesData
	{
		protozero::pbf_builder<pbf::TracesData> Builder{Data};

		{
			protozero::pbf_builder<pbf::ResourceSpans> Rs{Builder, pbf::TracesData::repeated_ResourceSpans_resource_spans};

			{
				protozero::pbf_builder<otel::Resource> Res{Rs, pbf::ResourceSpans::Resource_resource};

				AppendResourceAttributes(Res);
			}

			for (const zen::otel::Span* SpanPtr : Spans)
			{
				protozero::pbf_builder<pbf::ScopeSpans> Ss{Rs, pbf::ResourceSpans::repeated_ScopeSpans_scope_spans};

				{
					// InstrumentationScope
					protozero::pbf_builder<otel::InstrumentationScope> Is{Ss, pbf::ScopeSpans::InstrumentationScope_scope};
					Is.add_string(otel::InstrumentationScope::string_name, "scope_name");
				}

				const SpanId ThisSpanId = SpanPtr->GetSpanId();

				{
					protozero::pbf_builder<pbf::Span> Sb{Ss, pbf::ScopeSpans::repeated_Span_spans};

					Sb.add_bytes(pbf::Span::required_bytes_trace_id, Id.GetData(), TraceId::kSize);
					Sb.add_bytes(pbf::Span::required_bytes_span_id, ThisSpanId.GetData(), SpanId::kSize);
					// Sb.add_string(pbf::Span::string_trace_state, "state-value");
					//
					if (const otel::Span* ParentSpan = SpanPtr->GetParentSpan())
					{
						const SpanId ParentSpanId = ParentSpan->GetSpanId();
						Sb.add_bytes(pbf::Span::bytes_parent_span_id, ParentSpanId.GetData(), SpanId::kSize);
					}

					Sb.add_fixed32(pbf::Span::fixed32_flags, 0);
					Sb.add_string(pbf::Span::required_string_name, SpanPtr->GetName());
					Sb.add_enum(pbf::Span::SpanKind_kind, (int)SpanPtr->GetKind());
					Sb.add_fixed64(pbf::Span::required_fixed64_start_time_unix_nano, SpanPtr->GetStartTime());
					Sb.add_fixed64(pbf::Span::required_fixed64_end_time_unix_nano, SpanPtr->GetEndTime());

					AppendAttributesToBuilder(Sb, pbf::Span::repeated_KeyValue_attributes, SpanPtr->GetAttributes());

					for (const otel::Event* Event = SpanPtr->GetEvents(); Event != nullptr; Event = Event->NextEvent)
					{
						protozero::pbf_builder<pbf::Span_Event> EventBuilder{Sb, pbf::Span::repeated_Event_events};
						EventBuilder.add_fixed64(pbf::Span_Event::fixed64_time_unix_nano, Event->Timestamp);
						EventBuilder.add_string(pbf::Span_Event::string_name, Event->Name);

						AppendAttributesToBuilder(EventBuilder, pbf::Span_Event::repeated_KeyValue_attributes, Event->Attributes);
					}
				}
			}
		}
	}

	return Data;
}

}  // namespace zen

#endif