Skip to content

Commit 1fc0ebc

Browse files
rajkumar-rangarajrajkumar-rangaraj
authored
[otlp] OTLP Exporter Custom serializer - Logs (open-telemetry#5941)
1 parent 09b654e commit 1fc0ebc

File tree

6 files changed

+616
-54
lines changed

6 files changed

+616
-54
lines changed

‎src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/Serializer/ProtobufOtlpLogFieldNumberConstants.cs‎

Lines changed: 88 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,88 @@
1+
// Copyright The OpenTelemetry Authors
2+
// SPDX-License-Identifier: Apache-2.0
3+
4+
namespace OpenTelemetry.Exporter.OpenTelemetryProtocol.Implementation.Serializer;
5+
6+
internal static class ProtobufOtlpLogFieldNumberConstants
7+
{
8+
// Resource Logs
9+
#pragma warning disable SA1310 // Field names should not contain underscore
10+
internal const int ResourceLogs_Resource = 1;
11+
internal const int ResourceLogs_Scope_Logs = 2;
12+
internal const int ResourceLogs_Schema_Url = 3;
13+
14+
// Resource
15+
internal const int Resource_Attributes = 1;
16+
17+
// ScopeLogs
18+
internal const int ScopeLogs_Scope = 1;
19+
internal const int ScopeLogs_Log_Records = 2;
20+
internal const int ScopeLogs_Schema_Url = 3;
21+
22+
// LogRecord
23+
internal const int LogRecord_Time_Unix_Nano = 1;
24+
internal const int LogRecord_Observed_Time_Unix_Nano = 11;
25+
internal const int LogRecord_Severity_Number = 2;
26+
internal const int LogRecord_Severity_Text = 3;
27+
internal const int LogRecord_Body = 5;
28+
internal const int LogRecord_Attributes = 6;
29+
internal const int LogRecord_Dropped_Attributes_Count = 7;
30+
internal const int LogRecord_Flags = 8;
31+
internal const int LogRecord_Trace_Id = 9;
32+
internal const int LogRecord_Span_Id = 10;
33+
34+
// SeverityNumber
35+
internal const int Severity_Number_Unspecified = 0;
36+
internal const int Severity_Number_Trace = 1;
37+
internal const int Severity_Number_Trace2 = 2;
38+
internal const int Severity_Number_Trace3 = 3;
39+
internal const int Severity_Number_Trace4 = 4;
40+
internal const int Severity_Number_Debug = 5;
41+
internal const int Severity_Number_Debug2 = 6;
42+
internal const int Severity_Number_Debug3 = 7;
43+
internal const int Severity_Number_Debug4 = 8;
44+
internal const int Severity_Number_Info = 9;
45+
internal const int Severity_Number_Info2 = 10;
46+
internal const int Severity_Number_Info3 = 11;
47+
internal const int Severity_Number_Info4 = 12;
48+
internal const int Severity_Number_Warn = 13;
49+
internal const int Severity_Number_Warn2 = 14;
50+
internal const int Severity_Number_Warn3 = 15;
51+
internal const int Severity_Number_Warn4 = 16;
52+
internal const int Severity_Number_Error = 17;
53+
internal const int Severity_Number_Error2 = 18;
54+
internal const int Severity_Number_Error3 = 19;
55+
internal const int Severity_Number_Error4 = 20;
56+
internal const int Severity_Number_Fatal = 21;
57+
internal const int Severity_Number_Fatal2 = 22;
58+
internal const int Severity_Number_Fatal3 = 23;
59+
internal const int Severity_Number_Fatal4 = 24;
60+
61+
// LogRecordFlags
62+
63+
internal const int LogRecord_Flags_Do_Not_Use = 0;
64+
internal const int LogRecord_Flags_Trace_Flags_Mask = 0x000000FF;
65+
66+
// InstrumentationScope
67+
internal const int InstrumentationScope_Name = 1;
68+
internal const int InstrumentationScope_Version = 2;
69+
internal const int InstrumentationScope_Attributes = 3;
70+
internal const int InstrumentationScope_Dropped_Attributes_Count = 4;
71+
72+
// KeyValue
73+
internal const int KeyValue_Key = 1;
74+
internal const int KeyValue_Value = 2;
75+
76+
// AnyValue
77+
internal const int AnyValue_String_Value = 1;
78+
internal const int AnyValue_Bool_Value = 2;
79+
internal const int AnyValue_Int_Value = 3;
80+
internal const int AnyValue_Double_Value = 4;
81+
internal const int AnyValue_Array_Value = 5;
82+
internal const int AnyValue_Kvlist_Value = 6;
83+
internal const int AnyValue_Bytes_Value = 7;
84+
85+
internal const int ArrayValue_Value = 1;
86+
#pragma warning restore SA1310 // Field names should not contain underscore
87+
}
88+

‎src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/Serializer/ProtobufOtlpLogSerializer.cs‎

Lines changed: 282 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,282 @@
1+
// Copyright The OpenTelemetry Authors
2+
// SPDX-License-Identifier: Apache-2.0
3+
4+
using OpenTelemetry.Internal;
5+
using OpenTelemetry.Logs;
6+
using OpenTelemetry.Trace;
7+
8+
namespace OpenTelemetry.Exporter.OpenTelemetryProtocol.Implementation.Serializer;
9+
10+
internal static class ProtobufOtlpLogSerializer
11+
{
12+
private const int ReserveSizeForLength = 4;
13+
private const int TraceIdSize = 16;
14+
private const int SpanIdSize = 8;
15+
16+
private static readonly Stack<List<LogRecord>> LogsListPool = [];
17+
private static readonly Dictionary<string, List<LogRecord>> ScopeLogsList = [];
18+
19+
internal static int WriteLogsData(byte[] buffer, int writePosition, SdkLimitOptions sdkLimitOptions, ExperimentalOptions experimentalOptions, Resources.Resource? resource, in Batch<LogRecord> logRecordBatch)
20+
{
21+
foreach (var logRecord in logRecordBatch)
22+
{
23+
var scopeName = logRecord.Logger.Name;
24+
if (!ScopeLogsList.TryGetValue(scopeName, out var logRecords))
25+
{
26+
logRecords = LogsListPool.Count > 0 ? LogsListPool.Pop() : [];
27+
ScopeLogsList[scopeName] = logRecords;
28+
}
29+
30+
logRecords.Add(logRecord);
31+
}
32+
33+
writePosition = WriteResourceLogs(buffer, writePosition, sdkLimitOptions, experimentalOptions, resource, ScopeLogsList);
34+
ReturnLogRecordListToPool();
35+
36+
return writePosition;
37+
}
38+
39+
internal static void ReturnLogRecordListToPool()
40+
{
41+
if (ScopeLogsList.Count != 0)
42+
{
43+
foreach (var entry in ScopeLogsList)
44+
{
45+
entry.Value.Clear();
46+
LogsListPool.Push(entry.Value);
47+
}
48+
49+
ScopeLogsList.Clear();
50+
}
51+
}
52+
53+
internal static int WriteResourceLogs(byte[] buffer, int writePosition, SdkLimitOptions sdkLimitOptions, ExperimentalOptions experimentalOptions, Resources.Resource? resource, Dictionary<string, List<LogRecord>> scopeLogs)
54+
{
55+
writePosition = ProtobufOtlpResourceSerializer.WriteResource(buffer, writePosition, resource);
56+
writePosition = WriteScopeLogs(buffer, writePosition, sdkLimitOptions, experimentalOptions, scopeLogs);
57+
return writePosition;
58+
}
59+
60+
internal static int WriteScopeLogs(byte[] buffer, int writePosition, SdkLimitOptions sdkLimitOptions, ExperimentalOptions experimentalOptions, Dictionary<string, List<LogRecord>> scopeLogs)
61+
{
62+
if (scopeLogs != null)
63+
{
64+
foreach (KeyValuePair<string, List<LogRecord>> entry in scopeLogs)
65+
{
66+
writePosition = ProtobufSerializer.WriteTag(buffer, writePosition, ProtobufOtlpLogFieldNumberConstants.ResourceLogs_Scope_Logs, ProtobufWireType.LEN);
67+
int resourceLogsScopeLogsLengthPosition = writePosition;
68+
writePosition += ReserveSizeForLength;
69+
70+
writePosition = WriteScopeLog(buffer, writePosition, sdkLimitOptions, experimentalOptions, entry.Value[0].Logger.Name, entry.Value);
71+
ProtobufSerializer.WriteReservedLength(buffer, resourceLogsScopeLogsLengthPosition, writePosition - (resourceLogsScopeLogsLengthPosition + ReserveSizeForLength));
72+
}
73+
}
74+
75+
return writePosition;
76+
}
77+
78+
internal static int WriteScopeLog(byte[] buffer, int writePosition, SdkLimitOptions sdkLimitOptions, ExperimentalOptions experimentalOptions, string loggerName, List<LogRecord> logRecords)
79+
{
80+
var value = loggerName.AsSpan();
81+
var numberOfUtf8CharsInString = ProtobufSerializer.GetNumberOfUtf8CharsInString(value);
82+
var serializedLengthSize = ProtobufSerializer.ComputeVarInt64Size((ulong)numberOfUtf8CharsInString);
83+
84+
// numberOfUtf8CharsInString + tagSize + length field size.
85+
writePosition = ProtobufSerializer.WriteTagAndLength(buffer, writePosition, numberOfUtf8CharsInString + 1 + serializedLengthSize, ProtobufOtlpLogFieldNumberConstants.ScopeLogs_Scope, ProtobufWireType.LEN);
86+
writePosition = ProtobufSerializer.WriteStringWithTag(buffer, writePosition, ProtobufOtlpLogFieldNumberConstants.InstrumentationScope_Name, numberOfUtf8CharsInString, value);
87+
88+
for (int i = 0; i < logRecords.Count; i++)
89+
{
90+
writePosition = WriteLogRecord(buffer, writePosition, sdkLimitOptions, experimentalOptions, logRecords[i]);
91+
}
92+
93+
return writePosition;
94+
}
95+
96+
internal static int WriteLogRecord(byte[] buffer, int writePosition, SdkLimitOptions sdkLimitOptions, ExperimentalOptions experimentalOptions, LogRecord logRecord)
97+
{
98+
var attributeValueLengthLimit = sdkLimitOptions.LogRecordAttributeValueLengthLimit;
99+
var attributeCountLimit = sdkLimitOptions.LogRecordAttributeCountLimit ?? int.MaxValue;
100+
101+
ProtobufOtlpTagWriter.OtlpTagWriterState otlpTagWriterState = new ProtobufOtlpTagWriter.OtlpTagWriterState
102+
{
103+
Buffer = buffer,
104+
WritePosition = writePosition,
105+
TagCount = 0,
106+
DroppedTagCount = 0,
107+
};
108+
109+
otlpTagWriterState.WritePosition = ProtobufSerializer.WriteTag(otlpTagWriterState.Buffer, otlpTagWriterState.WritePosition, ProtobufOtlpLogFieldNumberConstants.ScopeLogs_Log_Records, ProtobufWireType.LEN);
110+
int logRecordLengthPosition = otlpTagWriterState.WritePosition;
111+
otlpTagWriterState.WritePosition += ReserveSizeForLength;
112+
113+
var timestamp = (ulong)logRecord.Timestamp.ToUnixTimeNanoseconds();
114+
otlpTagWriterState.WritePosition = ProtobufSerializer.WriteFixed64WithTag(otlpTagWriterState.Buffer, otlpTagWriterState.WritePosition, ProtobufOtlpLogFieldNumberConstants.LogRecord_Time_Unix_Nano, timestamp);
115+
otlpTagWriterState.WritePosition = ProtobufSerializer.WriteFixed64WithTag(otlpTagWriterState.Buffer, otlpTagWriterState.WritePosition, ProtobufOtlpLogFieldNumberConstants.LogRecord_Observed_Time_Unix_Nano, timestamp);
116+
117+
otlpTagWriterState.WritePosition = ProtobufSerializer.WriteEnumWithTag(otlpTagWriterState.Buffer, otlpTagWriterState.WritePosition, ProtobufOtlpLogFieldNumberConstants.LogRecord_Severity_Number, logRecord.Severity.HasValue ? (int)logRecord.Severity : 0);
118+
119+
if (!string.IsNullOrWhiteSpace(logRecord.SeverityText))
120+
{
121+
otlpTagWriterState.WritePosition = ProtobufSerializer.WriteStringWithTag(otlpTagWriterState.Buffer, otlpTagWriterState.WritePosition, ProtobufOtlpLogFieldNumberConstants.LogRecord_Severity_Text, logRecord.SeverityText!);
122+
}
123+
else if (logRecord.Severity.HasValue)
124+
{
125+
otlpTagWriterState.WritePosition = ProtobufSerializer.WriteStringWithTag(otlpTagWriterState.Buffer, otlpTagWriterState.WritePosition, ProtobufOtlpLogFieldNumberConstants.LogRecord_Severity_Text, logRecord.Severity.Value.ToShortName());
126+
}
127+
128+
if (experimentalOptions.EmitLogEventAttributes)
129+
{
130+
if (logRecord.EventId.Id != default)
131+
{
132+
otlpTagWriterState = AddLogAttribute(ref otlpTagWriterState, ExperimentalOptions.LogRecordEventIdAttribute, logRecord.EventId.Id, attributeCountLimit, attributeValueLengthLimit);
133+
}
134+
135+
if (!string.IsNullOrEmpty(logRecord.EventId.Name))
136+
{
137+
otlpTagWriterState = AddLogAttribute(ref otlpTagWriterState, ExperimentalOptions.LogRecordEventNameAttribute, logRecord.EventId.Name!, attributeCountLimit, attributeValueLengthLimit);
138+
}
139+
}
140+
141+
if (logRecord.Exception != null)
142+
{
143+
otlpTagWriterState = AddLogAttribute(ref otlpTagWriterState, SemanticConventions.AttributeExceptionType, logRecord.Exception.GetType().Name, attributeCountLimit, attributeValueLengthLimit);
144+
otlpTagWriterState = AddLogAttribute(ref otlpTagWriterState, SemanticConventions.AttributeExceptionMessage, logRecord.Exception.Message, attributeCountLimit, attributeValueLengthLimit);
145+
otlpTagWriterState = AddLogAttribute(ref otlpTagWriterState, SemanticConventions.AttributeExceptionStacktrace, logRecord.Exception.ToInvariantString(), attributeCountLimit, attributeValueLengthLimit);
146+
}
147+
148+
bool bodyPopulatedFromFormattedMessage = false;
149+
bool isLogRecordBodySet = false;
150+
151+
if (logRecord.FormattedMessage != null)
152+
{
153+
otlpTagWriterState.WritePosition = WriteLogRecordBody(otlpTagWriterState.Buffer, otlpTagWriterState.WritePosition, logRecord.FormattedMessage.AsSpan());
154+
bodyPopulatedFromFormattedMessage = true;
155+
isLogRecordBodySet = true;
156+
}
157+
158+
if (logRecord.Attributes != null)
159+
{
160+
foreach (var attribute in logRecord.Attributes)
161+
{
162+
// Special casing {OriginalFormat}
163+
// See https://github.com/open-telemetry/opentelemetry-dotnet/pull/3182
164+
// for explanation.
165+
if (attribute.Key.Equals("{OriginalFormat}") && !bodyPopulatedFromFormattedMessage)
166+
{
167+
otlpTagWriterState.WritePosition = WriteLogRecordBody(otlpTagWriterState.Buffer, otlpTagWriterState.WritePosition, (attribute.Value as string).AsSpan());
168+
isLogRecordBodySet = true;
169+
}
170+
else
171+
{
172+
otlpTagWriterState = AddLogAttribute(ref otlpTagWriterState, attribute, attributeCountLimit, attributeValueLengthLimit);
173+
}
174+
}
175+
176+
// Supports setting Body directly on LogRecord for the Logs Bridge API.
177+
if (!isLogRecordBodySet && logRecord.Body != null)
178+
{
179+
// If {OriginalFormat} is not present in the attributes,
180+
// use logRecord.Body if it is set.
181+
otlpTagWriterState.WritePosition = WriteLogRecordBody(otlpTagWriterState.Buffer, otlpTagWriterState.WritePosition, logRecord.Body.AsSpan());
182+
}
183+
}
184+
185+
if (logRecord.TraceId != default && logRecord.SpanId != default)
186+
{
187+
otlpTagWriterState.WritePosition = ProtobufSerializer.WriteTagAndLength(otlpTagWriterState.Buffer, otlpTagWriterState.WritePosition, TraceIdSize, ProtobufOtlpLogFieldNumberConstants.LogRecord_Trace_Id, ProtobufWireType.LEN);
188+
otlpTagWriterState.WritePosition = ProtobufOtlpTraceSerializer.WriteTraceId(otlpTagWriterState.Buffer, otlpTagWriterState.WritePosition, logRecord.TraceId);
189+
190+
otlpTagWriterState.WritePosition = ProtobufSerializer.WriteTagAndLength(otlpTagWriterState.Buffer, otlpTagWriterState.WritePosition, SpanIdSize, ProtobufOtlpLogFieldNumberConstants.LogRecord_Span_Id, ProtobufWireType.LEN);
191+
otlpTagWriterState.WritePosition = ProtobufOtlpTraceSerializer.WriteSpanId(otlpTagWriterState.Buffer, otlpTagWriterState.WritePosition, logRecord.SpanId);
192+
193+
otlpTagWriterState.WritePosition = ProtobufSerializer.WriteFixed32WithTag(otlpTagWriterState.Buffer, otlpTagWriterState.WritePosition, ProtobufOtlpLogFieldNumberConstants.LogRecord_Flags, (uint)logRecord.TraceFlags);
194+
}
195+
196+
/*
197+
* TODO: Handle scopes, otlpTagWriterState needs to be passed as ref.
198+
logRecord.ForEachScope(ProcessScope, otlpTagWriterState);
199+
200+
void ProcessScope(LogRecordScope scope, ProtobufOtlpTagWriter.OtlpTagWriterState otlpTagWriterState)
201+
{
202+
foreach (var scopeItem in scope)
203+
{
204+
if (scopeItem.Key.Equals("{OriginalFormat}") || string.IsNullOrEmpty(scopeItem.Key))
205+
{
206+
// Ignore if the scope key is empty.
207+
// Ignore if the scope key is {OriginalFormat}
208+
// Attributes should not contain duplicates,
209+
// and it is expensive to de-dup, so this
210+
// exporter is going to pass the scope items as is.
211+
// {OriginalFormat} is going to be the key
212+
// if one uses formatted string for scopes
213+
// and if there are nested scopes, this is
214+
// guaranteed to create duplicate keys.
215+
// Similar for empty keys, which is what the
216+
// key is going to be if user simply
217+
// passes a string as scope.
218+
// To summarize this exporter only allows
219+
// IReadOnlyList<KeyValuePair<string, object?>>
220+
// or IEnumerable<KeyValuePair<string, object?>>.
221+
// and expect users to provide unique keys.
222+
// Note: It is possible that we allow users
223+
// to override this exporter feature. So not blocking
224+
// empty/{OriginalFormat} in the SDK itself.
225+
}
226+
else
227+
{
228+
otlpTagWriterState = AddLogAttribute(ref otlpTagWriterState, scopeItem, attributeCountLimit, attributeValueLengthLimit);
229+
}
230+
}
231+
}
232+
*/
233+
234+
if (otlpTagWriterState.DroppedTagCount > 0)
235+
{
236+
otlpTagWriterState.WritePosition = ProtobufSerializer.WriteTag(buffer, otlpTagWriterState.WritePosition, ProtobufOtlpLogFieldNumberConstants.LogRecord_Dropped_Attributes_Count, ProtobufWireType.VARINT);
237+
otlpTagWriterState.WritePosition = ProtobufSerializer.WriteVarInt32(buffer, otlpTagWriterState.WritePosition, (uint)otlpTagWriterState.DroppedTagCount);
238+
}
239+
240+
ProtobufSerializer.WriteReservedLength(otlpTagWriterState.Buffer, logRecordLengthPosition, otlpTagWriterState.WritePosition - (logRecordLengthPosition + ReserveSizeForLength));
241+
242+
return otlpTagWriterState.WritePosition;
243+
}
244+
245+
private static int WriteLogRecordBody(byte[] buffer, int writePosition, ReadOnlySpan<char> value)
246+
{
247+
var numberOfUtf8CharsInString = ProtobufSerializer.GetNumberOfUtf8CharsInString(value);
248+
var serializedLengthSize = ProtobufSerializer.ComputeVarInt64Size((ulong)numberOfUtf8CharsInString);
249+
250+
// length = numberOfUtf8CharsInString + tagSize + length field size.
251+
writePosition = ProtobufSerializer.WriteTagAndLength(buffer, writePosition, numberOfUtf8CharsInString + 1 + serializedLengthSize, ProtobufOtlpLogFieldNumberConstants.LogRecord_Body, ProtobufWireType.LEN);
252+
writePosition = ProtobufSerializer.WriteStringWithTag(buffer, writePosition, ProtobufOtlpTraceFieldNumberConstants.AnyValue_String_Value, numberOfUtf8CharsInString, value);
253+
return writePosition;
254+
}
255+
256+
private static ProtobufOtlpTagWriter.OtlpTagWriterState AddLogAttribute(ref ProtobufOtlpTagWriter.OtlpTagWriterState otlpTagWriterState, KeyValuePair<string, object?> attribute, int maxAttributeCount, int? maxValueLength)
257+
{
258+
return AddLogAttribute(ref otlpTagWriterState, attribute.Key, attribute.Value, maxAttributeCount, maxValueLength);
259+
}
260+
261+
private static ProtobufOtlpTagWriter.OtlpTagWriterState AddLogAttribute(ref ProtobufOtlpTagWriter.OtlpTagWriterState otlpTagWriterState, string key, object? value, int maxAttributeCount, int? maxValueLength)
262+
{
263+
if (otlpTagWriterState.TagCount == maxAttributeCount)
264+
{
265+
otlpTagWriterState.DroppedTagCount++;
266+
}
267+
else
268+
{
269+
otlpTagWriterState.WritePosition = ProtobufSerializer.WriteTag(otlpTagWriterState.Buffer, otlpTagWriterState.WritePosition, ProtobufOtlpLogFieldNumberConstants.LogRecord_Attributes, ProtobufWireType.LEN);
270+
int logAttributesLengthPosition = otlpTagWriterState.WritePosition;
271+
otlpTagWriterState.WritePosition += ReserveSizeForLength;
272+
273+
ProtobufOtlpTagWriter.Instance.TryWriteTag(ref otlpTagWriterState, key, value, maxValueLength);
274+
275+
var logAttributesLength = otlpTagWriterState.WritePosition - (logAttributesLengthPosition + ReserveSizeForLength);
276+
ProtobufSerializer.WriteReservedLength(otlpTagWriterState.Buffer, logAttributesLengthPosition, logAttributesLength);
277+
otlpTagWriterState.TagCount++;
278+
}
279+
280+
return otlpTagWriterState;
281+
}
282+
}

‎src/OpenTelemetry.Exporter.OpenTelemetryProtocol/Implementation/Serializer/ProtobufOtlpTagWriter.cs‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -84,6 +84,8 @@ protected override void OnUnsupportedTagDropped(
8484
internal struct OtlpTagWriterState
8585
{
8686
public byte[] Buffer;
87+
public int DroppedTagCount;
88+
public int TagCount;
8789
public int WritePosition;
8890
}
8991

0 commit comments

Comments
 (0)